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) {
@@ -707,10 +721,32 @@ func (d *DispatcherImpl) populateTaskData(
 
 	inputs, 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.
+	var notFoundErr *v1.PayloadNotFoundError
+
+	if err != nil && errors.As(err, &notFoundErr) {
+		// Tasks whose payloads are permanently gone are failed (otherwise they'd be requeued
+		// forever); the rest of the batch proceeds with the partial results.
+		missing := make(map[v1.RetrievePayloadOpts]struct{}, len(notFoundErr.Missing))
+
+		for _, opt := range notFoundErr.Missing {
+			missing[opt] = struct{}{}
+		}
+
+		remaining := make([]*sqlcv1.V1Task, 0, len(bulkDatas))
+
+		for i, task := range bulkDatas {
+			if _, ok := missing[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
+		err = nil
+	}
+
 	if err != nil {
 		for _, task := range bulkDatas {
 			requeue(task)
@@ -956,6 +992,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/payloadstore.go

@@ -22,6 +22,42 @@ 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")
+
+// ExternalStoreNotFoundError is returned by ExternalStore.Retrieve implementations when one or
+// more requested payloads are permanently missing from the store. The returned result map still
+// contains every payload that was found, so callers can handle the missing payloads and proceed
+// with the rest.
+type ExternalStoreNotFoundError struct {
+	Missing []RetrieveFromExternalOpts
+}
+
+func (e *ExternalStoreNotFoundError) Error() string {
+	return fmt.Sprintf("%d payload(s) not found in external store", len(e.Missing))
+}
+
+func (e *ExternalStoreNotFoundError) Unwrap() error {
+	return ErrPayloadNotFound
+}
+
+// PayloadNotFoundError is returned by PayloadStoreRepository.Retrieve when one or more requested
+// payloads are permanently missing from the external store. The returned result map still
+// contains every payload that was found, so callers can handle the missing payloads and proceed
+// with the rest.
+type PayloadNotFoundError struct {
+	Missing []RetrievePayloadOpts
+}
+
+func (e *PayloadNotFoundError) Error() string {
+	return fmt.Sprintf("%d payload(s) not found in external store", len(e.Missing))
+}
+
+func (e *PayloadNotFoundError) Unwrap() error {
+	return ErrPayloadNotFound
+}
+
 type StorePayloadOpts struct {
 	Id         int64
 	InsertedAt pgtype.Timestamptz
@@ -422,7 +458,10 @@ func (p *payloadStoreRepositoryImpl) retrieve(ctx context.Context, tx sqlcv1.DBT
 
 	if len(retrieveFromExternalOpts) > 0 {
 		externalData, err := p.RetrieveFromExternal(ctx, retrieveFromExternalOpts...)
-		if err != nil {
+
+		var notFoundErr *ExternalStoreNotFoundError
+
+		if err != nil && !errors.As(err, &notFoundErr) {
 			return nil, fmt.Errorf("failed to retrieve external payloads: %w", err)
 		}
 
@@ -431,6 +470,20 @@ func (p *payloadStoreRepositoryImpl) retrieve(ctx context.Context, tx sqlcv1.DBT
 				optsToPayload[opt] = data
 			}
 		}
+
+		if notFoundErr != nil {
+			missing := make([]RetrievePayloadOpts, 0, len(notFoundErr.Missing))
+
+			for _, externalOpt := range notFoundErr.Missing {
+				if opt, exists := retrieveFromExternalOptsToOpts[externalOpt]; exists {
+					missing = append(missing, opt)
+				}
+			}
+
+			// return the payloads which were found alongside the error so callers can proceed
+			// with the rest of the batch
+			return optsToPayload, &PayloadNotFoundError{Missing: missing}
+		}
 	}
 
 	return optsToPayload, nil