From 1a64c7ca22aff2aaa16bbd9cc8c16b316e359297 Mon Sep 17 00:00:00 2001 From: Bretton Date: Thu, 20 Aug 2026 04:43:27 +0000 Subject: [PATCH] consume: every dead-letter write sanitizes; remote claims are quoted and bounded Chunk-3 finding 1 from the second-opinion re-review (5 streams). The NUL-byte DLQ fix sanitized AddDeadLetter but not its two siblings writing the same last_error column: MarkRedriveAttempt and RetireDeadLetter passed the string straight through, so a NUL riding in a handler error (reachable: resolver.go interpolated up to 1 KiB of an attacker-controlled well-known response body into a transient error) made the post-redrive UPDATE fail — attempts never incremented, the row could never retire, it was fully re-executed on every pass forever, and the forward-progress guard stalled the entire DLQ drain behind it. All three writers now go through one funnel, execDeadLetterWrite, which sanitizes the error argument immediately before ExecContext — placed inside the store so a fourth writer can't skip it, with a loud typed guard on the convention. sanitizeErrorText gains a head-preserving 4096-byte cap applied before UTF-8 repair so a mid-rune cut becomes a replacement rune. At the source, the three sites echoing a remote-supplied DID claim (both well-known messages and the DNS-TXT message — the TXT value is remote-controlled too) now quote and cap it at 128 bytes. Repro tests pin the store writes with NUL-bearing errors, the bound, the redriver burning exactly MaxRedriveAttempts on a poison row and the drain progressing past it, and the quoted claims on all three resolver paths. Co-Authored-By: Claude Opus 4.8 --- internal/consume/redrive_test.go | 116 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/consume/resolver.go | 38 +++++++++++++++++++++++++++++++++++--- internal/consume/resolver_test.go | 80 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ internal/consume/state_store.go | 81 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------ internal/consume/store_test.go | 97 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 5 file(s) changed, 397 insertion(s)(+), 15 deletion(s)(-) diff --git a/internal/consume/redrive_test.go b/internal/consume/redrive_test.go --- a/internal/consume/redrive_test.go +++ b/internal/consume/redrive_test.go @@ -4,8 +4,11 @@ import ( "context" "database/sql" "fmt" + "strings" + "sync" "testing" "time" + "unicode/utf8" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -214,3 +217,116 @@ return countRows(t, database, "jetstream_dead_letters") == 0 }) assert.Equal(t, backlog, handler.Calls()) } + +// --------------------------------------------------------------------------- +// A row whose failure carries poison bytes must still burn its budget +// --------------------------------------------------------------------------- + +// poisonHandler fails for ONE event time with an error carrying the bytes a +// remote body can plant — a NUL and invalid UTF-8 — and succeeds for every +// other. This is not exotic: resolver.go echoes a stranger's +// /.well-known/atproto-did body into a transient error, and that error is what +// MarkRedriveAttempt writes back into the last_error TEXT column. +type poisonHandler struct { + mu sync.Mutex + poisonTimeUS int64 + calls int + poisonCalls int +} + +func (h *poisonHandler) HandleEvent(_ context.Context, event *JetstreamEvent) error { + h.mu.Lock() + defer h.mu.Unlock() + h.calls++ + if event.TimeUS == h.poisonTimeUS { + h.poisonCalls++ + return fmt.Errorf("verify handle alice.coves.social: well-known claims \x00\xff\xfe, not did:plc:x") + } + return nil +} + +func (h *poisonHandler) counts() (calls, poisonCalls int) { + h.mu.Lock() + defer h.mu.Unlock() + return h.calls, h.poisonCalls +} + +// TestRedriver_PoisonFailureStillBurnsItsBudgetAndUnblocksTheDrain is the +// redriver half of the NUL defense, and the reason it is CRITICAL rather than +// cosmetic. +// +// If MarkRedriveAttempt's UPDATE is rejected by postgres, `attempts` never +// increments. The row can therefore never reach MaxRedriveAttempts, so +// ListRetryable returns it again on the very next pass and the handler — DNS +// lookup, two outbound fetches and all — is fully re-executed FOREVER, with +// nothing in the attempts or backlog counters to show for it. And because it is +// always the OLDEST row, redriveAll's forward-progress guard (redriven+retired +// == 0 → break) parks the whole consumer's drain behind it: the good row +// queued after it is never reached. +// +// The passes are driven directly, one per call, so the assertion is about the +// redriver's pass semantics rather than about wall-clock time. +func TestRedriver_PoisonFailureStillBurnsItsBudgetAndUnblocksTheDrain(t *testing.T) { + database := redriveTestDB(t) + state := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + const poisonTimeUS, healthyTimeUS = 6_000, 6_001 + // The poison row is OLDEST, so it is claimed first every pass and stands in + // front of the healthy one. + require.NoError(t, state.AddDeadLetter(ctx, ConsumerNative, poisonTimeUS, + connFrame(poisonTimeUS, "3lzrev0000001", "aaa"), "first failure", 0)) + require.NoError(t, state.AddDeadLetter(ctx, ConsumerNative, healthyTimeUS, + connFrame(healthyTimeUS, "3lzrev0000002", "bbb"), "postgres blip", 0)) + + handler := &poisonHandler{poisonTimeUS: poisonTimeUS} + // Batch size 1: one row per claim, so each pass is exactly one attempt on + // the oldest retryable row. + redriver := NewDeadLetterRedriver(state, + map[string]EventHandler{ConsumerNative: handler}, + WithRedriveInterval(time.Hour), WithRedriveBatchSize(1)) + + for pass := 1; pass <= MaxRedriveAttempts; pass++ { + redriver.redriveAll(ctx) + + var attempts int + require.NoError(t, database.QueryRowContext(ctx, + `SELECT attempts FROM jetstream_dead_letters WHERE event_time_us = $1`, + int64(poisonTimeUS)).Scan(&attempts)) + require.Equal(t, pass, attempts, + "pass %d must have burnt an attempt on the poison row: a failed last_error "+ + "UPDATE leaves attempts at 0, and a row that cannot count its attempts "+ + "can never retire — it is re-handled on every pass forever", pass) + } + + retryable, err := state.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, retryable, 1, + "the poison row has exhausted its budget and left the retryable set; only the "+ + "healthy row behind it is still queued") + assert.Equal(t, int64(healthyTimeUS), retryable[0].EventTimeUS) + + // And the drain is no longer parked behind it. + redriver.redriveAll(ctx) + + remaining, err := state.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + assert.Empty(t, remaining, + "the row queued BEHIND the poison one is finally redriven — a row that can never "+ + "retire stalls the whole consumer's dead letter drain") + + _, poisonCalls := handler.counts() + assert.Equal(t, MaxRedriveAttempts, poisonCalls, + "the poison row costs exactly its budget of handler executions, not an unbounded "+ + "number: each replay re-runs the full handler, DNS and outbound fetches included") + + var stored string + require.NoError(t, database.QueryRowContext(ctx, + `SELECT last_error FROM jetstream_dead_letters WHERE event_time_us = $1`, + int64(poisonTimeUS)).Scan(&stored)) + assert.False(t, strings.ContainsRune(stored, 0), + "and the stored diagnostic carries no NUL") + assert.True(t, utf8.ValidString(stored), "and is valid UTF-8 for the operator triaging it") + assert.Contains(t, stored, "verify handle alice.coves.social", + "while still saying why the row is STILL failing") +} diff --git a/internal/consume/resolver.go b/internal/consume/resolver.go --- a/internal/consume/resolver.go +++ b/internal/consume/resolver.go @@ -10,6 +10,7 @@ "log/slog" "net" "net/http" "net/url" + "strconv" "strings" "github.com/bluesky-social/indigo/atproto/syntax" @@ -271,7 +272,10 @@ if claimed == did { return nil } // DNS itself named a different DID: authoritative impersonation. - return fmt.Errorf("%w: handle %s DNS claims %s, not %s", ErrPermanentEvent, handle, claimed, did) + // The TXT value is written by whoever runs the handle's zone, so it + // is quoted and capped like any other remote claim. + return fmt.Errorf("%w: handle %s DNS claims %s, not %s", + ErrPermanentEvent, handle, quoteRemoteClaim(claimed), did) } dnsAuthoritative = authoritative } @@ -357,9 +361,37 @@ if !dnsAuthoritative { // DNS was unreachable, so we never learned the owner's authoritative // claim; a mismatched well-known cannot be trusted as impersonation. // Transient, so the redrive re-checks once DNS recovers. - return fmt.Errorf("handle %s well-known claims %s, not %s, but DNS was unreachable", handle, claimed, did) + return fmt.Errorf("handle %s well-known claims %s, not %s, but DNS was unreachable", + handle, quoteRemoteClaim(claimed), did) } - return fmt.Errorf("%w: handle %s claims %s, not %s", ErrPermanentEvent, handle, claimed, did) + return fmt.Errorf("%w: handle %s claims %s, not %s", + ErrPermanentEvent, handle, quoteRemoteClaim(claimed), did) +} + +// maxQuotedClaimBytes bounds how much of a remote claim an error message +// repeats. A did:plc is 32 characters; anything past this is not a claim being +// reported, it is a body being transcribed. +const maxQuotedClaimBytes = 128 + +// quoteRemoteClaim renders a DID claimed by a REMOTE party — a well-known body +// from a host named in a stranger's DID document, or a TXT record from that +// handle's zone — as operator-facing evidence rather than a transcript. +// +// Two properties, both load-bearing rather than cosmetic. It QUOTES +// (strconv.Quote escapes a NUL to the four printable characters `\x00` and +// coerces invalid UTF-8 to escapes), because these errors become a dead +// letter's last_error, a postgres TEXT column that rejects a NUL outright — an +// echoed raw body would let a stranger's server fail the write meant to capture +// its own failure. And it CAPS the length, because the read is bounded at 1 KiB +// but the error is rewritten into that column on every redrive pass. +func quoteRemoteClaim(claimed string) string { + if len(claimed) > maxQuotedClaimBytes { + // Cut on the byte, then let Quote escape whatever partial rune the cut + // left behind; the marker is outside the quotes so it cannot be read as + // part of what the remote actually said. + return strconv.Quote(claimed[:maxQuotedClaimBytes]) + "…" + } + return strconv.Quote(claimed) } // validatePLCDID rejects everything this task cannot resolve, and does it diff --git a/internal/consume/resolver_test.go b/internal/consume/resolver_test.go --- a/internal/consume/resolver_test.go +++ b/internal/consume/resolver_test.go @@ -4,7 +4,9 @@ import ( "context" "net" "net/http" + "strings" "testing" + "unicode/utf8" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -366,3 +368,81 @@ "a 404 well-known is NOT a disavowal: the handle may be mid-setup, or publish "+ "its claim only over DNS. Treating it as permanent would strand a user who "+ "finishes configuring their PDS a minute later") } + +// --------------------------------------------------------------------------- +// The remote claim is EVIDENCE, not a transcript +// --------------------------------------------------------------------------- +// +// verifyWellKnown embeds the body served by https://{handle}/... — a host named +// in a stranger's DID document — into its error. That error is not just read by +// an operator: it becomes a dead letter's last_error, a postgres TEXT column +// that rejects NUL outright. Echoing the body verbatim hands an attacker a +// string that can fail the very write meant to capture the failure. The claim +// must be quoted (control bytes become printable escapes) and length-capped +// before it goes anywhere near an error message. + +// poisonWellKnownBody is what a hostile — or merely broken — server can return: +// NULs, invalid UTF-8, and far more bytes than a DID could ever need. +var poisonWellKnownBody = "did:plc:\x00\x00evil\xff\xfe" + strings.Repeat("A", 4096) + +func TestHandleResolver_PoisonWellKnownClaimIsQuotedAndBounded(t *testing.T) { + assertSafeEvidence := func(t *testing.T, err error) { + t.Helper() + message := err.Error() + assert.False(t, strings.ContainsRune(message, 0), + "the error must carry no NUL: it lands in the last_error TEXT column, and a "+ + "NUL there fails the dead-letter write — the fallback that must never "+ + "itself fail") + assert.True(t, utf8.ValidString(message), + "and must be valid UTF-8, so the operator triaging the DLQ can read it") + assert.LessOrEqual(t, len(message), 512, + "and must be BOUNDED: the well-known read is capped at 1 KiB, but none of "+ + "that belongs in an error message verbatim — the claim is evidence, not "+ + "a transcript") + assert.Contains(t, message, resolveHandle, + "while still naming the handle whose claim disagreed") + } + + t.Run("transient, DNS unreachable", func(t *testing.T) { + fake := newFakeIdentity(t) + fake.claimOneWay(resolveDID, resolveHandle) + fake.txtFails(resolveHandle, &net.DNSError{ + Err: "server misbehaving", Name: atprotoTXTPrefix + resolveHandle, IsTemporary: true}) + fake.wellKnownReturns(resolveHandle, poisonWellKnownBody) + + _, err := fake.resolver(t).ResolveDIDHandle(context.Background(), resolveDID) + + require.Error(t, err) + assert.NotErrorIs(t, err, ErrPermanentEvent, + "DNS was unreachable, so this stays redrivable — which is exactly why the "+ + "error string matters: it will be written to last_error again on every "+ + "redrive pass") + assertSafeEvidence(t, err) + }) + + t.Run("permanent, NXDOMAIN", func(t *testing.T) { + fake := newFakeIdentity(t) + fake.claimOneWay(resolveDID, resolveHandle) // no TXT registered → NXDOMAIN + fake.wellKnownReturns(resolveHandle, poisonWellKnownBody) + + _, err := fake.resolver(t).ResolveDIDHandle(context.Background(), resolveDID) + + require.Error(t, err) + assert.ErrorIs(t, err, ErrPermanentEvent) + assertSafeEvidence(t, err) + }) + + t.Run("permanent, contradicting DNS TXT", func(t *testing.T) { + // DNS TXT is remote-supplied too: whoever runs the handle's zone writes + // those bytes, and the resolver echoes them the same way. + fake := newFakeIdentity(t) + fake.claimOneWay(resolveDID, resolveHandle) + fake.txtRecords(resolveHandle, "did="+poisonWellKnownBody) + + _, err := fake.resolver(t).ResolveDIDHandle(context.Background(), resolveDID) + + require.Error(t, err) + assert.ErrorIs(t, err, ErrPermanentEvent) + assertSafeEvidence(t, err) + }) +} diff --git a/internal/consume/state_store.go b/internal/consume/state_store.go --- a/internal/consume/state_store.go +++ b/internal/consume/state_store.go @@ -86,20 +86,17 @@ // already-captured event is a no-op success (the dedup index absorbs it) so // the cursor may advance past a poison frame. func (s *PostgresStateStore) AddDeadLetter(ctx context.Context, consumerName string, eventTimeUS int64, eventData []byte, handleErr string, redriveAttempts int) error { // event_data is written as raw bytes so byte-corrupt frames are capturable. - // last_error is TEXT, so it is SANITIZED first: a malformed frame's error - // can carry the very bytes that made it malformed (a NUL, invalid UTF-8), - // and postgres TEXT rejects a NUL outright. An unsanitized error would fail - // the dead-letter write, the connector would tear the connection down - // without advancing the cursor, and the same poison frame would replay - // forever — this is the fallback that must never itself fail. + // last_error is TEXT, so the write goes through execDeadLetterWrite, which + // sanitizes it at the last point before the SQL — see the comment there for + // why every writer of this column must. // // redriveAttempts seeds the budget: 0 for transient failures, and // MaxRedriveAttempts for permanent ones, which are kept for forensics only. - _, err := s.db.ExecContext(ctx, ` - INSERT INTO jetstream_dead_letters (consumer_name, event_time_us, event_data, last_error, attempts) + err := s.execDeadLetterWrite(ctx, ` + INSERT INTO jetstream_dead_letters (consumer_name, event_time_us, event_data, attempts, last_error) VALUES ($1, $2, $3, $4, $5) ON CONFLICT DO NOTHING`, - consumerName, eventTimeUS, eventData, sanitizeErrorText(handleErr), redriveAttempts, + consumerName, eventTimeUS, eventData, redriveAttempts, handleErr, ) if err != nil { return fmt.Errorf("add dead letter for %s: %w", consumerName, err) @@ -107,17 +104,69 @@ } return nil } +// execDeadLetterWrite runs a statement against jetstream_dead_letters whose +// LAST positional argument is the last_error value, and sanitizes that value +// here — one funnel, at the last point before the SQL. +// +// The funnel exists because sanitizing at the CALL SITES did not hold. Three +// statements write this column and only the INSERT was scrubbed; the two +// UPDATEs on the redrive path passed the string straight through. A NUL there +// does not merely lose a diagnostic — it fails the UPDATE, so `attempts` never +// increments, the row can never reach MaxRedriveAttempts, and it is re-selected +// and fully re-handled on every redrive pass forever while redriveAll's +// forward-progress guard parks the rest of the drain behind it. A fourth writer +// reaching for s.db.ExecContext directly would reopen exactly that hole, so +// last_error is written through here and nowhere else. +func (s *PostgresStateStore) execDeadLetterWrite(ctx context.Context, query string, args ...any) error { + last := len(args) - 1 + if last < 0 { + return fmt.Errorf("dead letter write: no arguments, so no last_error to sanitize") + } + errorText, ok := args[last].(string) + if !ok { + // A mistake in this file, not a runtime condition: the convention the + // funnel enforces is "the last argument is last_error". + return fmt.Errorf("dead letter write: last argument must be the last_error string, got %T", args[last]) + } + args[last] = sanitizeErrorText(errorText) + _, err := s.db.ExecContext(ctx, query, args...) + return err +} + +// Bounds on what one dead letter's diagnostic may cost. last_error is +// operator-facing EVIDENCE, not a transcript: the errors that reach it can +// quote a body fetched from a host a stranger named, and a redriven row +// rewrites the column once per pass. +const ( + maxLastErrorBytes = 4096 + lastErrorTruncationMarker = " …(truncated)" +) + // sanitizeErrorText makes an error string safe for a postgres TEXT column // while keeping it readable. NUL bytes are stripped (postgres rejects them // outright) and any remaining invalid UTF-8 is coerced to the replacement // rune, so an operator can still read the diagnostic out of the DLQ. Scrubbing // the bad bytes, not discarding the message. +// +// The result is also capped, keeping the HEAD: what identifies a failure is the +// front of its message, and the tail is where an echoed remote body would sit. func sanitizeErrorText(s string) string { if s == "" { return s } s = strings.ReplaceAll(s, "\x00", "") - return strings.ToValidUTF8(s, "�") + truncated := len(s) > maxLastErrorBytes + if truncated { + // Cut first, validate second: the cut can land mid-rune, and + // ToValidUTF8 then turns that trailing fragment into the replacement + // rune rather than leaving bytes postgres would reject. + s = s[:maxLastErrorBytes] + } + s = strings.ToValidUTF8(s, "�") + if truncated { + s += lastErrorTruncationMarker + } + return s } // ListRetryable returns up to limit dead letters for the consumer that have @@ -174,7 +223,11 @@ } // MarkRedriveAttempt increments the attempt counter after a failed redrive. func (s *PostgresStateStore) MarkRedriveAttempt(ctx context.Context, id int64, handleErr string) error { - _, err := s.db.ExecContext(ctx, ` + // The counter and the error ride ONE statement, through the sanitizing + // funnel: this write is what makes a row's budget finite, so a diagnostic + // the column cannot hold must never be able to take the increment down + // with it. + err := s.execDeadLetterWrite(ctx, ` UPDATE jetstream_dead_letters SET attempts = attempts + 1, last_error = $2, updated_at = now() WHERE id = $1`, @@ -193,7 +246,11 @@ // never succeed (an unparseable frame, a lexicon rejection) must stop // costing redrive passes after ONE call, not after the whole budget is // burnt down one attempt at a time. The row STAYS — retiring is about the // redriver, not about forgetting. - _, err := s.db.ExecContext(ctx, ` + // + // Through the same funnel, and for the same reason: the reason string + // carries the parse error of a frame that would not parse, which is exactly + // the frame whose bytes the TEXT column cannot hold. + err := s.execDeadLetterWrite(ctx, ` UPDATE jetstream_dead_letters SET attempts = GREATEST(attempts, $2), last_error = $3, updated_at = now() WHERE id = $1`, diff --git a/internal/consume/store_test.go b/internal/consume/store_test.go --- a/internal/consume/store_test.go +++ b/internal/consume/store_test.go @@ -340,3 +340,100 @@ // the TEXT error column is sanitized. assert.Equal(t, []byte(`{"kind":"commit"}`), dead[0].EventData, "the raw frame is preserved verbatim in the BYTEA column for a faithful redrive") } + +// --------------------------------------------------------------------------- +// The OTHER two last_error writers +// --------------------------------------------------------------------------- +// +// AddDeadLetter is only the first of three statements that write the same +// last_error TEXT column. MarkRedriveAttempt and RetireDeadLetter write it too, +// and a poison error reaches them by the most ordinary route there is: the +// handler that first failed fails again on redrive with the same bytes. If +// either UPDATE passes the string through, postgres rejects the write, the +// attempt counter never increments, and the row is re-selected and fully +// re-executed on every redrive pass FOREVER — invisible to the attempts and +// backlog counters that are supposed to show exactly this. + +// poisonErrorText is what a handler failing on a malformed frame — or on a +// remote body it echoed — hands the DLQ: a NUL postgres TEXT rejects outright, +// plus invalid UTF-8. +const poisonErrorText = "still failing: rkey \x00\x00 is invalid \xff\xfe" + +func TestDeadLetters_MarkRedriveAttemptSanitizesPoisonError(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 1, []byte(`{"time_us":1}`), "first", 0)) + listed, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, listed, 1) + + require.NoError(t, store.MarkRedriveAttempt(ctx, listed[0].ID, poisonErrorText), + "a NUL in the redrive failure must NOT fail the UPDATE: the attempt counter is "+ + "what retires a row, so a failed mark means the row can never reach "+ + "MaxRedriveAttempts and is re-handled on every pass forever") + + listed, err = store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, listed, 1) + assert.Equal(t, 1, listed[0].Attempts, + "the attempt was actually burnt — the row is one pass closer to retirement") + assert.False(t, strings.ContainsRune(listed[0].LastError, 0), + "the stored last_error carries no NUL") + assert.True(t, utf8.ValidString(listed[0].LastError), + "and is valid UTF-8, so an operator can read it out of the queue") + assert.Contains(t, listed[0].LastError, "rkey", + "while keeping the readable part: sanitizing scrubs bad bytes, it does not "+ + "discard the diagnostic") +} + +func TestDeadLetters_RetireSanitizesPoisonReason(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 1, []byte(`not json`), "parse", 0)) + listed, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, listed, 1) + + // The retirement reason embeds the parse error, which for a byte-corrupt + // frame quotes the offending bytes. + require.NoError(t, store.RetireDeadLetter(ctx, listed[0].ID, "unparseable event: "+poisonErrorText), + "retiring is the escape hatch for a row that can never succeed — it must not "+ + "itself be defeated by the bytes that made the row unsucceedable") + + retryable, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + assert.Empty(t, retryable, "the row is exhausted after exactly one call") + + var stored string + require.NoError(t, database.QueryRowContext(ctx, + `SELECT last_error FROM jetstream_dead_letters`).Scan(&stored)) + assert.False(t, strings.ContainsRune(stored, 0), "the retirement reason carries no NUL") + assert.True(t, utf8.ValidString(stored), "and is valid UTF-8") + assert.Contains(t, stored, "unparseable event", + "the reason still explains why the row was retired") +} + +// TestDeadLetters_LastErrorIsBounded keeps one unbounded remote string from +// becoming an unbounded row. last_error is operator-facing EVIDENCE, not a +// transcript of whatever a stranger's server returned. +func TestDeadLetters_LastErrorIsBounded(t *testing.T) { + database := consumeStateTestDB(t) + store := NewPostgresStateStore(database, CursorSchemaVersion) + ctx := context.Background() + + huge := "verify handle alice.example: " + strings.Repeat("A", 200_000) + require.NoError(t, store.AddDeadLetter(ctx, ConsumerNative, 1, []byte(`{"time_us":1}`), huge, 0)) + listed, err := store.ListRetryable(ctx, ConsumerNative, MaxRedriveAttempts, 10) + require.NoError(t, err) + require.Len(t, listed, 1) + + assert.LessOrEqual(t, len(listed[0].LastError), maxLastErrorBytes+len(lastErrorTruncationMarker), + "a dead letter's error is capped: an attacker-supplied body must not be able to "+ + "write an arbitrarily large row on every redrive pass") + assert.Contains(t, listed[0].LastError, "verify handle alice.example", + "the HEAD of the message is what identifies the failure, so that is what survives") +} -- tangled.sh