fix(engine): stop leaking errors from the operator

pkg/operator/dagoperator/dag.go

@@ -69,7 +69,7 @@ func isDagConditionEvalErr(err error) bool {
 }
 
 type dagEngineError struct {
-	errorType    string
+	errorType    v1contracts.DurableTaskErrorType
 	errorMessage string
 }
 
@@ -82,17 +82,30 @@ func isDagEngineErr(err error) bool {
 	return errors.As(err, &e)
 }
 
-// these errors can be returned to the client verbatim. everything else is an
-// internal failure whose details stay in the engine logs.
-func isUserFacingDagErr(err error) bool {
-	var nonDeterminismErr *repository.NonDeterminismError
+func clientErrorMessageForDagErr(err error) (message string, ok bool) {
+	var childFailedErr *dagChildFailedError
+	var cancelledErr *dagCancelledError
+	var conditionEvalErr *dagConditionEvalError
 	var celFailureErr *operator.DAGStepCELFailureError
-
-	return isDagChildFailedErr(err) ||
-		isDagConditionEvalErr(err) ||
-		isDagEngineErr(err) ||
-		errors.As(err, &celFailureErr) ||
-		errors.As(err, &nonDeterminismErr)
+	var nonDeterminismErr *repository.NonDeterminismError
+	var engineErr *dagEngineError
+
+	switch {
+	case errors.As(err, &childFailedErr):
+		return childFailedErr.Error(), true
+	case errors.As(err, &cancelledErr):
+		return cancelledErr.Error(), true
+	case errors.As(err, &conditionEvalErr):
+		return conditionEvalErr.Error(), true
+	case errors.As(err, &celFailureErr):
+		return celFailureErr.Error(), true
+	case errors.As(err, &nonDeterminismErr):
+		return nonDeterminismErr.Error(), true
+	case errors.As(err, &engineErr) && engineErr.errorType == v1contracts.DurableTaskErrorType_DURABLE_TASK_ERROR_TYPE_NONDETERMINISM:
+		return engineErr.errorMessage, true
+	default:
+		return "", false
+	}
 }
 
 func isDagNonRetryablePgErr(err error) bool {
@@ -678,7 +691,7 @@ func (d *dag) taskConsumer(ctx context.Context, resp *v1contracts.DurableTaskRes
 
 		// Dropping this would leave the run waiting forever for an ack or completion that
 		// will never arrive (e.g. a nondeterminism error means the wait was never committed).
-		d.err = &dagEngineError{errorType: m.Error.GetErrorType().String(), errorMessage: m.Error.GetErrorMessage()}
+		d.err = &dagEngineError{errorType: m.Error.GetErrorType(), errorMessage: m.Error.GetErrorMessage()}
 	}
 }
 

pkg/operator/dagoperator/operator.go

@@ -407,8 +407,9 @@ func (d *DAGOperator) run(action *contracts.AssignedAction) error {
 
 			return nil
 		}
-		if isDagCancelledErr(dagErr) {
-			return d.cancelDAG(span, action, dagErr.Error())
+		var cancelledErr *dagCancelledError
+		if errors.As(dagErr, &cancelledErr) {
+			return d.cancelDAG(span, action, cancelledErr.Error())
 		}
 		if errors.Is(dagErr, context.Canceled) {
 			return d.handleRunCancellation(span, action, tasks, dagErr)
@@ -554,22 +555,26 @@ func (d *DAGOperator) handleRunCancellation(span trace.Span, action *contracts.A
 		}
 	}
 
-	return d.cancelDAG(span, action, err.Error())
+	d.Logger().Debug().Err(err).
+		Str("task_run_external_id", action.TaskRunExternalId).
+		Msg("dag run cancelled")
+
+	return d.cancelDAG(span, action, repository.TaskCancelledErrorMessage)
 }
 
 func (d *DAGOperator) fail(span trace.Span, action *contracts.AssignedAction, err error, shouldNotRetry bool) error {
 	span.RecordError(err)
 	span.SetStatus(telemetry_codes.Error, "dag run failed")
 	span.SetAttributes(attribute.Bool("dagoperator.should_not_retry", shouldNotRetry))
 
-	clientErrorMessage := err.Error()
+	clientErrorMessage := fmt.Sprintf("internal error while orchestrating the dag run (task run %s)", action.TaskRunExternalId)
 
-	if !isUserFacingDagErr(err) {
+	if userFacingMessage, ok := clientErrorMessageForDagErr(err); ok {
+		clientErrorMessage = "dag failed: " + userFacingMessage
+	} else {
 		d.Logger().Error().Err(err).
 			Str("task_run_external_id", action.TaskRunExternalId).
 			Msg("dag orchestration failed with an internal error")
-
-		clientErrorMessage = fmt.Sprintf("internal error while orchestrating the dag run (task run %s)", action.TaskRunExternalId)
 	}
 
 	if reportErr := d.SendFailed(action, clientErrorMessage, shouldNotRetry); reportErr != nil {

pkg/operator/hostgrpc/host_test.go

@@ -752,7 +752,7 @@ func TestOpenDeliversActionsToHandler(t *testing.T) {
 	ev := session.sentEvents()[0]
 	assert.Equal(t, contracts.StepActionEventType_STEP_EVENT_TYPE_FAILED, ev.EventType)
 	assert.Equal(t, "run-2", ev.TaskRunExternalId)
-	assert.Equal(t, "no capacity", ev.EventPayload)
+	assert.Equal(t, refusedActionClientErrorMessage, ev.EventPayload, "the handler's error must not reach the client")
 	assert.Nil(t, ev.ShouldNotRetry)
 	require.NotNil(t, ev.RetryCount)
 	assert.Equal(t, int32(2), *ev.RetryCount)
@@ -1207,7 +1207,7 @@ func TestCancelsBypassBlockedStarts(t *testing.T) {
 	refused := fs.sentEvents()[0]
 	assert.Equal(t, "run-3", refused.TaskRunExternalId)
 	assert.Equal(t, contracts.StepActionEventType_STEP_EVENT_TYPE_FAILED, refused.EventType)
-	assert.Contains(t, refused.EventPayload, "start queue")
+	assert.Equal(t, refusedActionClientErrorMessage, refused.EventPayload)
 	assert.Nil(t, refused.ShouldNotRetry)
 
 	assert.Len(t, handler.handled(), 2, "the queued start waits for the handler")

pkg/operator/hostgrpc/session.go

@@ -22,6 +22,8 @@ const (
 	// refusalReportTimeout bounds the failure report for an action the handler refused.
 	refusalReportTimeout = 30 * time.Second
 
+	refusedActionClientErrorMessage = "the operator could not accept the task; it will be retried according to its retry policy"
+
 	// minStartQueueSize is the least number of starts a session queues ahead of its handler.
 	minStartQueueSize = 64
 
@@ -281,7 +283,7 @@ func (s *session) refuse(action *contracts.AssignedAction, err error) {
 		ActionId:          action.ActionId,
 		EventTimestamp:    timestamppb.Now(),
 		EventType:         contracts.StepActionEventType_STEP_EVENT_TYPE_FAILED,
-		EventPayload:      err.Error(),
+		EventPayload:      refusedActionClientErrorMessage,
 		RetryCount:        &retryCount,
 	}); reportErr != nil {
 		s.l.Error().Err(reportErr).Str("worker_id", s.workerId()).Str("task_run_external_id", action.TaskRunExternalId).Msg("could not report the refused action")