master
go 542 lines 17.1 KB
Raw
1 package migrations
2
3 import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "log"
9 "net/url"
10 "os"
11 "os/exec"
12 "path"
13 "runtime"
14 "strings"
15 "sync"
16
17 config "github.com/ipfs/kubo/config"
18 )
19
20 const (
21 // Migrations subdirectory in distribution. Empty for root (no subdir).
22 distMigsRoot = ""
23 distFSRM = "fs-repo-migrations"
24 )
25
26 // RunMigration finds, downloads, and runs the individual migrations needed to
27 // migrate the repo from its current version to the target version.
28 //
29 // Deprecated: This function downloads migration binaries from the internet and will be removed
30 // in a future version. Use RunHybridMigrations for modern migrations with embedded support,
31 // or RunEmbeddedMigrations for repo versions ≥16.
32 func RunMigration(ctx context.Context, fetcher Fetcher, targetVer int, ipfsDir string, allowDowngrade bool) error {
33 ipfsDir, err := CheckIpfsDir(ipfsDir)
34 if err != nil {
35 return err
36 }
37 fromVer, err := RepoVersion(ipfsDir)
38 if err != nil {
39 return fmt.Errorf("could not get repo version: %w", err)
40 }
41 if fromVer == targetVer {
42 // repo already at target version number
43 return nil
44 }
45 if fromVer > targetVer && !allowDowngrade {
46 return fmt.Errorf("downgrade not allowed from %d to %d", fromVer, targetVer)
47 }
48
49 logger := log.New(os.Stdout, "", 0)
50
51 logger.Print("Looking for suitable migration binaries.")
52
53 migrations, binPaths, err := findMigrations(ctx, fromVer, targetVer)
54 if err != nil {
55 return err
56 }
57
58 // Download migrations that were not found
59 if len(binPaths) < len(migrations) {
60 missing := make([]string, 0, len(migrations)-len(binPaths))
61 for _, mig := range migrations {
62 if _, ok := binPaths[mig]; !ok {
63 missing = append(missing, mig)
64 }
65 }
66
67 logger.Println("Need", len(missing), "migrations, downloading.")
68
69 tmpDir, err := os.MkdirTemp("", "migrations")
70 if err != nil {
71 return err
72 }
73 defer os.RemoveAll(tmpDir)
74
75 fetched, err := fetchMigrations(ctx, fetcher, missing, tmpDir, logger)
76 if err != nil {
77 logger.Print("Failed to download migrations.")
78 return err
79 }
80
81 for i := range missing {
82 binPaths[missing[i]] = fetched[i]
83 }
84 }
85
86 var revert bool
87 if fromVer > targetVer {
88 revert = true
89 }
90 for _, migration := range migrations {
91 logger.Println("Running migration", migration, "...")
92 err = runMigration(ctx, binPaths[migration], ipfsDir, revert, logger)
93 if err != nil {
94 return fmt.Errorf("migration %s failed: %w", migration, err)
95 }
96 }
97 logger.Printf("Success: fs-repo migrated to version %d.\n", targetVer)
98
99 return nil
100 }
101
102 func NeedMigration(target int) (bool, error) {
103 vnum, err := RepoVersion("")
104 if err != nil {
105 return false, fmt.Errorf("could not get repo version: %w", err)
106 }
107
108 return vnum != target, nil
109 }
110
111 func ExeName(name string) string {
112 if runtime.GOOS == "windows" {
113 return name + ".exe"
114 }
115 return name
116 }
117
118 // ReadMigrationConfig reads the Migration section of the IPFS config, avoiding
119 // reading anything other than the Migration section. That way, we're free to
120 // make arbitrary changes to all _other_ sections in migrations.
121 //
122 // Deprecated: This function is used by legacy migration downloads and will be removed
123 // in a future version. Use RunHybridMigrations or RunEmbeddedMigrations instead.
124 func ReadMigrationConfig(repoRoot string, userConfigFile string) (*config.Migration, error) {
125 var cfg struct {
126 Migration config.Migration
127 }
128
129 cfgPath, err := config.Filename(repoRoot, userConfigFile)
130 if err != nil {
131 return nil, err
132 }
133
134 cfgFile, err := os.Open(cfgPath)
135 if err != nil {
136 return nil, err
137 }
138 defer cfgFile.Close()
139
140 err = json.NewDecoder(cfgFile).Decode(&cfg)
141 if err != nil {
142 return nil, err
143 }
144
145 switch cfg.Migration.Keep {
146 case "":
147 cfg.Migration.Keep = config.DefaultMigrationKeep
148 case "discard", "cache", "keep":
149 default:
150 return nil, errors.New("unknown config value, Migrations.Keep must be 'cache', 'pin', or 'discard'")
151 }
152
153 if len(cfg.Migration.DownloadSources) == 0 {
154 cfg.Migration.DownloadSources = config.DefaultMigrationDownloadSources
155 }
156
157 return &cfg.Migration, nil
158 }
159
160 // GetMigrationFetcher creates one or more fetchers from downloadSources.
161 // Multiple fetchers are wrapped in a MultiFetcher that rotates to the next
162 // gateway when one errors and quarantines failed gateways for the session.
163 //
164 // Deprecated: This function is used by legacy migration downloads and will be removed
165 // in a future version. Use RunHybridMigrations or RunEmbeddedMigrations instead.
166 func GetMigrationFetcher(downloadSources []string, distPath string, newIpfsFetcher func(string) Fetcher) (Fetcher, error) {
167 const httpUserAgent = "kubo/migration"
168
169 var fetchers []Fetcher
170 for _, src := range downloadSources {
171 src := strings.TrimSpace(src)
172 switch src {
173 case "HTTPS", "https", "HTTP", "http":
174 // Expand the alias into the full ordered list of trustless
175 // community-provided gateways so migration survives a
176 // single-gateway outage.
177 for _, gw := range defaultMigrationGateways {
178 fetchers = append(fetchers, NewHttpFetcher(distPath, gw, httpUserAgent, 0))
179 }
180 case "IPFS", "ipfs":
181 return nil, errors.New("IPFS downloads are not supported for legacy migrations (repo versions <16). Please use only HTTPS in Migration.DownloadSources")
182 case "":
183 // Ignore empty string
184 default:
185 u, err := url.Parse(src)
186 if err != nil {
187 return nil, fmt.Errorf("bad gateway address: %w", err)
188 }
189 switch u.Scheme {
190 case "":
191 u.Scheme = "https"
192 case "https", "http":
193 default:
194 return nil, errors.New("bad gateway address: url scheme must be http or https")
195 }
196 fetchers = append(fetchers, NewHttpFetcher(distPath, u.String(), httpUserAgent, 0))
197 }
198 }
199
200 switch len(fetchers) {
201 case 0:
202 return nil, errors.New("no sources specified")
203 case 1:
204 return fetchers[0], nil
205 }
206
207 // Wrap fetchers in a MultiFetcher to try them in order
208 return NewMultiFetcher(fetchers...), nil
209 }
210
211 func migrationName(from, to int) string {
212 return fmt.Sprintf("fs-repo-%d-to-%d", from, to)
213 }
214
215 // findMigrations returns a list of migrations, ordered from first to last
216 // migration to apply, and a map of locations of migration binaries of any
217 // migrations that were found.
218 //
219 // Deprecated: This function is used by legacy migration downloads and will be removed
220 // in a future version.
221 func findMigrations(ctx context.Context, from, to int) ([]string, map[string]string, error) {
222 step := 1
223 count := to - from
224 if from > to {
225 step = -1
226 count = from - to
227 }
228
229 migrations := make([]string, 0, count)
230 binPaths := make(map[string]string, count)
231
232 for cur := from; cur != to; cur += step {
233 if ctx.Err() != nil {
234 return nil, nil, ctx.Err()
235 }
236 var migName string
237 if step == -1 {
238 migName = migrationName(cur+step, cur)
239 } else {
240 migName = migrationName(cur, cur+step)
241 }
242 migrations = append(migrations, migName)
243 bin, err := exec.LookPath(migName)
244 if err != nil {
245 continue
246 }
247 binPaths[migName] = bin
248 }
249 return migrations, binPaths, nil
250 }
251
252 func runMigration(ctx context.Context, binPath, ipfsDir string, revert bool, logger *log.Logger) error {
253 pathArg := fmt.Sprintf("-path=%s", ipfsDir)
254 var cmd *exec.Cmd
255 if revert {
256 logger.Println(" => Running:", binPath, pathArg, "-verbose=true -revert")
257 cmd = exec.CommandContext(ctx, binPath, pathArg, "-verbose=true", "-revert")
258 } else {
259 logger.Println(" => Running:", binPath, pathArg, "-verbose=true")
260 cmd = exec.CommandContext(ctx, binPath, pathArg, "-verbose=true")
261 }
262 cmd.Stdout = os.Stdout
263 cmd.Stderr = os.Stderr
264 return cmd.Run()
265 }
266
267 // fetchMigrations downloads the requested migrations, and returns a slice with
268 // the paths of each binary, in the same order specified by needed.
269 //
270 // Deprecated: This function downloads migration binaries from the internet and will be removed
271 // in a future version. Use RunHybridMigrations or RunEmbeddedMigrations instead.
272 func fetchMigrations(ctx context.Context, fetcher Fetcher, needed []string, destDir string, logger *log.Logger) ([]string, error) {
273 osv, err := osWithVariant()
274 if err != nil {
275 return nil, err
276 }
277 if osv == "linux-musl" {
278 return nil, fmt.Errorf("linux-musl not supported, you must build the binary from source for your platform")
279 }
280
281 var wg sync.WaitGroup
282 wg.Add(len(needed))
283 bins := make([]string, len(needed))
284 // Download and unpack all requested migrations concurrently.
285 for i, name := range needed {
286 logger.Printf("Downloading migration: %s...", name)
287 go func(i int, name string) {
288 defer wg.Done()
289 dist := path.Join(distMigsRoot, name)
290 ver, err := LatestDistVersion(ctx, fetcher, dist, false)
291 if err != nil {
292 logger.Printf("could not get latest version of migration %s: %s", name, err)
293 return
294 }
295 loc, err := FetchBinary(ctx, fetcher, dist, ver, name, destDir)
296 if err != nil {
297 logger.Printf("could not download %s: %s", name, err)
298 return
299 }
300 logger.Printf("Downloaded and unpacked migration: %s (%s)", loc, ver)
301 bins[i] = loc
302 }(i, name)
303 }
304 wg.Wait()
305
306 var fails []string
307 for i := range bins {
308 if bins[i] == "" {
309 fails = append(fails, needed[i])
310 }
311 }
312 if len(fails) != 0 {
313 err = fmt.Errorf("failed to download migrations: %s", strings.Join(fails, " "))
314 if ctx.Err() != nil {
315 err = fmt.Errorf("%s, %w", ctx.Err(), err)
316 }
317 return nil, err
318 }
319
320 return bins, nil
321 }
322
323 // RunHybridMigrations intelligently runs migrations using external tools for legacy versions
324 // and embedded migrations for modern versions. This handles the transition from external
325 // fs-repo-migrations binaries (for repo versions <16) to embedded migrations (for repo versions ≥16).
326 //
327 // The function automatically:
328 // 1. Uses external migrations to get from current version to v16 (if needed)
329 // 2. Uses embedded migrations for v16+ steps
330 // 3. Handles pure external, pure embedded, or mixed migration scenarios
331 //
332 // Legacy external migrations (repo versions <16) only support HTTPS downloads.
333 //
334 // Parameters:
335 // - ctx: Context for cancellation and timeouts
336 // - targetVer: Target repository version to migrate to
337 // - ipfsDir: Path to the IPFS repository directory
338 // - allowDowngrade: Whether to allow downgrade migrations
339 //
340 // Returns error if migration fails at any step.
341 func RunHybridMigrations(ctx context.Context, targetVer int, ipfsDir string, allowDowngrade bool) error {
342 const embeddedMigrationsMinVersion = 16
343
344 // Get current repo version
345 currentVer, err := RepoVersion(ipfsDir)
346 if err != nil {
347 return fmt.Errorf("could not get current repo version: %w", err)
348 }
349
350 var logger = log.New(os.Stdout, "", 0)
351
352 // Check if migration is needed
353 if currentVer == targetVer {
354 logger.Printf("Repository is already at version %d", targetVer)
355 return nil
356 }
357
358 // Validate downgrade request
359 if targetVer < currentVer && !allowDowngrade {
360 return fmt.Errorf("downgrade from version %d to %d requires allowDowngrade=true", currentVer, targetVer)
361 }
362
363 // Determine migration strategy based on version ranges
364 needsExternal := currentVer < embeddedMigrationsMinVersion
365 needsEmbedded := targetVer >= embeddedMigrationsMinVersion
366
367 // Case 1: Pure embedded migration (both current and target ≥ 16)
368 if !needsExternal && needsEmbedded {
369 return RunEmbeddedMigrations(ctx, targetVer, ipfsDir, allowDowngrade)
370 }
371
372 // For cases requiring external migrations, we check if migration binaries
373 // are available in PATH before attempting network downloads
374
375 // Case 2: Pure external migration (target < 16)
376 if needsExternal && !needsEmbedded {
377
378 // Check for migration binaries in PATH first (for testing/local development)
379 migrations, binPaths, err := findMigrations(ctx, currentVer, targetVer)
380 if err != nil {
381 return fmt.Errorf("could not determine migration paths: %w", err)
382 }
383
384 foundAll := true
385 for _, migName := range migrations {
386 if _, exists := binPaths[migName]; !exists {
387 foundAll = false
388 break
389 }
390 }
391
392 if foundAll {
393 return runMigrationsFromPath(ctx, migrations, binPaths, ipfsDir, logger, false)
394 }
395
396 // Fall back to network download (original behavior)
397 migrationCfg, err := ReadMigrationConfig(ipfsDir, "")
398 if err != nil {
399 return fmt.Errorf("could not read migration config: %w", err)
400 }
401
402 // Use existing RunMigration which handles network downloads properly (HTTPS only for legacy migrations)
403 fetcher, err := GetMigrationFetcher(migrationCfg.DownloadSources, GetDistPathEnv(CurrentIpfsDist), nil)
404 if err != nil {
405 return fmt.Errorf("failed to get migration fetcher: %w", err)
406 }
407 defer fetcher.Close()
408 return RunMigration(ctx, fetcher, targetVer, ipfsDir, allowDowngrade)
409 }
410
411 // Case 3: Hybrid migration (current < 16, target ≥ 16)
412 if needsExternal && needsEmbedded {
413 logger.Printf("Starting hybrid migration from version %d to %d", currentVer, targetVer)
414 logger.Print("Using hybrid migration strategy: external to v16, then embedded")
415
416 // Phase 1: Use external migrations to get to v16
417 logger.Printf("Phase 1: External migration from v%d to v%d", currentVer, embeddedMigrationsMinVersion)
418
419 // Check for external migration binaries in PATH first
420 migrations, binPaths, err := findMigrations(ctx, currentVer, embeddedMigrationsMinVersion)
421 if err != nil {
422 return fmt.Errorf("could not determine external migration paths: %w", err)
423 }
424
425 foundAll := true
426 for _, migName := range migrations {
427 if _, exists := binPaths[migName]; !exists {
428 foundAll = false
429 break
430 }
431 }
432
433 if foundAll {
434 if err = runMigrationsFromPath(ctx, migrations, binPaths, ipfsDir, logger, false); err != nil {
435 return fmt.Errorf("external migration phase failed: %w", err)
436 }
437 } else {
438 migrationCfg, err := ReadMigrationConfig(ipfsDir, "")
439 if err != nil {
440 return fmt.Errorf("could not read migration config: %w", err)
441 }
442
443 // Legacy migrations only support HTTPS downloads
444 fetcher, err := GetMigrationFetcher(migrationCfg.DownloadSources, GetDistPathEnv(CurrentIpfsDist), nil)
445 if err != nil {
446 return fmt.Errorf("failed to get migration fetcher: %w", err)
447 }
448 defer fetcher.Close()
449
450 if err = RunMigration(ctx, fetcher, embeddedMigrationsMinVersion, ipfsDir, allowDowngrade); err != nil {
451 return fmt.Errorf("external migration phase failed: %w", err)
452 }
453 }
454
455 // Phase 2: Use embedded migrations for v16+
456 logger.Printf("Phase 2: Embedded migration from v%d to v%d", embeddedMigrationsMinVersion, targetVer)
457 err = RunEmbeddedMigrations(ctx, targetVer, ipfsDir, allowDowngrade)
458 if err != nil {
459 return fmt.Errorf("embedded migration phase failed: %w", err)
460 }
461
462 logger.Printf("Hybrid migration completed successfully: v%d → v%d", currentVer, targetVer)
463 return nil
464 }
465
466 // Case 4: Reverse hybrid migration (≥16 to <16)
467 // Use embedded migrations for ≥16 steps, then external migrations for <16 steps
468 logger.Printf("Starting reverse hybrid migration from version %d to %d", currentVer, targetVer)
469 logger.Print("Using reverse hybrid migration strategy: embedded to v16, then external")
470
471 // Phase 1: Use embedded migrations from current version down to v16 (if needed)
472 if currentVer > embeddedMigrationsMinVersion {
473 logger.Printf("Phase 1: Embedded downgrade from v%d to v%d", currentVer, embeddedMigrationsMinVersion)
474 err = RunEmbeddedMigrations(ctx, embeddedMigrationsMinVersion, ipfsDir, allowDowngrade)
475 if err != nil {
476 return fmt.Errorf("embedded downgrade phase failed: %w", err)
477 }
478 }
479
480 // Phase 2: Use external migrations from v16 to target (if needed)
481 if embeddedMigrationsMinVersion > targetVer {
482 logger.Printf("Phase 2: External downgrade from v%d to v%d", embeddedMigrationsMinVersion, targetVer)
483
484 // Check for external migration binaries in PATH first
485 migrations, binPaths, err := findMigrations(ctx, embeddedMigrationsMinVersion, targetVer)
486 if err != nil {
487 return fmt.Errorf("could not determine external migration paths: %w", err)
488 }
489
490 foundAll := true
491 for _, migName := range migrations {
492 if _, exists := binPaths[migName]; !exists {
493 foundAll = false
494 break
495 }
496 }
497
498 if foundAll {
499 if err = runMigrationsFromPath(ctx, migrations, binPaths, ipfsDir, logger, true); err != nil {
500 return fmt.Errorf("external downgrade phase failed: %w", err)
501 }
502 } else {
503 migrationCfg, err := ReadMigrationConfig(ipfsDir, "")
504 if err != nil {
505 return fmt.Errorf("could not read migration config: %w", err)
506 }
507
508 // Legacy migrations only support HTTPS downloads
509 fetcher, err := GetMigrationFetcher(migrationCfg.DownloadSources, GetDistPathEnv(CurrentIpfsDist), nil)
510 if err != nil {
511 return fmt.Errorf("failed to get migration fetcher: %w", err)
512 }
513 defer fetcher.Close()
514
515 if err = RunMigration(ctx, fetcher, targetVer, ipfsDir, allowDowngrade); err != nil {
516 return fmt.Errorf("external downgrade phase failed: %w", err)
517 }
518 }
519 }
520
521 logger.Printf("Reverse hybrid migration completed successfully: v%d → v%d", currentVer, targetVer)
522 return nil
523 }
524
525 // runMigrationsFromPath runs migrations using binaries found in PATH
526 func runMigrationsFromPath(ctx context.Context, migrations []string, binPaths map[string]string, ipfsDir string, logger *log.Logger, revert bool) error {
527 for _, migName := range migrations {
528 binPath, exists := binPaths[migName]
529 if !exists {
530 return fmt.Errorf("migration binary %s not found in PATH", migName)
531 }
532
533 logger.Printf("Running migration %s using binary from PATH: %s", migName, binPath)
534
535 // Run the migration binary directly
536 err := runMigration(ctx, binPath, ipfsDir, revert, logger)
537 if err != nil {
538 return fmt.Errorf("migration %s failed: %w", migName, err)
539 }
540 }
541 return nil
542 }