Method recoverIndexesFromAllPacks

cli/command_index_recover.go:100–185

(ctx context.Context, rep repo.DirectRepositoryWriter, prefixes []blob.ID, processedBlobCount, recoveredContentCount *atomic.Int32)

}

func (c *commandIndexRecover) recoverIndexesFromAllPacks(ctx context.Context, rep repo.DirectRepositoryWriter, prefixes []blob.ID, processedBlobCount, recoveredContentCount *atomic.Int32) error {
	var (
		discoveredBlobCount  atomic.Int32
		discoveringBlobCount atomic.Int32
		tt                   timetrack.Throttle
	)

	// recover indexes from all pack blobs in parallel.
	// this is actually quite fast since we typically need to read only 8KB from each blob.
	eg, ctx := errgroup.WithContext(ctx)

	go func() {
		for _, prefix := range prefixes {
			//nolint:errcheck
			rep.BlobStorage().ListBlobs(ctx, prefix, func(_ blob.Metadata) error {
				discoveringBlobCount.Add(1)
				return nil
			})
		}

		discoveredBlobCount.Store(discoveringBlobCount.Load())
	}()

	est := timetrack.Start()

	blobCh := make(chan blob.Metadata)

	// goroutine to populate blob metadata into a channel.
	eg.Go(func() error {
		defer close(blobCh)

		for _, prefix := range prefixes {
			if err := rep.BlobStorage().ListBlobs(ctx, prefix, func(bm blob.Metadata) error {
				blobCh <- bm
				return nil
			}); err != nil {
				return errors.Wrapf(err, "error listing blobs with prefix %q", prefix)
			}
		}

		return nil
	})

	// N goroutines to recover from incoming blobs.
	for worker := range c.parallel {
		eg.Go(func() error {
			cnt := 0

			for bm := range blobCh {
				finishedBlobs := processedBlobCount.Load()

				log(ctx).Debugf("worker %v got %v", worker, cnt)

				cnt++

				if tt.ShouldOutput(time.Second) {
					if disc := discoveredBlobCount.Load(); disc > 0 {
						e, ok := est.Estimate(float64(finishedBlobs), float64(disc))