fix(cloud): drop tasks past payload store retention

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()

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 {

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]

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)

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]
 		}