diff --git a/AGENTS.md b/AGENTS.md index b92244fdd..4f972639f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -408,6 +408,19 @@ manually on every API change even though the template won't remind you. bundled drive-by cleanup. CI must be green. - **Coverage floors** only move up (see Testing strategy). - **Postgres**: local dev runs on port **5433** (not 5432) via docker compose. +- **Row locks in multi-statement transactions**: an `INSERT` holds + `FOR KEY SHARE` on every row it references by foreign key until commit, and + `FOR UPDATE` conflicts with that. So a `SELECT … FOR UPDATE` on a parent row + taken *after* inserting a child in the same transaction deadlocks against a + concurrent insert for the same parent (v1.9.0: the accept transaction + inserted the message, then the gate locked the agent `FOR UPDATE`; two + parallel sends → SQLSTATE 40P01). Lock the parent `FOR NO KEY UPDATE` + (excludes updates, deletes and other lockers, coexists with key shares), or + lock it before the insert. The accept transaction's full lock order is in + `docs/design/async-message-pipeline.md`; any new lock on that path must be + checked against it, and any parallel-write path needs a concurrency test + (see `TestPrepareDoesNotDeadlockAgainstConcurrentInsert` for the + deterministic two-transaction shape). - The Mailpit service in `docker-compose.yaml` is local-dev only — production deployments must drop it and point `E2A_OUTBOUND_SMTP_*` at a real relay. diff --git a/docs/design/async-message-pipeline.md b/docs/design/async-message-pipeline.md index 89c4f08f0..a2547fd9e 100644 --- a/docs/design/async-message-pipeline.md +++ b/docs/design/async-message-pipeline.md @@ -345,6 +345,22 @@ a paused account's message job is also left unstamped for the worker's hold path. The workers resolve legacy jobs themselves, so the command is a convenience for a clean cutover, not a prerequisite. +**Lock order of the accept transaction.** Every lock the accept path takes, +in order, so the next change can check itself against it: the message +insert takes `FOR KEY SHARE` on the agent row (foreign key) and an +exclusive lock on the account's `account_usage` row (storage trigger); the +gate's Prepare then takes `FOR NO KEY UPDATE` on the agent, `FOR UPDATE` on +the message, and the `account_sending_controls` upsert (which holds `KEY +SHARE` on the user); then the operation insert, the River job insert, and +the message's own stamp. The gate's agent lock is `NO KEY UPDATE` and must +stay that way: `FOR UPDATE` conflicts with the key share every concurrent +insert for the same agent already holds, and v1.9.0 deadlocked two parallel +sends exactly there. The rule generalizes: a `FOR UPDATE` taken after an +`INSERT` that references the locked row by foreign key, in the same +transaction, deadlocks under concurrency. An approval and a reply hold a +message row before the gate runs, so "agent before message" is a property +of Prepare itself, not of every caller. + Two consequences worth knowing. Notification and feedback mail now cross the same submitter as customer mail, so it carries `X-SES-CONFIGURATION-SET` and SES publishes delivery feedback for it; none of it correlates to a diff --git a/internal/e2e/sending_concurrency_e2e_test.go b/internal/e2e/sending_concurrency_e2e_test.go new file mode 100644 index 000000000..00ea7cdf8 --- /dev/null +++ b/internal/e2e/sending_concurrency_e2e_test.go @@ -0,0 +1,71 @@ +//go:build integration + +package e2e_test + +import ( + "fmt" + "io" + "net/http" + "strings" + "sync" + "testing" + + "github.com/tokencanopy/e2a/internal/testutil" +) + +// TestParallelSendsFromOneAgentAllAccept: eight concurrent sends from one +// agent must all be accepted. Each accept transaction inserts the message and +// then prepares its sending operation under the gate; the v1.9.0 staging +// conformance gate caught the two steps deadlocking against each other +// (SQLSTATE 40P01 → 500) when the gate locked the agent FOR UPDATE. +func TestParallelSendsFromOneAgentAllAccept(t *testing.T) { + pool := testutil.TestDB(t) + ts := testutil.TestServer(t, pool, testutil.WithOutboundSMTP("127.0.0.1", 1025, "test.e2a.dev")) + _, key, agent := setupDomainAndAgent(t, ts, "agent@conc.example.com", "conc.example.com", "", "") + + const n = 8 + type result struct { + status int + body []byte + err error + } + results := make([]result, n) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + // No t.Fatal from a worker goroutine: collect and assert after Wait. + body := fmt.Sprintf(`{"to":["alice@example.com"],"subject":"parallel %d","text":"parallel send #%d"}`, i, i) + req, err := http.NewRequest("POST", sendURL(ts.HTTPServer.URL, agent.EmailAddress()), strings.NewReader(body)) + if err != nil { + results[i].err = err + return + } + req.Header.Set("Authorization", "Bearer "+key.PlaintextKey) + req.Header.Set("Content-Type", "application/json") + resp, err := http.DefaultClient.Do(req) + if err != nil { + results[i].err = err + return + } + defer resp.Body.Close() + out, _ := io.ReadAll(resp.Body) + results[i] = result{status: resp.StatusCode, body: out} + }(i) + } + wg.Wait() + + for i, r := range results { + if r.err != nil { + t.Errorf("send %d: %v", i, r.err) + continue + } + if r.status != 200 && r.status != 202 { + t.Errorf("send %d: status=%d body=%s", i, r.status, r.body) + } + if !strings.Contains(string(r.body), `"message_id":"msg_`) { + t.Errorf("send %d: no message id in %s", i, r.body) + } + } +} diff --git a/internal/sendingpolicy/operations.go b/internal/sendingpolicy/operations.go index e0781a498..d2e2cb52a 100644 --- a/internal/sendingpolicy/operations.go +++ b/internal/sendingpolicy/operations.go @@ -169,9 +169,23 @@ func (m *Module) PrepareExternalTx(ctx context.Context, tx pgx.Tx, messageID str return "", OperationRef{}, ErrSourceUnavailable } - // Agent before message, matching migration 113 and the irreversible - // deletion path: deletion locks an agent and then its messages, so taking - // them in the other order here would deadlock against a concurrent purge. + // Within this function: agent before message, matching migration 113 and + // the irreversible deletion path (which locks an agent and then its + // messages). The enclosing accept transaction may already hold a message + // row — an approval updates the held message first, a reply locks its + // parent — so the order is a property of this function, not a guarantee + // about every caller. + // + // FOR NO KEY UPDATE, not FOR UPDATE. This runs inside the accept + // transaction AFTER the message insert, and every concurrent insert of a + // message for the same agent holds a FOR KEY SHARE lock on the agent row + // through its foreign key. FOR UPDATE conflicts with KEY SHARE, so two + // parallel sends deadlocked: each held its own insert's share lock plus + // the account_usage row its storage trigger took, and each waited for + // the other's agent lock (seen as SQLSTATE 40P01 on staging, v1.9.0). + // NO KEY UPDATE still serializes gate callers against each other and + // against any update or delete of the agent, which is all the ordering + // this needs, and it does not conflict with a foreign-key share. var agentID string err := tx.QueryRow(ctx, `SELECT agent_id FROM messages WHERE id = $1 AND direction = 'outbound'`, messageID, @@ -185,7 +199,7 @@ func (m *Module) PrepareExternalTx(ctx context.Context, tx pgx.Tx, messageID str var userID string err = tx.QueryRow(ctx, - `SELECT user_id FROM agent_identities WHERE id = $1 FOR UPDATE`, agentID, + `SELECT user_id FROM agent_identities WHERE id = $1 FOR NO KEY UPDATE`, agentID, ).Scan(&userID) if errors.Is(err, pgx.ErrNoRows) { return "", OperationRef{}, ErrSourceUnavailable @@ -295,7 +309,7 @@ func (m *Module) PrepareNotificationTx(ctx context.Context, tx pgx.Tx, ref Notif // names another episode is detectably stale. var warnedAt, disabledAt *time.Time err = tx.QueryRow(ctx, - `SELECT user_id, warn_notified_at, auto_disabled_at FROM webhooks WHERE id = $1 FOR UPDATE`, ref.id, + `SELECT user_id, warn_notified_at, auto_disabled_at FROM webhooks WHERE id = $1 FOR NO KEY UPDATE`, ref.id, ).Scan(&userID, &warnedAt, &disabledAt) if errors.Is(err, pgx.ErrNoRows) { err = ErrSourceUnavailable @@ -358,7 +372,10 @@ func lockHITLSourceOwner(ctx context.Context, tx pgx.Tx, messageID string) (stri var userID string err = tx.QueryRow(ctx, - `SELECT user_id FROM agent_identities WHERE id = $1 FOR UPDATE`, agentID, + // NO KEY UPDATE for the same reason as PrepareExternalTx: the hold's + // accept transaction inserted the message first, and concurrent + // inserts hold the agent row FOR KEY SHARE. + `SELECT user_id FROM agent_identities WHERE id = $1 FOR NO KEY UPDATE`, agentID, ).Scan(&userID) if errors.Is(err, pgx.ErrNoRows) { return "", ErrSourceUnavailable diff --git a/internal/sendingpolicy/store_integration_test.go b/internal/sendingpolicy/store_integration_test.go index d2a53f432..3dc8aaeb3 100644 --- a/internal/sendingpolicy/store_integration_test.go +++ b/internal/sendingpolicy/store_integration_test.go @@ -1222,3 +1222,124 @@ func (f *fixture) tryTx(fn func(tx pgx.Tx) error) error { } return tx.Commit(f.ctx) } + +// TestPrepareDoesNotDeadlockAgainstConcurrentInsert reproduces the v1.9.0 +// staging failure: two accept transactions for the same agent each insert +// their message (taking a FOR KEY SHARE lock on the agent row through the +// foreign key, and the account_usage row through the storage trigger) and +// then prepare their operation. With the agent locked FOR UPDATE the second +// insert waits on the first's account_usage row while the first's prepare +// waits on the second's key share — SQLSTATE 40P01. The gate's NO KEY UPDATE +// lock lets the first prepare proceed. +// +// The interleaving is forced, not timed: B reports its backend pid and A +// waits until pg_stat_activity shows that backend blocked on a lock before +// preparing, so the test cannot pass vacuously by A finishing first. +func TestPrepareDoesNotDeadlockAgainstConcurrentInsert(t *testing.T) { + cases := []struct { + name string + status string + prepare func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error + }{ + {"external", "sent", func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error { + _, _, err := g.PrepareExternalTx(context.Background(), tx, messageID) + return err + }}, + {"hitl notification", "pending_review", func(g sendingpolicy.Gate, tx pgx.Tx, messageID string) error { + _, err := g.PrepareNotificationTx(context.Background(), tx, sendingpolicy.NewHITLNotificationRef(messageID)) + return err + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + f := newFixture(t) + g := f.gate(sendingpolicy.DisabledPolicy()) + user := f.user("standard") + agent := f.agent(user) + insert := func(tx pgx.Tx, id string) error { + _, err := tx.Exec(f.ctx, + `INSERT INTO messages (id, agent_id, direction, to_recipients, sent_as, status, body_text) + VALUES ($1, $2, 'outbound', ARRAY['rcpt@example.test'], 'relay', $3, 'x')`, + id, agent, tc.status) + return err + } + + txA, err := f.pool.Begin(f.ctx) + if err != nil { + t.Fatal(err) + } + defer func() { _ = txA.Rollback(f.ctx) }() + if err := insert(txA, "msg_lock_a"); err != nil { + t.Fatalf("insert A: %v", err) + } + + // B inserts concurrently: it takes its key share on the agent, then + // its storage-trigger upsert blocks on A's account_usage row. + bPID := make(chan int, 1) + bDone := make(chan error, 1) + go func() { + txB, err := f.pool.Begin(f.ctx) + if err != nil { + bDone <- err + return + } + defer func() { _ = txB.Rollback(f.ctx) }() + var pid int + if err := txB.QueryRow(f.ctx, `SELECT pg_backend_pid()`).Scan(&pid); err != nil { + bDone <- err + return + } + bPID <- pid + if err := insert(txB, "msg_lock_b"); err != nil { + bDone <- fmt.Errorf("insert B: %w", err) + return + } + if err := tc.prepare(g, txB, "msg_lock_b"); err != nil { + bDone <- fmt.Errorf("prepare B: %w", err) + return + } + bDone <- txB.Commit(f.ctx) + }() + + var pid int + select { + case pid = <-bPID: + case err := <-bDone: + t.Fatalf("B ended before starting: %v", err) + case <-time.After(10 * time.Second): + t.Fatal("B never reported its backend") + } + deadline := time.Now().Add(10 * time.Second) + for { + var blocked bool + if err := f.pool.QueryRow(f.ctx, + `SELECT EXISTS (SELECT 1 FROM pg_stat_activity WHERE pid = $1 AND wait_event_type = 'Lock')`, pid, + ).Scan(&blocked); err != nil { + t.Fatal(err) + } + if blocked { + break + } + if time.Now().After(deadline) { + t.Fatal("B never blocked on A's insert; the interleaving this test needs did not happen") + } + time.Sleep(20 * time.Millisecond) + } + + if err := tc.prepare(g, txA, "msg_lock_a"); err != nil { + t.Fatalf("prepare A must not deadlock against B's in-flight insert: %v", err) + } + if err := txA.Commit(f.ctx); err != nil { + t.Fatalf("commit A: %v", err) + } + select { + case err := <-bDone: + if err != nil { + t.Fatalf("B: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("B never completed after A committed") + } + }) + } +} diff --git a/tests/e2e-prod/suites/03-concurrency.test.ts b/tests/e2e-prod/suites/03-concurrency.test.ts index dee86f0d2..6946b185a 100644 --- a/tests/e2e-prod/suites/03-concurrency.test.ts +++ b/tests/e2e-prod/suites/03-concurrency.test.ts @@ -196,6 +196,43 @@ test("concurrency: parallel DELETE of the same agent is idempotent under content } }); +test("concurrency: 8 parallel sends from a normal agent — all accepted (no 5xx, no duplicates)", async () => { + // The accept transaction inserts the message and then prepares its sending + // operation under the gate; v1.9.0 deadlocked those two steps against each + // other (SQLSTATE 40P01 → 500) for parallel sends from one agent. The HITL + // case below caught it on the hold path; this covers the direct path, which + // has the same shape and carries almost all production traffic. + const slug = uniqueSlug("sendconc"); + const c = await client.post<{ email: string }>("/v1/agents", { + body: { email: `${slug}@${client.env.sharedDomain}`, name: "send-conc" }, + }); + assert.equal(c.status, 201); + const email = c.body!.email; + track("agent", email); + + const N = 8; + const sends = await Promise.all( + Array.from({ length: N }, (_, i) => + burst.post<{ message_id: string; status: string }>(`/v1/agents/${encodeURIComponent(email)}/messages`, { + body: { + to: [SINK_EMAIL], + subject: `parallel direct ${i}`, + text: `parallel direct send #${i}`, + }, + }), + ), + ); + + const ids = new Set(); + for (const r of sends) { + assert.ok(r.status === 202 || r.status === 200, `parallel direct send: status ${r.status}, body: ${r.raw.slice(0, 200)}`); + assert.ok(r.body?.message_id?.startsWith("msg_"), `message_id present and prefixed`); + ids.add(r.body!.message_id); + } + assert.equal(ids.size, N, `expected ${N} distinct message_ids, got ${ids.size}`); + info(SUITE, "parallel-direct-sends", `${N} parallel direct sends accepted with ${ids.size} distinct ids`); +}); + test("concurrency: 8 parallel sends from HITL agent — all queue (no dropped/duplicated)", async () => { const slug = uniqueSlug("hitlconc"); const c = await client.post<{ email: string }>("/v1/agents", {