Method run

cli/command_snapshot_migrate.go:48–141

(ctx context.Context, destRepo repo.RepositoryWriter)

}

func (c *commandSnapshotMigrate) run(ctx context.Context, destRepo repo.RepositoryWriter) error {
	sourceRepo, err := c.openSourceRepo(ctx)
	if err != nil {
		return errors.Wrap(err, "can't open source repository")
	}

	defer sourceRepo.Close(ctx) //nolint:errcheck

	sources, err := c.getSourcesToMigrate(ctx, sourceRepo)
	if err != nil {
		return errors.Wrap(err, "can't retrieve sources")
	}

	semaphore := make(chan struct{}, c.migrateParallel)

	var (
		wg              sync.WaitGroup
		mu              sync.Mutex
		canceled        bool
		activeUploaders = map[snapshot.SourceInfo]*upload.Uploader{}
	)

	c.svc.getProgress().StartShared()

	c.svc.onTerminate(func() {
		mu.Lock()
		defer mu.Unlock()

		if canceled {
			return
		}

		canceled = true

		for s, u := range activeUploaders {
			log(ctx).Infof("canceling active uploader for %v", s)
			u.Cancel()
		}
	})

	if c.migratePolicies {
		if c.migrateAll {
			err = c.migrateAllPolicies(ctx, sourceRepo, destRepo)
		} else {
			err = c.migratePoliciesForSources(ctx, sourceRepo, destRepo, sources)
		}

		if err != nil {
			return errors.Wrap(err, "unable to migrate policies")
		}
	}

	for _, s := range sources {
		// start a new uploader unless already canceled
		mu.Lock()

		if canceled {
			mu.Unlock()