diff --git a/internal/msgqueue/rabbitmq/permanent_preack_test.go b/internal/msgqueue/rabbitmq/permanent_preack_test.go deleted file mode 100644 index c96cf99f86..0000000000 --- a/internal/msgqueue/rabbitmq/permanent_preack_test.go +++ /dev/null @@ -1,33 +0,0 @@ -package rabbitmq - -import ( - "errors" - "fmt" - "testing" - - "github.com/jackc/pgerrcode" - "github.com/jackc/pgx/v5/pgconn" -) - -func TestIsPermanentPreAckError_PgError22P02(t *testing.T) { - pgErr := &pgconn.PgError{Code: pgerrcode.InvalidTextRepresentation, Message: "invalid input syntax for type json"} - wrapped := fmt.Errorf("wrap: %w", pgErr) - - if !isPermanentPreAckError(wrapped) { - t.Fatalf("expected true for wrapped pg error 22P02") - } -} - -func TestIsPermanentPreAckError_StringFallback(t *testing.T) { - err := errors.New("ERROR: invalid input syntax for type json (SQLSTATE 22P02)") - if !isPermanentPreAckError(err) { - t.Fatalf("expected true for sqlstate 22P02 string fallback") - } -} - -func TestIsPermanentPreAckError_OtherError(t *testing.T) { - err := errors.New("some transient error") - if isPermanentPreAckError(err) { - t.Fatalf("expected false for non-permanent error") - } -} diff --git a/internal/msgqueue/rabbitmq/rabbitmq.go b/internal/msgqueue/rabbitmq/rabbitmq.go index 9bbc73c309..73c81cdf56 100644 --- a/internal/msgqueue/rabbitmq/rabbitmq.go +++ b/internal/msgqueue/rabbitmq/rabbitmq.go @@ -4,18 +4,14 @@ import ( "context" "crypto/sha256" "encoding/json" - "errors" "fmt" "os" "slices" - "strings" "sync" "time" "github.com/google/uuid" lru "github.com/hashicorp/golang-lru/v2" - "github.com/jackc/pgerrcode" - "github.com/jackc/pgx/v5/pgconn" amqp "github.com/rabbitmq/amqp091-go" "github.com/rs/zerolog" "go.opentelemetry.io/otel/attribute" @@ -831,21 +827,6 @@ func (t *MessageQueueImpl) subscribe( t.l.Debug().Msgf("(session: %d) got msg", session) if err := preAck(msg); err != nil { - if isPermanentPreAckError(err) { - t.l.Error(). - Err(err). - Str("message_id", msg.ID). - Str("tenant_id", msg.TenantID.String()). - Int("num_payloads", len(msg.Payloads)). - Msg("dropping message due to permanent pre-ack error") - - if ackErr := rabbitMsg.Ack(false); ackErr != nil { - t.l.Error().Err(ackErr).Msg("error acknowledging message after permanent pre-ack error") - } - - return - } - t.l.Error().Msgf("error in pre-ack on msg %s: %v", msg.ID, err) // nack the message @@ -911,31 +892,6 @@ func (t *MessageQueueImpl) subscribe( return cleanup, nil } -func isPermanentPreAckError(err error) bool { - if err == nil { - return false - } - - var pgErr *pgconn.PgError - if errors.As(err, &pgErr) { - // invalid input syntax for type json / jsonb - if pgErr.Code == pgerrcode.InvalidTextRepresentation { - return true - } - } - - // Fallback: some error paths may lose pg error type info. - errStr := err.Error() - if strings.Contains(errStr, fmt.Sprintf("SQLSTATE %s", pgerrcode.InvalidTextRepresentation)) { - return true - } - if strings.Contains(errStr, "invalid input syntax for type json") { - return true - } - - return false -} - // identity returns the same host/process unique string for the lifetime of // this process so that subscriber reconnections reuse the same queue name. func identity() string { diff --git a/internal/queueutils/permanent_errors.go b/internal/queueutils/permanent_errors.go new file mode 100644 index 0000000000..1b105218c6 --- /dev/null +++ b/internal/queueutils/permanent_errors.go @@ -0,0 +1,43 @@ +package queueutils + +import ( + "errors" + "fmt" + "strings" + + "github.com/jackc/pgerrcode" + "github.com/jackc/pgx/v5/pgconn" + + "github.com/hatchet-dev/hatchet/pkg/repository" +) + +// PermanentConsumerErrorReason returns a stable reason for consumer errors that should be shed instead of retried. +func PermanentConsumerErrorReason(err error) string { + if err == nil { + return "" + } + + if errors.Is(err, repository.ErrExternalPayloadNotFound) { + return "external_payload_not_found" + } + + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) && pgErr.Code == pgerrcode.InvalidTextRepresentation { + return "invalid_json" + } + + errStr := err.Error() + if strings.Contains(errStr, fmt.Sprintf("SQLSTATE %s", pgerrcode.InvalidTextRepresentation)) { + return "invalid_json" + } + if strings.Contains(errStr, "invalid input syntax for type json") { + return "invalid_json" + } + + return "" +} + +// IsPermanentConsumerError returns true when a consumer can drop the offending payload from a buffered batch. +func IsPermanentConsumerError(err error) bool { + return PermanentConsumerErrorReason(err) != "" +} diff --git a/internal/queueutils/permanent_errors_test.go b/internal/queueutils/permanent_errors_test.go new file mode 100644 index 0000000000..cab6b75168 --- /dev/null +++ b/internal/queueutils/permanent_errors_test.go @@ -0,0 +1,51 @@ +package queueutils + +import ( + "errors" + "fmt" + "testing" + + "github.com/jackc/pgerrcode" + "github.com/jackc/pgx/v5/pgconn" + + "github.com/hatchet-dev/hatchet/pkg/repository" +) + +func TestIsPermanentConsumerError_PgError22P02(t *testing.T) { + pgErr := &pgconn.PgError{Code: pgerrcode.InvalidTextRepresentation, Message: "invalid input syntax for type json"} + wrapped := fmt.Errorf("wrap: %w", pgErr) + + if !IsPermanentConsumerError(wrapped) { + t.Fatalf("expected true for wrapped pg error 22P02") + } +} + +func TestIsPermanentConsumerError_StringFallback(t *testing.T) { + err := errors.New("ERROR: invalid input syntax for type json (SQLSTATE 22P02)") + if !IsPermanentConsumerError(err) { + t.Fatalf("expected true for sqlstate 22P02 string fallback") + } +} + +func TestIsPermanentConsumerError_ExternalPayloadNotFound(t *testing.T) { + err := fmt.Errorf("wrap: %w", &repository.ExternalPayloadNotFoundError{ + Kind: repository.ExternalPayloadNotFoundKindIndexFile, + Key: "index/2026-06-11/example.index", + Err: errors.New("key not found"), + }) + + if !IsPermanentConsumerError(err) { + t.Fatalf("expected true for external payload not found error") + } + + if got := PermanentConsumerErrorReason(err); got != "external_payload_not_found" { + t.Fatalf("expected external payload reason, got %q", got) + } +} + +func TestIsPermanentConsumerError_OtherError(t *testing.T) { + err := errors.New("some transient error") + if IsPermanentConsumerError(err) { + t.Fatalf("expected false for non-permanent error") + } +} diff --git a/internal/services/controllers/olap/controller.go b/internal/services/controllers/olap/controller.go index 26d34a671b..b91a5fe15f 100644 --- a/internal/services/controllers/olap/controller.go +++ b/internal/services/controllers/olap/controller.go @@ -588,6 +588,26 @@ func (tc *OLAPControllerImpl) handleCreatedTask(ctx context.Context, tenantId uu result, workflowRunIdsOfLocksNotAcquired, err := tc.repo.OLAP().CreateTasks(ctx, tenantId, extractCreatedTaskData(createTaskOpts)) if err != nil { + if queueutils.IsPermanentConsumerError(err) { + tc.l.Error(). + Ctx(ctx). + Err(err). + Str("permanent_consumer_error_reason", queueutils.PermanentConsumerErrorReason(err)). + Int("num_payloads", len(createTaskOpts)). + Msg("shedding created task batch after permanent consumer error") + + createTaskOpts, err = tc.shedPermanentCreatedTaskErrors(ctx, tenantId, createTaskOpts) + if err != nil { + return fmt.Errorf("failed to shed permanent created task errors: %w", err) + } + + if len(createTaskOpts) > 0 && attempts < maxLockAttempts { + tc.sleepWithBackoff(attempts) + } + + continue + } + return fmt.Errorf("failed to create tasks: %w", err) } @@ -683,6 +703,26 @@ func (tc *OLAPControllerImpl) handleCreatedDAG(ctx context.Context, tenantId uui workflowRunIdsOfLocksNotAcquired, err := tc.repo.OLAP().CreateDAGs(ctx, tenantId, extractCreatedDAGData(createDAGOpts)) if err != nil { + if queueutils.IsPermanentConsumerError(err) { + tc.l.Error(). + Ctx(ctx). + Err(err). + Str("permanent_consumer_error_reason", queueutils.PermanentConsumerErrorReason(err)). + Int("num_payloads", len(createDAGOpts)). + Msg("shedding created DAG batch after permanent consumer error") + + createDAGOpts, err = tc.shedPermanentCreatedDAGErrors(ctx, tenantId, createDAGOpts) + if err != nil { + return fmt.Errorf("failed to shed permanent created DAG errors: %w", err) + } + + if len(createDAGOpts) > 0 && attempts < maxLockAttempts { + tc.sleepWithBackoff(attempts) + } + + continue + } + return fmt.Errorf("failed to create DAGs: %w", err) } @@ -921,6 +961,7 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t timestamps := make([]pgtype.Timestamptz, 0) eventExternalIds := make([]uuid.UUID, 0) msgIxToSpanEvent := make(map[int]engineSpanEvent) + externalIdToSpanEvent := make(map[uuid.UUID]engineSpanEvent) for ix, msg := range msgs { taskMeta := taskIdsToMetas[msg.TaskId] @@ -950,7 +991,7 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t eventExternalIdToWorkflowRunId[externalId] = taskMeta.WorkflowRunID externalIdToMsg[externalId] = msg - msgIxToSpanEvent[ix] = engineSpanEvent{ + spanEvent := engineSpanEvent{ taskID: msg.TaskId, insertedAt: taskMeta.InsertedAt.Time, taskInsertedAt: taskMeta.InsertedAt.Time, @@ -968,6 +1009,8 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t workflowVersionID: taskMeta.WorkflowVersionID, stepID: taskMeta.StepID, } + msgIxToSpanEvent[ix] = spanEvent + externalIdToSpanEvent[externalId] = spanEvent if msg.WorkerId != nil { workerIds = append(workerIds, *msg.WorkerId) @@ -1085,6 +1128,26 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t result, workflowRunIdsOfLocksNotAcquired, err := tc.repo.OLAP().CreateTaskEvents(ctx, tenantId, opts, eventExternalIdToWorkflowRunId) if err != nil { + if queueutils.IsPermanentConsumerError(err) { + tc.l.Error(). + Ctx(ctx). + Err(err). + Str("permanent_consumer_error_reason", queueutils.PermanentConsumerErrorReason(err)). + Int("num_payloads", len(opts)). + Msg("shedding monitoring event batch after permanent consumer error") + + opts, err = tc.shedPermanentMonitoringEventErrors(ctx, tenantId, opts, eventExternalIdToWorkflowRunId, externalIdToMsg, externalIdToSpanEvent, processedRunIDs) + if err != nil { + return fmt.Errorf("failed to shed permanent monitoring event errors: %w", err) + } + + if len(opts) > 0 && attempts < maxLockAttempts { + tc.sleepWithBackoff(attempts) + } + + continue + } + return fmt.Errorf("failed to create task events: %w", err) } @@ -1152,6 +1215,137 @@ func (tc *OLAPControllerImpl) handleCreateMonitoringEvent(ctx context.Context, t return nil } +func (tc *OLAPControllerImpl) shedPermanentCreatedTaskErrors(ctx context.Context, tenantId uuid.UUID, createTaskOpts []*tasktypes.CreatedTaskPayload) ([]*tasktypes.CreatedTaskPayload, error) { + remainingOpts := make([]*tasktypes.CreatedTaskPayload, 0) + + for _, opt := range createTaskOpts { + result, workflowRunIdsOfLocksNotAcquired, err := tc.repo.OLAP().CreateTasks(ctx, tenantId, []*v1.V1TaskWithPayload{opt.V1TaskWithPayload}) + if err != nil { + if queueutils.IsPermanentConsumerError(err) { + tc.l.Error(). + Ctx(ctx). + Err(err). + Str("permanent_consumer_error_reason", queueutils.PermanentConsumerErrorReason(err)). + Int64("task_id", opt.ID). + Str("task_external_id", opt.ExternalID.String()). + Str("workflow_run_id", opt.WorkflowRunID.String()). + Msg("dropping created task after permanent consumer error") + continue + } + + return nil, err + } + + if _, notAcquired := workflowRunIdsOfLocksNotAcquired[opt.WorkflowRunID]; notAcquired { + remainingOpts = append(remainingOpts, opt) + continue + } + + if result != nil { + if err := tc.notifyStatusUpdates(ctx, result); err != nil { + return nil, err + } + } + + tc.emitStandaloneTaskRootSpans(ctx, tenantId, []*v1.V1TaskWithPayload{opt.V1TaskWithPayload}) + } + + return remainingOpts, nil +} + +func (tc *OLAPControllerImpl) shedPermanentCreatedDAGErrors(ctx context.Context, tenantId uuid.UUID, createDAGOpts []*tasktypes.CreatedDAGPayload) ([]*tasktypes.CreatedDAGPayload, error) { + remainingOpts := make([]*tasktypes.CreatedDAGPayload, 0) + + for _, opt := range createDAGOpts { + workflowRunIdsOfLocksNotAcquired, err := tc.repo.OLAP().CreateDAGs(ctx, tenantId, []*v1.DAGWithData{opt.DAGWithData}) + if err != nil { + if queueutils.IsPermanentConsumerError(err) { + tc.l.Error(). + Ctx(ctx). + Err(err). + Str("permanent_consumer_error_reason", queueutils.PermanentConsumerErrorReason(err)). + Int64("dag_id", opt.ID). + Str("dag_external_id", opt.ExternalID.String()). + Msg("dropping created DAG after permanent consumer error") + continue + } + + return nil, err + } + + if _, notAcquired := workflowRunIdsOfLocksNotAcquired[opt.ExternalID]; notAcquired { + remainingOpts = append(remainingOpts, opt) + continue + } + + tc.emitWorkflowRunRootSpans(ctx, tenantId, []*v1.DAGWithData{opt.DAGWithData}) + } + + return remainingOpts, nil +} + +func (tc *OLAPControllerImpl) shedPermanentMonitoringEventErrors( + ctx context.Context, + tenantId uuid.UUID, + opts []sqlcv1.CreateTaskEventsOLAPParams, + eventExternalIdToWorkflowRunId map[uuid.UUID]uuid.UUID, + externalIdToMsg map[uuid.UUID]*tasktypes.CreateMonitoringEventPayload, + externalIdToSpanEvent map[uuid.UUID]engineSpanEvent, + processedRunIDs map[uuid.UUID]bool, +) ([]sqlcv1.CreateTaskEventsOLAPParams, error) { + remainingOpts := make([]sqlcv1.CreateTaskEventsOLAPParams, 0) + + for _, opt := range opts { + workflowRunId, ok := eventExternalIdToWorkflowRunId[opt.ExternalID] + if !ok { + tc.l.Error().Ctx(ctx).Str("event_external_id", opt.ExternalID.String()).Msg("could not find workflow run id for monitoring event") + continue + } + + result, workflowRunIdsOfLocksNotAcquired, err := tc.repo.OLAP().CreateTaskEvents(ctx, tenantId, []sqlcv1.CreateTaskEventsOLAPParams{opt}, map[uuid.UUID]uuid.UUID{ + opt.ExternalID: workflowRunId, + }) + if err != nil { + if queueutils.IsPermanentConsumerError(err) { + event := tc.l.Error(). + Ctx(ctx). + Err(err). + Str("permanent_consumer_error_reason", queueutils.PermanentConsumerErrorReason(err)). + Str("event_external_id", opt.ExternalID.String()). + Str("event_type", string(opt.EventType)) + + if msg, ok := externalIdToMsg[opt.ExternalID]; ok { + event.Int64("task_id", msg.TaskId) + } + + event.Msg("dropping monitoring event after permanent consumer error") + continue + } + + return nil, err + } + + if _, notAcquired := workflowRunIdsOfLocksNotAcquired[workflowRunId]; notAcquired { + remainingOpts = append(remainingOpts, opt) + continue + } + + if result != nil { + if err := tc.notifyStatusUpdates(ctx, result); err != nil { + return nil, err + } + } + + if spanEvent, ok := externalIdToSpanEvent[opt.ExternalID]; ok { + tc.synthesizeEngineSpans(ctx, tenantId, []engineSpanEvent{spanEvent}) + } + + processedRunIDs[workflowRunId] = true + } + + return remainingOpts, nil +} + func (tc *OLAPControllerImpl) republishCreatedTasks(ctx context.Context, tenantId uuid.UUID, tasks []*tasktypes.CreatedTaskPayload) error { for _, task := range tasks { updated := *task diff --git a/pkg/repository/payloadstore.go b/pkg/repository/payloadstore.go index f3aa42f66b..cc4e11e7dc 100644 --- a/pkg/repository/payloadstore.go +++ b/pkg/repository/payloadstore.go @@ -89,6 +89,42 @@ type PayloadLocation string type ExternalPayloadLocationKey string type ExternalIndexFileLocationKey string +var ErrExternalPayloadNotFound = errors.New("external payload not found") + +type ExternalPayloadNotFoundKind string + +const ( + ExternalPayloadNotFoundKindIndexFile ExternalPayloadNotFoundKind = "index_file" +) + +type ExternalPayloadNotFoundError struct { + Err error + Kind ExternalPayloadNotFoundKind + Key string +} + +func (e *ExternalPayloadNotFoundError) Error() string { + msg := ErrExternalPayloadNotFound.Error() + if e.Kind != "" { + msg = fmt.Sprintf("%s: %s", msg, e.Kind) + } + if e.Key != "" { + msg = fmt.Sprintf("%s %s", msg, e.Key) + } + if e.Err != nil { + msg = fmt.Sprintf("%s: %v", msg, e.Err) + } + return msg +} + +func (e *ExternalPayloadNotFoundError) Unwrap() error { + return e.Err +} + +func (e *ExternalPayloadNotFoundError) Is(target error) bool { + return target == ErrExternalPayloadNotFound +} + type CreateIndexBlockOpts struct { PartitionDate PartitionDate BlockLowerExternalIdBound uuid.UUID