From 3521199da70d88079212a42a65ab97727c68314b Mon Sep 17 00:00:00 2001 From: Audrine Whrite Date: Fri, 16 Oct 2026 08:57:49 +0300 Subject: [PATCH 1/4] test: inject dead-letter write failures and verify rollback --- internal/queue/postgres_integration_test.go | 91 +++++++++++++++++++++ 1 file changed, 91 insertions(+) diff --git a/internal/queue/postgres_integration_test.go b/internal/queue/postgres_integration_test.go index dff7864..ebaec8c 100644 --- a/internal/queue/postgres_integration_test.go +++ b/internal/queue/postgres_integration_test.go @@ -351,3 +351,94 @@ func TestPostgresQueueDeadLettersExpiredLeaseAfterMaximumAttempts(t *testing.T) t.Fatalf("failure reason = %q", letters[0].FailureReason) } } + + +func installDeadLetterFailureTrigger(t testing.TB, pool *pgxpool.Pool) { + t.Helper() + _, err := pool.Exec(context.Background(), ` + CREATE FUNCTION reject_dead_letter_insert() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + RAISE EXCEPTION 'injected dead-letter storage failure'; + END; + $$; + CREATE TRIGGER reject_dead_letter_insert + BEFORE INSERT ON dead_letter_jobs + FOR EACH ROW EXECUTE FUNCTION reject_dead_letter_insert(); + `) + if err != nil { + t.Fatalf("install dead-letter failure injection: %v", err) + } +} + +func TestPostgresQueueRetryRollsBackWhenDeadLetterWriteFails(t *testing.T) { + pool := integrationPool(t) + queueName := fmt.Sprintf("retry-dead-letter-failure-%d", time.Now().UnixNano()) + id := insertIntegrationJob(t, pool, queueName, 3, 3, "running", time.Now().Add(time.Minute)) + installDeadLetterFailureTrigger(t, pool) + + q := NewPostgresQueue(pool) + if err := q.Retry(context.Background(), id, "old-worker", time.Second); err == nil { + t.Fatal("expected injected dead-letter insert failure") + } + + var status string + var lockedBy *string + if err := pool.QueryRow( + context.Background(), + "SELECT status, locked_by FROM jobs WHERE id=$1", + id, + ).Scan(&status, &lockedBy); err != nil { + t.Fatalf("read job after injected failure: %v", err) + } + if status != "running" { + t.Fatalf("status = %q after rollback, expected running", status) + } + if lockedBy == nil || *lockedBy != "old-worker" { + t.Fatalf("locked_by = %v after rollback, expected old-worker", lockedBy) + } + + letters, err := q.ListDeadLetters(context.Background(), queueName, 10) + if err != nil { + t.Fatalf("list dead letters after rollback: %v", err) + } + if len(letters) != 0 { + t.Fatalf("dead letters after rollback = %d, expected 0", len(letters)) + } +} + +func TestPostgresQueueClaimRollsBackExpiredLeaseRecoveryWhenDeadLetterWriteFails(t *testing.T) { + pool := integrationPool(t) + queueName := fmt.Sprintf("claim-dead-letter-failure-%d", time.Now().UnixNano()) + id := insertIntegrationJob(t, pool, queueName, 3, 3, "running", time.Now().Add(-time.Minute)) + installDeadLetterFailureTrigger(t, pool) + + q := NewPostgresQueue(pool) + if _, err := q.ClaimNext(context.Background(), queueName, "replacement-worker", time.Minute); err == nil { + t.Fatal("expected injected dead-letter insert failure") + } + + var status string + var lockedBy *string + if err := pool.QueryRow( + context.Background(), + "SELECT status, locked_by FROM jobs WHERE id=$1", + id, + ).Scan(&status, &lockedBy); err != nil { + t.Fatalf("read expired job after injected failure: %v", err) + } + if status != "running" { + t.Fatalf("status = %q after rollback, expected running", status) + } + if lockedBy == nil || *lockedBy != "old-worker" { + t.Fatalf("locked_by = %v after rollback, expected old-worker", lockedBy) + } + + letters, err := q.ListDeadLetters(context.Background(), queueName, 10) + if err != nil { + t.Fatalf("list dead letters after recovery rollback: %v", err) + } + if len(letters) != 0 { + t.Fatalf("dead letters after recovery rollback = %d, expected 0", len(letters)) + } +} From a39ac463b8408d5c942ff14784663c1a21d7ee17 Mon Sep 17 00:00:00 2001 From: Audrine Whrite Date: Fri, 16 Oct 2026 08:57:58 +0300 Subject: [PATCH 2/4] test: create failure-injection trigger with separate SQL statements --- internal/queue/postgres_integration_test.go | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/internal/queue/postgres_integration_test.go b/internal/queue/postgres_integration_test.go index ebaec8c..a2cd6a9 100644 --- a/internal/queue/postgres_integration_test.go +++ b/internal/queue/postgres_integration_test.go @@ -357,17 +357,22 @@ func installDeadLetterFailureTrigger(t testing.TB, pool *pgxpool.Pool) { t.Helper() _, err := pool.Exec(context.Background(), ` CREATE FUNCTION reject_dead_letter_insert() RETURNS trigger - LANGUAGE plpgsql AS $$ + LANGUAGE plpgsql AS $ BEGIN RAISE EXCEPTION 'injected dead-letter storage failure'; END; - $$; + $ + `) + if err != nil { + t.Fatalf("create dead-letter failure function: %v", err) + } + _, err = pool.Exec(context.Background(), ` CREATE TRIGGER reject_dead_letter_insert BEFORE INSERT ON dead_letter_jobs - FOR EACH ROW EXECUTE FUNCTION reject_dead_letter_insert(); + FOR EACH ROW EXECUTE FUNCTION reject_dead_letter_insert() `) if err != nil { - t.Fatalf("install dead-letter failure injection: %v", err) + t.Fatalf("create dead-letter failure trigger: %v", err) } } From 76dee841bbc74d1807fc0aeef0ad3ccdbbce2f79 Mon Sep 17 00:00:00 2001 From: Audrine Whrite Date: Fri, 16 Oct 2026 08:58:06 +0300 Subject: [PATCH 3/4] docs: record dead-letter failure-injection coverage --- README.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/README.md b/README.md index ecf2b87..948a4e9 100644 --- a/README.md +++ b/README.md @@ -23,7 +23,7 @@ Implemented: Remaining in Milestone 4: - Explicit cancellation semantics -- Failure-injection tests for database outages and worker restarts +- End-to-end process-restart testing with a running worker and a deliberately interrupted handler ## Architecture From 5a85ff4f8f1d3764976ca6f474df5e052db1fe4f Mon Sep 17 00:00:00 2001 From: Audrine Whrite Date: Fri, 16 Oct 2026 08:59:34 +0300 Subject: [PATCH 4/4] test: use quoted PL/pgSQL body for failure injection --- internal/queue/postgres_integration_test.go | 15 ++------------- 1 file changed, 2 insertions(+), 13 deletions(-) diff --git a/internal/queue/postgres_integration_test.go b/internal/queue/postgres_integration_test.go index a2cd6a9..67e65c0 100644 --- a/internal/queue/postgres_integration_test.go +++ b/internal/queue/postgres_integration_test.go @@ -355,22 +355,11 @@ func TestPostgresQueueDeadLettersExpiredLeaseAfterMaximumAttempts(t *testing.T) func installDeadLetterFailureTrigger(t testing.TB, pool *pgxpool.Pool) { t.Helper() - _, err := pool.Exec(context.Background(), ` - CREATE FUNCTION reject_dead_letter_insert() RETURNS trigger - LANGUAGE plpgsql AS $ - BEGIN - RAISE EXCEPTION 'injected dead-letter storage failure'; - END; - $ - `) + _, err := pool.Exec(context.Background(), "CREATE FUNCTION reject_dead_letter_insert() RETURNS trigger LANGUAGE plpgsql AS 'BEGIN RAISE EXCEPTION ''injected dead-letter storage failure''; RETURN NEW; END;'") if err != nil { t.Fatalf("create dead-letter failure function: %v", err) } - _, err = pool.Exec(context.Background(), ` - CREATE TRIGGER reject_dead_letter_insert - BEFORE INSERT ON dead_letter_jobs - FOR EACH ROW EXECUTE FUNCTION reject_dead_letter_insert() - `) + _, err = pool.Exec(context.Background(), `CREATE TRIGGER reject_dead_letter_insert BEFORE INSERT ON dead_letter_jobs FOR EACH ROW EXECUTE FUNCTION reject_dead_letter_insert()`) if err != nil { t.Fatalf("create dead-letter failure trigger: %v", err) }