fix(sendingpolicy): lock the agent FOR NO KEY UPDATE in the accept path

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.
 

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

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)
+		}
+	}
+}

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

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")
+			}
+		})
+	}
+}

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<string>();
+  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", {