fix(server): SQS batch failure redelivers already-processed purchase messages (no BatchItemFailures)

internal/server/app.go

@@ -118,6 +118,13 @@ type ApplicationConfig struct {
 	DefaultTerm             int
 	EnableDashboard         bool
 	IsLambda                bool
+	// ReportSQSBatchItemFailures gates whether handleLambdaSQSEvent returns
+	// the partial events.SQSEventResponse{BatchItemFailures} shape (true) or
+	// the old whole-batch aggregate error (false, default). See
+	// loadReportSQSBatchItemFailures for why this must stay false until the
+	// SQS event source mapping's function_response_types actually includes
+	// "ReportBatchItemFailures" (issue #108 follow-up).
+	ReportSQSBatchItemFailures bool
 }
 
 // ExternalDeps holds pre-built external dependencies that require infrastructure.
@@ -534,6 +541,12 @@ func NewApplication(ctx context.Context, version string) (*Application, error) {
 		return nil, fmt.Errorf("invalid analytics configuration: %w", err)
 	}
 
+	reportSQSBatchItemFailures, err := loadReportSQSBatchItemFailures()
+	if err != nil {
+		return nil, fmt.Errorf("invalid SQS configuration: %w", err)
+	}
+	cfg.ReportSQSBatchItemFailures = reportSQSBatchItemFailures
+
 	log.Printf("CUDly Server initializing, version: %s", cfg.Version)
 
 	// Initialize configuration store (PostgreSQL). The store connects lazily on

internal/server/lambda.go

