Merge #157145

157145: backup, jobs: reliably emit context cancelled on drain, treat as retryable  r=dt a=dt

Closes #156337.

Co-authored-by: David Taylor <davidt@davidt.io>

Author: craig[bot]

pkg/ccl/changefeedccl/changefeed_processors.go

@@ -814,10 +814,10 @@ func (ca *changeAggregator) Next() (rowenc.EncDatumRow, *execinfrapb.ProducerMet
 					err = kvFeedErr
 				}
 			}
-			// Shut down the poller if it wasn't already.
-			ca.cancel()
 			log.Changefeed.Warningf(ca.Ctx(), "moving to draining due to error from tick: %v", err)
 			ca.MoveToDraining(err)
+			// Shut down the poller if it wasn't already.
+			ca.cancel()
 			break
 		}
 	}

pkg/jobs/joberror/errors.go

@@ -6,6 +6,8 @@
 package joberror
 
 import (
+	"context"
+
 	"github.com/cockroachdb/cockroach/pkg/kv/kvclient/kvcoord"
 	"github.com/cockroachdb/cockroach/pkg/sql/flowinfra"
 	"github.com/cockroachdb/cockroach/pkg/sql/sqlerrors"
@@ -32,5 +34,6 @@ func IsPermanentBulkJobError(err error) bool {
 		!errors.Is(err, circuit.ErrBreakerOpen) &&
 		!sysutil.IsErrConnectionReset(err) &&
 		!sysutil.IsErrConnectionRefused(err) &&
-		!errors.Is(err, sqlinstance.NonExistentInstanceError)
+		!errors.Is(err, sqlinstance.NonExistentInstanceError) &&
+		!errors.Is(err, context.Canceled)
 }

pkg/sql/execinfra/processorsbase.go

@@ -526,7 +526,9 @@ const (
 // at init() time), then we move straight to the StateTrailingMeta.
 //
 // An error can be optionally passed. It will be the first piece of metadata
-// returned by DrainHelper().
+// returned by DrainHelper(), unless the processor's context has already been
+// canceled, in which case the context's error is returned instead (as often an
+// error passed to MoveToDraining() is a consequence of context cancellation).
 //
 // MoveToDraining should only be called from the main goroutine of the
 // processor.
@@ -547,6 +549,18 @@ func (pb *ProcessorBaseNoHelper) MoveToDraining(err error) {
 	}
 
 	if err != nil {
+		// If processor ctx was canceled, reply with that err rather than whatever
+		// error was passed to MoveToDraining by a processor running on top of a
+		// canceled context, which is expected to error. Generally the error passed
+		// in this case will be context.Canceled anyway, but doing this ensures that
+		// distsql can promise that if it cancels a context, the emitted error will
+		// reflect that, making cancellation detectable by callers.
+		if pb.Ctx().Err() != nil {
+			if !errors.Is(err, pb.Ctx().Err()) {
+				log.Dev.Warningf(pb.Ctx(), "overriding non-cancelation emitted after context cancellation: %+v", err)
+			}
+			err = pb.Ctx().Err()
+		}
 		pb.trailingMeta = append(pb.trailingMeta, execinfrapb.ProducerMetadata{Err: err})
 	}
 	if pb.curInputToDrain < len(pb.inputsToDrain) {