diff --git a/internal/services/dispatcher/dispatcher.go b/internal/services/dispatcher/dispatcher.go index a160b997a8..7e0ab04a8d 100644 --- a/internal/services/dispatcher/dispatcher.go +++ b/internal/services/dispatcher/dispatcher.go @@ -509,6 +509,15 @@ func (d *DispatcherImpl) handleTaskBulkAssignedTask(ctx context.Context, msg *ms toRetryMu.Unlock() } + toFail := []*sqlcv1.V1Task{} + toFailMu := sync.Mutex{} + + fail := func(task *sqlcv1.V1Task) { + toFailMu.Lock() + toFail = append(toFail, task) + toFailMu.Unlock() + } + for _, innerMsg := range msgs { // load the step runs from the database taskIds := make([]int64, 0) @@ -517,7 +526,7 @@ func (d *DispatcherImpl) handleTaskBulkAssignedTask(ctx context.Context, msg *ms taskIds = append(taskIds, tasks...) } - taskIdToData, err := d.populateTaskData(ctx, requeue, msg.TenantID, taskIds) + taskIdToData, err := d.populateTaskData(ctx, requeue, fail, msg.TenantID, taskIds) if err != nil { // we've already handled the requeue in populateTaskData, and we've logged the error, so we just continue @@ -544,6 +553,10 @@ func (d *DispatcherImpl) handleTaskBulkAssignedTask(ctx context.Context, msg *ms outerErr = multierror.Append(outerErr, fmt.Errorf("could not retry failed tasks: %w", err)) } + if err := d.handleNonRetryableFailures(ctx, msg.TenantID, toFail); err != nil { + outerErr = multierror.Append(outerErr, fmt.Errorf("could not fail tasks with missing payloads: %w", err)) + } + if outerErr != nil { d.l.Error().Ctx(ctx).Err(outerErr).Msg("failed to handle task assigned bulk message") } @@ -642,6 +655,7 @@ type V1TaskWithPayloadAndInvocationCount struct { func (d *DispatcherImpl) populateTaskData( ctx context.Context, requeue func(task *sqlcv1.V1Task), + fail func(task *sqlcv1.V1Task), tenantId uuid.UUID, taskIds []int64, ) (map[int64]*V1TaskWithPayloadAndInvocationCount, error) { @@ -705,12 +719,8 @@ func (d *DispatcherImpl) populateTaskData( } } - inputs, err := d.repov1.Payloads().Retrieve(ctx, nil, retrievePayloadOpts...) + inputs, missing, err := d.repov1.Payloads().Retrieve(ctx, nil, retrievePayloadOpts...) - // FIXME: we should differentiate between a retryable error and a non-retryable error here; - // for example, if we're hitting an S3 rate limit for payloads that exist in S3, we should retry; - // however, if the payloads simply don't exist, we should fail the tasks instead of requeuing them. - // The tasks will eventually fail but the extra retries are wasteful. if err != nil { for _, task := range bulkDatas { requeue(task) @@ -720,6 +730,29 @@ func (d *DispatcherImpl) populateTaskData( return nil, err } + if len(missing) > 0 { + // Tasks whose payloads are permanently gone are failed (otherwise they'd be requeued + // forever); the rest of the batch proceeds with the partial results. + missingSet := make(map[v1.RetrievePayloadOpts]struct{}, len(missing)) + + for _, opt := range missing { + missingSet[opt] = struct{}{} + } + + remaining := make([]*sqlcv1.V1Task, 0, len(bulkDatas)) + + for i, task := range bulkDatas { + if _, ok := missingSet[retrievePayloadOpts[i]]; ok { + d.l.Error().Ctx(ctx).Int64("task_id", task.ID).Msg("task input payload no longer exists in external store, failing task") + fail(task) + } else { + remaining = append(remaining, task) + } + } + + bulkDatas = remaining + } + // this is to avoid a nil pointer dereference in the code below if inputs == nil { inputs = make(map[v1.RetrievePayloadOpts][]byte) @@ -956,6 +989,55 @@ func (d *DispatcherImpl) handleRetries( return retryGroup.Wait() } +// handleNonRetryableFailures permanently fails tasks whose input payloads no longer exist in the +// external payload store, since requeueing them can never succeed. +func (d *DispatcherImpl) handleNonRetryableFailures( + ctx context.Context, + tenantId uuid.UUID, + toFail []*sqlcv1.V1Task, +) error { + if len(toFail) == 0 { + return nil + } + + failCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + failGroup := errgroup.Group{} + + for _, _task := range toFail { + task := _task + + failGroup.Go(func() error { + msg, err := tasktypesv1.FailedTaskMessage( + tenantId, + task.ID, + task.InsertedAt, + task.ExternalID, + task.WorkflowRunID, + task.RetryCount, + true, + "Could not retrieve task input: the payload no longer exists in the external payload store.", + true, + ) + + if err != nil { + return fmt.Errorf("could not create failed task message: %w", err) + } + + err = d.mqv1.SendMessage(failCtx, msgqueue.TASK_PROCESSING_QUEUE, msg) + + if err != nil { + return fmt.Errorf("could not send failed task message: %w", err) + } + + return nil + }) + } + + return failGroup.Wait() +} + func (d *DispatcherImpl) handleTaskCancelled(ctx context.Context, msg *msgqueue.Message) error { ctx, span := telemetry.NewSpanWithCarrier(ctx, "tasks-cancelled", msg.OtelCarrier) defer span.End() diff --git a/pkg/repository/durable_events.go b/pkg/repository/durable_events.go index 20d6cdbc0e..76c35fcfc0 100644 --- a/pkg/repository/durable_events.go +++ b/pkg/repository/durable_events.go @@ -505,12 +505,16 @@ func (r *durableEventsRepository) GetSatisfiedDurableEvents(ctx context.Context, } } - payloads, err := r.payloadStore.Retrieve(ctx, r.pool, retrievePayloadOpts...) + payloads, missing, err := r.payloadStore.Retrieve(ctx, r.pool, retrievePayloadOpts...) if err != nil { return nil, fmt.Errorf("failed to retrieve payloads for satisfied callbacks: %w", err) } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, fmt.Errorf("failed to retrieve payloads for satisfied callbacks: %w", missingErr) + } + result := make([]*SatisfiedEventWithPayload, 0, len(rows)) for _, row := range rows { @@ -960,7 +964,8 @@ func (r *durableEventsRepository) getOrCreateEventLogEntries( var existingPayloads map[RetrievePayloadOpts][]byte if len(retrieveOpts) > 0 { var err error - existingPayloads, err = r.payloadStore.Retrieve(ctx, tx, retrieveOpts...) + // missing payloads are skipped: lookups on existingPayloads fall back to nil below + existingPayloads, _, err = r.payloadStore.Retrieve(ctx, tx, retrieveOpts...) if err != nil { existingPayloads = nil } @@ -1214,7 +1219,7 @@ func (r *durableEventsRepository) IngestDurableTaskEvent(ctx context.Context, op var nde *NonDeterminismError if errors.As(err, &nde) { var existingPayload []byte - payloads, retrieveErr := r.payloadStore.Retrieve(ctx, tx, RetrievePayloadOpts{ + payloads, _, retrieveErr := r.payloadStore.Retrieve(ctx, tx, RetrievePayloadOpts{ Id: nde.ExistingEntryId, InsertedAt: nde.ExistingEntryInsertedAt, Type: sqlcv1.V1PayloadTypeDURABLEEVENTLOGENTRYDATA, @@ -1566,12 +1571,16 @@ func (r *durableEventsRepository) handleEventLookback(ctx context.Context, tenan }) } - retrieveOptsToPayload, err := r.payloadStore.Retrieve(ctx, lookbackTx, retrievePayloadOpts...) + retrieveOptsToPayload, missing, err := r.payloadStore.Retrieve(ctx, lookbackTx, retrievePayloadOpts...) if err != nil { return nil, fmt.Errorf("failed to retrieve payloads for recent user events for retroactive matching: %w", err) } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, fmt.Errorf("failed to retrieve payloads for recent user events for retroactive matching: %w", missingErr) + } + retroCandidates := make([]CandidateEventMatch, 0, len(previousEventsFound)) for _, row := range previousEventsFound { diff --git a/pkg/repository/ids.go b/pkg/repository/ids.go index c4dddd3103..1e8479e06b 100644 --- a/pkg/repository/ids.go +++ b/pkg/repository/ids.go @@ -189,12 +189,16 @@ func (s *sharedRepository) generateExternalIdsForChildWorkflows(ctx context.Cont } } - payloads, err := s.payloadStore.Retrieve(ctx, tx, retrievePayloadOpts...) + payloads, missing, err := s.payloadStore.Retrieve(ctx, tx, retrievePayloadOpts...) if err != nil { return err } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return missingErr + } + // for each locked event, write the correct external id to the opt for _, lockedEvent := range lockedEvents { opt := spawnKeyToOpt[lockedEvent.EventKey.String] diff --git a/pkg/repository/match.go b/pkg/repository/match.go index e5824e790e..21b5434c2f 100644 --- a/pkg/repository/match.go +++ b/pkg/repository/match.go @@ -569,12 +569,16 @@ func (m *sharedRepository) processEventMatches(ctx context.Context, tx sqlcv1.DB } } - payloads, err := m.payloadStore.Retrieve(ctx, tx, retrievePayloadOpts...) + payloads, missing, err := m.payloadStore.Retrieve(ctx, tx, retrievePayloadOpts...) if err != nil { return nil, fmt.Errorf("failed to retrieve dag input payloads: %w", err) } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, fmt.Errorf("failed to retrieve dag input payloads: %w", missingErr) + } + dagIdsToInput := make(map[int64][]byte) dagIdsToMetadata := make(map[int64][]byte) dagIdsToDesiredWorkerLabels := make(map[int64][]byte) diff --git a/pkg/repository/olap.go b/pkg/repository/olap.go index a7edff63ea..8ccb917396 100644 --- a/pkg/repository/olap.go +++ b/pkg/repository/olap.go @@ -3167,12 +3167,16 @@ func (r *OLAPRepositoryImpl) readPayloads(ctx context.Context, tx sqlcv1.DBTX, t } if len(retrieveFromExternalOpts) > 0 && r.payloadStore.ExternalStoreEnabled() { - keyToPayload, err := r.payloadStore.RetrieveFromExternal(ctx, retrieveFromExternalOpts...) + keyToPayload, missing, err := r.payloadStore.RetrieveFromExternal(ctx, retrieveFromExternalOpts...) if err != nil { return nil, err } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, missingErr + } + for externalId, opt := range externalIdToRetrieveFromExternalOpt { externalIdToPayload[externalId] = keyToPayload[opt] } diff --git a/pkg/repository/payloadstore.go b/pkg/repository/payloadstore.go index f3aa42f66b..24472e3649 100644 --- a/pkg/repository/payloadstore.go +++ b/pkg/repository/payloadstore.go @@ -22,6 +22,20 @@ import ( "github.com/hatchet-dev/hatchet/pkg/telemetry" ) +// ErrPayloadNotFound indicates that a payload is permanently missing from the external store +// (e.g. it was deleted or never uploaded), as opposed to a transient retrieval failure. +var ErrPayloadNotFound = errors.New("payload not found in external store") + +// MissingPayloadsError converts a non-empty list of missing payloads into an error wrapping +// ErrPayloadNotFound. Callers which cannot proceed with a partial result can use this to fail. +func MissingPayloadsError[T any](missing []T) error { + if len(missing) == 0 { + return nil + } + + return fmt.Errorf("%d payload(s) missing: %w", len(missing), ErrPayloadNotFound) +} + type StorePayloadOpts struct { Id int64 InsertedAt pgtype.Timestamptz @@ -98,14 +112,19 @@ type CreateIndexBlockOpts struct { type ExternalStore interface { Store(ctx context.Context, payloads ...OffloadToExternalStoreOpts) (*ExternalIndexFileLocationKey, error) - Retrieve(ctx context.Context, opts ...RetrieveFromExternalOpts) (map[RetrieveFromExternalOpts][]byte, error) + // Retrieve returns the payloads which were found, alongside the opts for any payloads which + // are permanently missing from the store, so callers can proceed with the rest of the batch. + Retrieve(ctx context.Context, opts ...RetrieveFromExternalOpts) (found map[RetrieveFromExternalOpts][]byte, missing []RetrieveFromExternalOpts, err error) } type PayloadStoreRepository interface { Store(ctx context.Context, tx sqlcv1.DBTX, payloads ...StorePayloadOpts) error - Retrieve(ctx context.Context, tx sqlcv1.DBTX, opts ...RetrievePayloadOpts) (map[RetrievePayloadOpts][]byte, error) + // Retrieve returns the payloads which were found, alongside the opts for any payloads which + // are permanently missing from the external store, so callers can proceed with the rest of + // the batch. + Retrieve(ctx context.Context, tx sqlcv1.DBTX, opts ...RetrievePayloadOpts) (found map[RetrievePayloadOpts][]byte, missing []RetrievePayloadOpts, err error) RetrieveSingle(ctx context.Context, tx sqlcv1.DBTX, opt RetrievePayloadOpts) ([]byte, error) - RetrieveFromExternal(ctx context.Context, opts ...RetrieveFromExternalOpts) (map[RetrieveFromExternalOpts][]byte, error) + RetrieveFromExternal(ctx context.Context, opts ...RetrieveFromExternalOpts) (found map[RetrieveFromExternalOpts][]byte, missing []RetrieveFromExternalOpts, err error) OverwriteExternalStore(store ExternalStore) DualWritesEnabled() bool TaskEventDualWritesEnabled() bool @@ -261,7 +280,7 @@ func (p *payloadStoreRepositoryImpl) Store(ctx context.Context, tx sqlcv1.DBTX, return err } -func (p *payloadStoreRepositoryImpl) Retrieve(ctx context.Context, tx sqlcv1.DBTX, opts ...RetrievePayloadOpts) (map[RetrievePayloadOpts][]byte, error) { +func (p *payloadStoreRepositoryImpl) Retrieve(ctx context.Context, tx sqlcv1.DBTX, opts ...RetrievePayloadOpts) (map[RetrievePayloadOpts][]byte, []RetrievePayloadOpts, error) { if tx == nil { tx = p.pool } @@ -274,12 +293,16 @@ func (p *payloadStoreRepositoryImpl) RetrieveSingle(ctx context.Context, tx sqlc tx = p.pool } - optsToPayload, err := p.retrieve(ctx, tx, opt) + optsToPayload, missing, err := p.retrieve(ctx, tx, opt) if err != nil && err != pgx.ErrNoRows { return nil, err } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, missingErr + } + if len(optsToPayload) == 0 || err == pgx.ErrNoRows { return nil, nil } @@ -287,17 +310,17 @@ func (p *payloadStoreRepositoryImpl) RetrieveSingle(ctx context.Context, tx sqlc return optsToPayload[opt], nil } -func (p *payloadStoreRepositoryImpl) RetrieveFromExternal(ctx context.Context, opts ...RetrieveFromExternalOpts) (map[RetrieveFromExternalOpts][]byte, error) { +func (p *payloadStoreRepositoryImpl) RetrieveFromExternal(ctx context.Context, opts ...RetrieveFromExternalOpts) (map[RetrieveFromExternalOpts][]byte, []RetrieveFromExternalOpts, error) { if !p.externalStoreEnabled { - return nil, fmt.Errorf("external store not enabled") + return nil, nil, fmt.Errorf("external store not enabled") } return p.externalStore.Retrieve(ctx, opts...) } -func (p *payloadStoreRepositoryImpl) retrieve(ctx context.Context, tx sqlcv1.DBTX, opts ...RetrievePayloadOpts) (map[RetrievePayloadOpts][]byte, error) { +func (p *payloadStoreRepositoryImpl) retrieve(ctx context.Context, tx sqlcv1.DBTX, opts ...RetrievePayloadOpts) (map[RetrievePayloadOpts][]byte, []RetrievePayloadOpts, error) { if len(opts) == 0 { - return make(map[RetrievePayloadOpts][]byte), nil + return make(map[RetrievePayloadOpts][]byte), nil, nil } ids := make([]int64, len(opts)) @@ -320,7 +343,7 @@ func (p *payloadStoreRepositoryImpl) retrieve(ctx context.Context, tx sqlcv1.DBT }) if err != nil { - return nil, fmt.Errorf("failed to read payload metadata: %w", err) + return nil, nil, fmt.Errorf("failed to read payload metadata: %w", err) } optsToPayload := make(map[RetrievePayloadOpts][]byte) @@ -396,7 +419,7 @@ func (p *payloadStoreRepositoryImpl) retrieve(ctx context.Context, tx sqlcv1.DBT }) if err != nil && !errors.Is(err, pgx.ErrNoRows) { - return nil, fmt.Errorf("failed to get offloaded payload index block: %w", err) + return nil, nil, fmt.Errorf("failed to get offloaded payload index block: %w", err) } for _, k := range indexFileKeys { @@ -420,10 +443,13 @@ func (p *payloadStoreRepositoryImpl) retrieve(ctx context.Context, tx sqlcv1.DBT } } + missing := make([]RetrievePayloadOpts, 0) + if len(retrieveFromExternalOpts) > 0 { - externalData, err := p.RetrieveFromExternal(ctx, retrieveFromExternalOpts...) + externalData, missingExternal, err := p.RetrieveFromExternal(ctx, retrieveFromExternalOpts...) + if err != nil { - return nil, fmt.Errorf("failed to retrieve external payloads: %w", err) + return nil, nil, fmt.Errorf("failed to retrieve external payloads: %w", err) } for retrieveFromExternalOpt, data := range externalData { @@ -431,9 +457,15 @@ func (p *payloadStoreRepositoryImpl) retrieve(ctx context.Context, tx sqlcv1.DBT optsToPayload[opt] = data } } + + for _, externalOpt := range missingExternal { + if opt, exists := retrieveFromExternalOptsToOpts[externalOpt]; exists { + missing = append(missing, opt) + } + } } - return optsToPayload, nil + return optsToPayload, missing, nil } func (p *payloadStoreRepositoryImpl) OverwriteExternalStore(store ExternalStore) { @@ -1019,6 +1051,6 @@ func (n *NoOpExternalStore) Store(ctx context.Context, payloads ...OffloadToExte return nil, fmt.Errorf("external store disabled") } -func (n *NoOpExternalStore) Retrieve(ctx context.Context, opts ...RetrieveFromExternalOpts) (map[RetrieveFromExternalOpts][]byte, error) { - return nil, fmt.Errorf("external store disabled") +func (n *NoOpExternalStore) Retrieve(ctx context.Context, opts ...RetrieveFromExternalOpts) (map[RetrieveFromExternalOpts][]byte, []RetrieveFromExternalOpts, error) { + return nil, nil, fmt.Errorf("external store disabled") } diff --git a/pkg/repository/task.go b/pkg/repository/task.go index 93327c0b26..9b7b979e02 100644 --- a/pkg/repository/task.go +++ b/pkg/repository/task.go @@ -1259,12 +1259,16 @@ func (r *TaskRepositoryImpl) listTaskOutputEvents(ctx context.Context, tx sqlcv1 matchedEventToRetrieveOpts[event] = opt } - payloads, err := r.payloadStore.Retrieve(ctx, tx, retrieveOpts...) + payloads, missing, err := r.payloadStore.Retrieve(ctx, tx, retrieveOpts...) if err != nil { return nil, err } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, missingErr + } + res := make([]*TaskOutputEvent, 0, len(matchedEvents)) for _, event := range matchedEvents { @@ -3217,12 +3221,16 @@ func (r *TaskRepositoryImpl) ReplayTasks(ctx context.Context, tenantId uuid.UUID } } - payloads, err := r.payloadStore.Retrieve(ctx, tx, retrieveOpts...) + payloads, missing, err := r.payloadStore.Retrieve(ctx, tx, retrieveOpts...) if err != nil { return nil, fmt.Errorf("failed to bulk retrieve task inputs: %w", err) } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, fmt.Errorf("failed to bulk retrieve task inputs: %w", missingErr) + } + for _, task := range lockedTasks { // check whether to discard the task if task.DagID.Valid && !successfullyLockedDAGsMap[task.DagID.Int64] { @@ -3840,12 +3848,16 @@ func (r *TaskRepositoryImpl) ListTaskParentOutputs(ctx context.Context, tenantId retrieveOptToPayload[opt] = outputTask.Output } - payloads, err := r.payloadStore.Retrieve(ctx, r.pool, retrieveOpts...) + payloads, missing, err := r.payloadStore.Retrieve(ctx, r.pool, retrieveOpts...) if err != nil { return nil, fmt.Errorf("failed to retrieve task output payloads: %w", err) } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, fmt.Errorf("failed to retrieve task output payloads: %w", missingErr) + } + workflowRunIdsToOutputs := make(map[string][]*TaskOutputEvent) for retrieveOpts, workflowRunId := range retrieveOptsToWorkflowRunId { @@ -3916,12 +3928,16 @@ func (r *TaskRepositoryImpl) ListSignalCompletedEvents(ctx context.Context, tena retrieveOpts[i] = retrieveOpt } - payloads, err := r.payloadStore.Retrieve(ctx, r.pool, retrieveOpts...) + payloads, missing, err := r.payloadStore.Retrieve(ctx, r.pool, retrieveOpts...) if err != nil { return nil, fmt.Errorf("failed to retrieve task event payloads: %w", err) } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, fmt.Errorf("failed to retrieve task event payloads: %w", missingErr) + } + res := make([]*V1TaskEventWithPayload, len(signalEvents)) for i, event := range signalEvents { @@ -4348,12 +4364,16 @@ func (r *TaskRepositoryImpl) GetWorkflowRunResultDetails(ctx context.Context, te } } - payloads, err := r.payloadStore.Retrieve(ctx, r.pool, inputRetrieveOpt) + payloads, missing, err := r.payloadStore.Retrieve(ctx, r.pool, inputRetrieveOpt) if err != nil { return nil, fmt.Errorf("failed to retrieve payloads: %w", err) } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, fmt.Errorf("failed to retrieve payloads: %w", missingErr) + } + input := payloads[inputRetrieveOpt] if !isDag && len(input) > 0 { diff --git a/pkg/repository/trigger.go b/pkg/repository/trigger.go index 5f33141438..a17f604763 100644 --- a/pkg/repository/trigger.go +++ b/pkg/repository/trigger.go @@ -1601,12 +1601,16 @@ func (r *sharedRepository) registerChildWorkflows( } } - payloads, err := r.payloadStore.Retrieve(ctx, tx, retrievePayloadOpts...) + payloads, missing, err := r.payloadStore.Retrieve(ctx, tx, retrievePayloadOpts...) if err != nil { return nil, fmt.Errorf("failed to retrieve payloads for signal created events: %w", err) } + if missingErr := MissingPayloadsError(missing); missingErr != nil { + return nil, fmt.Errorf("failed to retrieve payloads for signal created events: %w", missingErr) + } + // parse the event match data, and determine whether the child external ID has already been written // we're safe to do this read since we've acquired a lock on the relevant rows rootExternalIdsToLookup := make([]uuid.UUID, 0, len(matchingEvents))