Method GetCompleteIndexSet

internal/epoch/epoch_manager.go:795–830

GetCompleteIndexSet returns the set of blobs forming a complete index set up to the provided epoch number.

(ctx context.Context, maxEpoch int)

// GetCompleteIndexSet returns the set of blobs forming a complete index set up to the provided epoch number.
func (e *Manager) GetCompleteIndexSet(ctx context.Context, maxEpoch int) ([]blob.Metadata, time.Time, error) {
	for {
		cs, err := e.committedState(ctx, 0)
		if err != nil {
			return nil, time.Time{}, err
		}

		if maxEpoch == LatestEpoch {
			maxEpoch = cs.WriteEpoch + 1
		}

		result, err := e.getCompleteIndexSetForCommittedState(ctx, cs, 0, maxEpoch)
		if e.timeFunc().Before(cs.ValidUntil) {
			contentlog.Log4(ctx, e.log,
				"complete index set",
				logparam.Int("maxEpoch", maxEpoch),
				logparam.Int("resultLength", len(result)),
				blobparam.BlobIDList("result", blob.IDsFromMetadata(result)),
				logparam.Time("deletionWatermark", cs.DeletionWatermark),
			)

			return result, cs.DeletionWatermark, err
		}

		// We need to retry if local process took too long (e.g. because the machine went
		// to sleep at the wrong moment) and committed state is no longer valid.
		//
		// One scenario where this matters is cleanup: if determining the set of indexes takes
		// too long, it's possible for a cleanup process to delete some of the uncompacted
		// indexes that are still treated as authoritative according to old committed state.
		//
		// Retrying will re-examine the state of the world and re-do the logic.
		contentlog.Log(ctx, e.log, "GetCompleteIndexSet took too long, retrying to ensure correctness")
		atomic.AddInt32(e.getCompleteIndexSetTooSlow, 1)
	}
}

var errWriteIndexTryAgain = errors.New("try again")