Method runWithPrimaryClient

pkg/ring/kv/multi.go:247–284

Runs supplied fn with current primary client. If primary client changes, fn is restarted. When fn finishes (with or without error), this method returns given error value.

(origCtx context.Context, fn func(newCtx context.Context, primary kvclient) error)

// Runs supplied fn with current primary client. If primary client changes, fn is restarted.
// When fn finishes (with or without error), this method returns given error value.
func (m *MultiClient) runWithPrimaryClient(origCtx context.Context, fn func(newCtx context.Context, primary kvclient) error) error {
	cancelFn := context.CancelFunc(nil)
	cancelFnID := 0

	cleanup := func() {
		if cancelFn != nil {
			cancelFn()
		}
		if cancelFnID > 0 {
			m.unregisterCancelFn(cancelFnID)
		}
	}

	defer cleanup()

	// This only loops if switchover to a new primary backend happens while calling 'fn', which is very rare.
	for {
		cleanup()
		pid, kv := m.getPrimaryClient()

		var cancelCtx context.Context
		cancelCtx, cancelFn = context.WithCancel(origCtx)
		cancelFnID = m.registerCancelFn(pid, cancelFn)

		err := fn(cancelCtx, kv)

		if err == nil {
			return nil
		}

		if cancelCtx.Err() == context.Canceled && origCtx.Err() == nil {
			// our context was cancelled, but outer context is not done yet. retry
			continue
		}

		return err
	}
}

// List is a part of the kv.Client interface.
func (m *MultiClient) List(ctx context.Context, prefix string) ([]string, error) {