@@ -6,6 +6,8 @@ import (
 	"encoding/json"
 	"fmt"
 	"log"
+	"os"
+	"strconv"
 	"strings"
 
 	"github.com/aws/aws-lambda-go/events"
@@ -223,28 +225,105 @@ func isTextContentType(ct string) bool {
 }
 
 // handleLambdaSQSEvent processes SQS messages (for async purchase processing).
+//
+// When app.appConfig.ReportSQSBatchItemFailures is true, returns a partial
+// batch response (events.SQSEventResponse) naming only the records that
+// actually failed, with a nil error, instead of a single error for the
+// whole batch (issue #108). Lambda's SQS integration treats ANY non-nil
+// handler error as "the entire batch failed" and redelivers every record,
+// including ones HandleSQSMessage already processed successfully. On
+// redelivery those already-succeeded records typically fail a second time
+// (status already advanced past what the CAS-guarded handlers expect),
+// which redelivers the whole batch again and burns the redrive budget on
+// records that were never actually broken, until the one genuinely poison
+// message drags the rest of the batch to the DLQ with it.
+//
+// The partial-response shape is only honored by AWS when the SQS
+// event-source mapping sets function_response_types =
+// ["ReportBatchItemFailures"] (Terraform; see loadReportSQSBatchItemFailures
+// for why no such resource currently exists in this repo). Without that
+// setting live, AWS ignores the response body entirely and treats a nil
+// error as the WHOLE BATCH having succeeded -- deleting every message from
+// the queue, including the ones that just failed. That is a strictly worse
+// outcome than the old all-or-nothing behavior (which at least redelivered
+// and eventually DLQ'd a genuine failure): a transient error would now
+// silently drop an approval, and a poison message would vanish instead of
+// landing in the DLQ for investigation. ReportSQSBatchItemFailures defaults
+// to false specifically to keep the old, safe, all-or-nothing behavior
+// until an operator has confirmed the event source mapping is actually
+// configured to honor partial responses and flips
+// SQS_REPORT_BATCH_ITEM_FAILURES=true.
+//
+// Regardless of the flag, a batch where every record failed always returns
+// an error: there is no reason to ack (or partially ack) a batch that
+// accomplished nothing, flag or no flag.
 func (app *Application) handleLambdaSQSEvent(ctx context.Context, rawEvent json.RawMessage) (any, error) {
 	var sqsEvent events.SQSEvent
 	if err := json.Unmarshal(rawEvent, &sqsEvent); err != nil {
 		log.Printf("Failed to parse SQS event: %v", err)
 		return nil, err
 	}
 
-	var failures []string
+	failures := []events.SQSBatchItemFailure{}
 	for _rvc := range sqsEvent.Records {
 		record := sqsEvent.Records[_rvc]
 		log.Printf("Processing SQS message: %s", record.MessageId)
 		if err := app.HandleSQSMessage(ctx, record.Body); err != nil {
 			log.Printf("Failed to process message %s: %v", record.MessageId, err)
-			failures = append(failures, record.MessageId)
+			failures = append(failures, events.SQSBatchItemFailure{ItemIdentifier: record.MessageId})
 		}
 	}
 
-	if len(failures) > 0 {
-		return nil, fmt.Errorf("failed to process %d SQS message(s): %v", len(failures), failures)
+	if len(failures) == 0 {
+		return events.SQSEventResponse{BatchItemFailures: failures}, nil
 	}
 
-	return map[string]string{"status": "processed"}, nil
+	if len(failures) == len(sqsEvent.Records) {
+		return nil, fmt.Errorf("failed to process all %d SQS message(s) in batch: %v", len(failures), sqsBatchItemFailureIDs(failures))
+	}
+
+	if !app.appConfig.ReportSQSBatchItemFailures {
+		// Fail closed: see the function doc comment above for why acking
+		// (even partially) without the Terraform side confirmed live would
+		// silently drop failed messages instead of redelivering them.
+		return nil, fmt.Errorf("failed to process %d SQS message(s): %v", len(failures), sqsBatchItemFailureIDs(failures))
+	}
+
+	return events.SQSEventResponse{BatchItemFailures: failures}, nil
+}
+
+// sqsBatchItemFailureIDs extracts the message IDs from a batch item failure
+// list for the aggregate-error log/error text.
+func sqsBatchItemFailureIDs(failures []events.SQSBatchItemFailure) []string {
+	ids := make([]string, len(failures))
+	for i, f := range failures {
+		ids[i] = f.ItemIdentifier
+	}
+	return ids
+}
+
+// loadReportSQSBatchItemFailures parses SQS_REPORT_BATCH_ITEM_FAILURES at
+// startup (called once from NewApplication). Unset means false: keep the
+// old all-or-nothing batch behavior, which is safe (if noisier) until an
+// operator confirms the SQS event source mapping's function_response_types
+// actually includes "ReportBatchItemFailures" -- a Terraform resource that
+// does not exist anywhere in this repo today (issue #108 follow-up: no
+// aws_sqs_queue / AWS::SQS::Queue / EventSourceMapping / batch_size found
+// across terraform/, cloudformation/, iac/, or arm/). A set but
+// unparseable value is a startup error rather than a silently-ignored
+// typo: flipping this flag is a deploy-time contract with that Terraform
+// setting, and getting it wrong risks silently dropping failed purchase
+// messages instead of redelivering them (see handleLambdaSQSEvent).
+func loadReportSQSBatchItemFailures() (bool, error) {
+	raw := os.Getenv("SQS_REPORT_BATCH_ITEM_FAILURES")
+	if raw == "" {
+		return false, nil
+	}
+	val, err := strconv.ParseBool(raw)
+	if err != nil {
+		return false, fmt.Errorf("invalid SQS_REPORT_BATCH_ITEM_FAILURES %q: must be a boolean: %w", raw, err)
+	}
+	return val, nil
 }
 
 // handleLambdaScheduledEvent processes scheduled/cron events.

internal/server/lambda_test.go

@@ -3,12 +3,15 @@ package server
 import (
 	"context"
 	"encoding/json"
+	"fmt"
+	"os"
 	"strings"
 	"testing"
 
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/api"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/scheduler"
 	"github.com/LeanerCloud/cloud-commitments-platform/internal/testutil"
+	"github.com/aws/aws-lambda-go/events"
 )
 
 func TestDetectLambdaEventType(t *testing.T) {
@@ -212,6 +215,198 @@ func TestHandleLambdaSQSEvent(t *testing.T) {
 	}
 }
 
+// partialFailureRawEvent is a 3-record batch where only "msg-2" fails,
+// shared by the flag-on and flag-off partial-failure tests below.
+const partialFailureRawEvent = `{
+	"Records": [
+		{"messageId": "msg-1", "eventSource": "aws:sqs", "body": "{\"execution_id\": \"msg-1\"}"},
+		{"messageId": "msg-2", "eventSource": "aws:sqs", "body": "{\"execution_id\": \"msg-2\"}"},
+		{"messageId": "msg-3", "eventSource": "aws:sqs", "body": "{\"execution_id\": \"msg-3\"}"}
+	]
+}`
+
+func failOnMsg2Purchase() *testutil.MockPurchaseManager {
+	return &testutil.MockPurchaseManager{
+		ProcessMessageFunc: func(_ context.Context, body string) error {
+			if strings.Contains(body, "msg-2") {
+				return fmt.Errorf("execution not found: msg-2")
+			}
+			return nil
+		},
+	}
+}
+
+// TestHandleLambdaSQSEvent_PartialBatchFailure_FlagOn pins issue #108's
+// fix, gated behind ReportSQSBatchItemFailures=true: a batch where only
+// some records fail must leave the succeeded records ACKed rather than
+// redelivering the whole batch. Pre-fix, handleLambdaSQSEvent collected
+// every failed message ID and returned a single aggregate error for the
+// batch; Lambda's SQS integration treats ANY non-nil handler error as "the
+// entire batch failed", so records 1 and 3 (which HandleSQSMessage already
+// processed successfully) would be redelivered right alongside the
+// genuinely failed record 2 -- and would typically fail a second time
+// since their status already advanced past what the CAS-guarded handlers
+// expect, repeating until the whole batch (not just the poison message)
+// hits the DLQ.
+//
+// Asserted on the returned events.SQSEventResponse.BatchItemFailures set,
+// per the issue's own fix direction, rather than on the error value: the
+// function must return a nil error alongside the partial-failure response,
+// since a non-nil error here is exactly the all-or-nothing behavior being
+// removed.
+func TestHandleLambdaSQSEvent_PartialBatchFailure_FlagOn(t *testing.T) {
+	ctx := testutil.TestContext(t)
+
+	app := &Application{
+		Purchase:  failOnMsg2Purchase(),
+		appConfig: ApplicationConfig{ReportSQSBatchItemFailures: true},
+	}
+
+	result, err := app.handleLambdaSQSEvent(ctx, json.RawMessage(partialFailureRawEvent))
+	testutil.AssertNoError(t, err)
+
+	resp, ok := result.(events.SQSEventResponse)
+	if !ok {
+		t.Fatalf("expected events.SQSEventResponse, got %T", result)
+	}
+	if len(resp.BatchItemFailures) != 1 {
+		t.Fatalf("expected exactly 1 batch item failure, got %d: %+v", len(resp.BatchItemFailures), resp.BatchItemFailures)
+	}
+	if resp.BatchItemFailures[0].ItemIdentifier != "msg-2" {
+		t.Fatalf("expected the failed record's own messageId (msg-2) in BatchItemFailures, got %q",
+			resp.BatchItemFailures[0].ItemIdentifier)
+	}
+}
+
+// TestHandleLambdaSQSEvent_PartialBatchFailure_FlagOff pins the review
+// finding on PR #382: without the SQS event source mapping's
+// function_response_types actually including "ReportBatchItemFailures",
+// AWS ignores the events.SQSEventResponse body entirely, so returning a
+// nil error there means the WHOLE BATCH is acked and deleted -- silently
+// dropping the genuinely failed record instead of redelivering it. With
+// ReportSQSBatchItemFailures left at its default (false), a partial
+// failure must still return the old aggregate error so the whole batch is
+// redelivered (safe, if noisier) until an operator confirms the Terraform
+// side is live and flips the flag.
+func TestHandleLambdaSQSEvent_PartialBatchFailure_FlagOff(t *testing.T) {
+	ctx := testutil.TestContext(t)
+
+	// Zero-value Application.appConfig: ReportSQSBatchItemFailures defaults
+	// to false, matching a real, unconfigured deployment.
+	app := &Application{Purchase: failOnMsg2Purchase()}
+
+	result, err := app.handleLambdaSQSEvent(ctx, json.RawMessage(partialFailureRawEvent))
+	testutil.AssertError(t, err)
+	if result != nil {
+		t.Fatalf("expected a nil result alongside the aggregate error, got %+v", result)
+	}
+	if !strings.Contains(err.Error(), "msg-2") {
+		t.Fatalf("expected the aggregate error to name the failed message, got: %v", err)
+	}
+}
+
+// TestHandleLambdaSQSEvent_AllFailed_ReturnsErrorRegardlessOfFlag pins the
+// review finding that a batch where every record failed must never be
+// acked, flag or no flag -- there is nothing to partially succeed on, so
+// returning the partial-success shape (or a nil error) would ack a batch
+// that accomplished nothing.
+func TestHandleLambdaSQSEvent_AllFailed_ReturnsErrorRegardlessOfFlag(t *testing.T) {
+	for _, flagValue := range []bool{true, false} {
+		t.Run(fmt.Sprintf("flag=%v", flagValue), func(t *testing.T) {
+			ctx := testutil.TestContext(t)
+
+			app := &Application{
+				Purchase: &testutil.MockPurchaseManager{
+					ProcessMessageFunc: func(_ context.Context, _ string) error {
+						return fmt.Errorf("execution not found")
+					},
+				},
+				appConfig: ApplicationConfig{ReportSQSBatchItemFailures: flagValue},
+			}
+
+			rawEvent := `{"Records": [
+				{"messageId": "msg-1", "eventSource": "aws:sqs", "body": "{}"},
+				{"messageId": "msg-2", "eventSource": "aws:sqs", "body": "{}"}
+			]}`
+
+			result, err := app.handleLambdaSQSEvent(ctx, json.RawMessage(rawEvent))
+			testutil.AssertError(t, err)
+			if result != nil {
+				t.Fatalf("expected a nil result alongside the all-failed error, got %+v", result)
+			}
+		})
+	}
+}
+
+// TestLoadReportSQSBatchItemFailures pins the "error on garbage" contract
+// for SQS_REPORT_BATCH_ITEM_FAILURES: unset defaults to false (fail
+// closed), a recognized boolean parses, and anything else is a startup
+// error rather than a silently-ignored typo.
+func TestLoadReportSQSBatchItemFailures(t *testing.T) {
+	t.Cleanup(func() { os.Unsetenv("SQS_REPORT_BATCH_ITEM_FAILURES") })
+
+	tests := []struct {
+		name      string
+		envValue  string
+		unset     bool
+		wantVal   bool
+		wantError bool
+	}{
+		{name: "unset defaults to false", unset: true, wantVal: false},
+		{name: "true", envValue: "true", wantVal: true},
+		{name: "false", envValue: "false", wantVal: false},
+		{name: "garbage errors", envValue: "yes-please", wantError: true},
+		{name: "empty string treated as unset", envValue: "", wantVal: false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.unset {
+				os.Unsetenv("SQS_REPORT_BATCH_ITEM_FAILURES")
+			} else {
+				os.Setenv("SQS_REPORT_BATCH_ITEM_FAILURES", tt.envValue)
+			}
+
+			got, err := loadReportSQSBatchItemFailures()
+			if tt.wantError {
+				testutil.AssertError(t, err)
+				return
+			}
+			testutil.AssertNoError(t, err)
+			if got != tt.wantVal {
+				t.Fatalf("expected %v, got %v", tt.wantVal, got)
+			}
+		})
+	}
+}
+
+// TestHandleLambdaSQSEvent_AllSucceedReturnsEmptyFailures documents the
+// all-succeed shape: a nil error and an explicitly empty (not omitted)
+// BatchItemFailures list, matching the SQS partial-batch-response contract.
+// This holds regardless of ReportSQSBatchItemFailures, since an empty
+// failure list is equivalent to plain success either way.
+func TestHandleLambdaSQSEvent_AllSucceedReturnsEmptyFailures(t *testing.T) {
+	ctx := testutil.TestContext(t)
+
+	mockPurchase := &testutil.MockPurchaseManager{
+		ProcessMessageFunc: func(_ context.Context, _ string) error { return nil },
+	}
+	app := &Application{Purchase: mockPurchase}
+
+	rawEvent := `{"Records": [{"messageId": "msg-1", "eventSource": "aws:sqs", "body": "{}"}]}`
+
+	result, err := app.handleLambdaSQSEvent(ctx, json.RawMessage(rawEvent))
+	testutil.AssertNoError(t, err)
+
+	resp, ok := result.(events.SQSEventResponse)
+	if !ok {
+		t.Fatalf("expected events.SQSEventResponse, got %T", result)
+	}
+	if len(resp.BatchItemFailures) != 0 {
+		t.Fatalf("expected no batch item failures, got %+v", resp.BatchItemFailures)
+	}
+}
+
 // testOwnerToken is a fixed, well-formed collection owner token. It has to be
 // a real UUID because ParseScheduledEvent rejects malformed non-empty tokens
 // at the boundary rather than letting them reach the UUID-typed column.