diff --git a/internal/server/app.go b/internal/server/app.go index 1d33d8df4..7025ea88e 100644 --- a/internal/server/app.go +++ b/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 diff --git a/internal/server/lambda.go b/internal/server/lambda.go index 348d349b1..98c22c807 100644 --- a/internal/server/lambda.go +++ b/internal/server/lambda.go @@ -6,6 +6,8 @@ import ( "encoding/json" "fmt" "log" + "os" + "strconv" "strings" "github.com/aws/aws-lambda-go/events" @@ -223,6 +225,38 @@ 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 { @@ -230,21 +264,66 @@ func (app *Application) handleLambdaSQSEvent(ctx context.Context, rawEvent json. 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. diff --git a/internal/server/lambda_test.go b/internal/server/lambda_test.go index f9d1fa002..da2824fd8 100644 --- a/internal/server/lambda_test.go +++ b/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.