Method generalProcessPipeline

redis.go:1138–1210

(
	ctx context.Context, cmds []Cmder, p pipelineProcessor, operationName string,
)

type pipelineProcessor func(context.Context, *pool.Conn, []Cmder) (bool, error)

func (c *baseClient) generalProcessPipeline(
	ctx context.Context, cmds []Cmder, p pipelineProcessor, operationName string,
) error {
	// Only call time.Now() if pipeline operation duration callback is set to avoid overhead
	var operationStart time.Time
	pipelineOpDurationCallback := otel.GetPipelineOperationDurationCallback()
	if pipelineOpDurationCallback != nil {
		operationStart = time.Now()
	}
	var lastConn *pool.Conn
	totalAttempts := 0

	var lastErr error
	for attempt := 0; attempt <= c.opt.MaxRetries; attempt++ {
		totalAttempts++
		if attempt > 0 {
			if err := internal.Sleep(ctx, c.retryBackoff(attempt)); err != nil {
				setCmdsErr(cmds, err)
				if pipelineOpDurationCallback != nil {
					operationDuration := time.Since(operationStart)
					pipelineOpDurationCallback(ctx, operationDuration, operationName, len(cmds), totalAttempts, err, lastConn, c.opt.DB)
				}
				return err
			}
		}

		// Enable retries by default to retry dial errors returned by withConn.
		canRetry := true
		lastErr = c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
			lastConn = cn
			// Process any pending push notifications before executing the pipeline
			if err := c.processPushNotifications(ctx, cn); err != nil {
				internal.Logger.Printf(ctx, "push: error processing pending notifications before processing pipeline: %v", err)
			}
			var err error
			canRetry, err = p(ctx, cn, cmds)
			return err
		})
		// Don't retry if any command in the pipeline explicitly disables retries
		// (e.g., RawWriteToCmd which writes directly to an io.Writer and cannot
		// undo partial writes on retry)
		if lastErr == nil || !canRetry || !shouldRetry(lastErr, true) || cmdsContainNoRetry(cmds) {
			// The error should be set here only when failing to obtain the conn.
			if !isRedisError(lastErr) {
				setCmdsErr(cmds, lastErr)
			}
			if pipelineOpDurationCallback != nil {
				operationDuration := time.Since(operationStart)
				pipelineOpDurationCallback(ctx, operationDuration, operationName, len(cmds), totalAttempts, lastErr, lastConn, c.opt.DB)
			}

			if lastErr != nil {
				if errorCallback := pool.GetMetricErrorCallback(); errorCallback != nil {
					errorType, statusCode, isInternal := classifyCommandError(lastErr)
					errorCallback(ctx, errorType, lastConn, statusCode, isInternal, totalAttempts-1)
				}
			}
			return lastErr