Function DoUntilQuorumWithoutSuccessfulContextCancellation

ring/replication_set.go:244–401

DoUntilQuorumWithoutSuccessfulContextCancellation behaves the same as DoUntilQuorum, except it does not cancel the context.Context passed to invocations of f whose results are returned. For example, this is useful in situations where DoUntilQuorumWithoutSuccessfulContextCancellation is used to esta

(ctx context.Context, r ReplicationSet, cfg DoUntilQuorumConfig, f func(context.Context, *InstanceDesc, context.CancelCauseFunc) (T, error), cleanupFunc func(T))

//
// Failing to do this may result in a memory leak.
func DoUntilQuorumWithoutSuccessfulContextCancellation[T any](ctx context.Context, r ReplicationSet, cfg DoUntilQuorumConfig, f func(context.Context, *InstanceDesc, context.CancelCauseFunc) (T, error), cleanupFunc func(T)) ([]T, error) {
	if err := cfg.Validate(); err != nil {
		return nil, err
	}

	if r.ZoneAwarenessEnabled && r.MaxErrors > 0 {
		return nil, fmt.Errorf("invalid ReplicationSet: MaxErrors is non-zero (is %v) and ZoneAwarenessEnabled is true", r.MaxErrors)
	}

	var logger kitlog.Logger = cfg.Logger
	if cfg.Logger == nil {
		logger = kitlog.NewNopLogger()
	}

	if cfg.IncludeReplicaCount {
		ctx = ContextWithAvailableReplicas(ctx, len(r.Instances))
	}

	resultsChan := make(chan instanceResult[T], len(r.Instances))
	resultsRemaining := len(r.Instances)

	defer func() {
		go func() {
			for resultsRemaining > 0 {
				result := <-resultsChan
				resultsRemaining--

				if result.err == nil {
					cleanupFunc(result.result)
				}
			}
		}()
	}()

	var resultTracker replicationSetResultTracker
	var contextTracker replicationSetContextTracker
	if r.MaxUnavailableZones > 0 || r.ZoneAwarenessEnabled {
		resultTracker = newZoneAwareResultTracker(r.Instances, r.MaxUnavailableZones, cfg.ZoneSorter, logger)
		contextTracker = newZoneAwareContextTracker(ctx, r.Instances)
	} else {
		resultTracker = newDefaultResultTracker(r.Instances, r.MaxErrors, logger)
		contextTracker = newDefaultContextTracker(ctx, r.Instances)
	}

	if cfg.MinimizeRequests {
		resultTracker.startMinimumRequests()
	} else {
		resultTracker.startAllRequests()
	}

	for i := range r.Instances {
		instance := &r.Instances[i]
		ctx, ctxCancel := contextTracker.contextFor(instance)

		go func(desc *InstanceDesc) {
			if err := resultTracker.awaitStart(ctx, desc); err != nil {
				// Post to resultsChan so that the deferred cleanup handler above eventually terminates.
				resultsChan <- instanceResult[T]{