From adacc301d24dc591de00a337c225db1ed7182cfa Mon Sep 17 00:00:00 2001 From: mrkaye97 Date: Wed, 14 Oct 2026 17:42:29 -0400 Subject: [PATCH] fix: stop leaking errors over the operator --- pkg/operator/dagoperator/dag.go | 37 +++++++++++++++++++--------- pkg/operator/dagoperator/operator.go | 19 ++++++++------ pkg/operator/hostgrpc/host_test.go | 4 +-- pkg/operator/hostgrpc/session.go | 4 ++- 4 files changed, 42 insertions(+), 22 deletions(-) diff --git a/pkg/operator/dagoperator/dag.go b/pkg/operator/dagoperator/dag.go index 90ce5e2e2b..a530bbcf8d 100644 --- a/pkg/operator/dagoperator/dag.go +++ b/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()} } } diff --git a/pkg/operator/dagoperator/operator.go b/pkg/operator/dagoperator/operator.go index c925d2904b..0f06141c72 100644 --- a/pkg/operator/dagoperator/operator.go +++ b/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,7 +555,11 @@ 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 { @@ -562,14 +567,14 @@ func (d *DAGOperator) fail(span trace.Span, action *contracts.AssignedAction, er 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 { diff --git a/pkg/operator/hostgrpc/host_test.go b/pkg/operator/hostgrpc/host_test.go index b180c0fd2f..d5150cc7f4 100644 --- a/pkg/operator/hostgrpc/host_test.go +++ b/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") diff --git a/pkg/operator/hostgrpc/session.go b/pkg/operator/hostgrpc/session.go index e543dd4e0a..2c1c052bbc 100644 --- a/pkg/operator/hostgrpc/session.go +++ b/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")