Stop task actions and invitation emails deadlocking with a member leaving

apps/reminder-worker/src/invitations.ts

@@ -105,6 +105,11 @@ export async function sendInvitationEmail(pool: Pool, actor: string, token: stri
   let payload: Payload;
   try {
     await client.query("BEGIN");
+    // An admin leaving, removed or made a member locks the commune and then
+    // withdraws their invitations, so the sender takes the commune first and
+    // the two wait in turn.
+    await client.query(`SELECT 1 FROM public.communes c JOIN public.commune_invites i ON c.id=i.commune_id
+      WHERE i.token_hash=$1 FOR KEY SHARE OF c`, [tokenHash]);
     const { rows } = await client.query(`
       SELECT i.*, c.name AS commune_name, u.name AS inviter_name FROM public.commune_invites i
       JOIN public.communes c ON c.id=i.commune_id

apps/reminder-worker/tests/invitations.mjs

@@ -11,7 +11,7 @@ const pool = new Pool({ connectionString: process.env.DATABASE_URL });
 const senderPool = new Pool({ connectionString: process.env.DATABASE_URL, options: "-c search_path=pg_catalog" });
 const send = (actorId, token, mail = config) => sendInvitationEmail(senderPool, actorId, token, mail);
 const tokenHash = (token) => createHash("sha256").update(token).digest();
-const actor = randomUUID(), outsider = randomUUID(), bulkSender = randomUUID();
+const actor = randomUUID(), outsider = randomUUID(), bulkSender = randomUUID(), otherAdmin = randomUUID(), removedAdmin = randomUUID();
 const config = { apiKey: "test-only", from: "Commune <test@example.test>", appUrl: "https://example.test" };
 const calls = [];
 let providerStatus = 200;
@@ -27,12 +27,23 @@ async function asUser(userId, fn) {
 }
 const asActor = (fn) => asUser(actor, fn);
 const refusal = (status, message) => (e) => e.status === status && (!message || e.message === message);
+const settle = (promise) => promise.then((result) => ({ result }), (error) => ({ error }));
+// Polls until the query's row says ok, so a race can wait for a session to
+// be stuck behind a lock.
+async function waitUntil(sql, values) {
+  for (let attempt = 0; attempt < 400; attempt += 1) {
+    if ((await pool.query(sql, values)).rows[0]?.ok) return;
+    await new Promise((resolve) => setTimeout(resolve, 25));
+  }
+  throw new Error(`Gave up waiting for: ${sql}`);
+}
+const UNAVAILABLE = "Invitation is unavailable or you are no longer a commune admin.";
 const NAME_REFUSAL = "This invitation can't be emailed because a name in it looks like a link or phone number. Copy the link and share it yourself.";
 const SUPPRESSED = "This address can't receive Commune emails. Share the link another way.";
 const NEW_ACCOUNT = "New accounts can email up to 5 invitations a day. Copy the link and share it yourself.";
 const PAUSED = "Invitation emails are paused for today. Copy the link and share it yourself.";
 const deliveries = async () => +(await pool.query("SELECT count(*) FROM public.invitation_email_deliveries WHERE sender_id=$1", [actor])).rows[0].count;
-let commune, bulkCommune;
+let commune, bulkCommune, removalCommune;
 try {
   // The email names its inviter, quotes the commune and offers a report link
   // that names the invitation by id, never by token.
@@ -197,11 +208,45 @@ try {
   assert.equal(calls.length, before + 1, "simultaneous sends dispatch only once");
   await asActor(c => c.query("UPDATE public.commune_members SET role='member' WHERE commune_id=$1 AND user_id=$2", [commune, actor]));
   await assert.rejects(send(actor, parallelToken), refusal(404), "former admins cannot send");
-  console.log("Invitation email: authorization, inviter and commune naming, report link, contact-like names, suppressed addresses, new-account and Commune-wide limits, index use, provider failure/retry, deduplication, rate limit, expiry, token privacy and HTML escaping passed. No real email sent.");
+
+  // An admin removed while their invitation is being emailed. The removal
+  // locks the commune and then withdraws the admin's invitations, so the
+  // sender takes the commune first, waits for the removal and then finds the
+  // invitation gone. Holding the admin's membership row stops the removal
+  // between its two locks, so this interleaving happens every time.
+  await pool.query('INSERT INTO neon_auth."user" (id,name,email,"emailVerified") VALUES ($1,\'Other admin\',$2,true), ($3,\'Removed admin\',$4,true)',
+    [otherAdmin, `${otherAdmin}@example.test`, removedAdmin, `${removedAdmin}@example.test`]);
+  removalCommune = (await asUser(otherAdmin, c => c.query("SELECT public.create_commune('Removal test','household','GBP') AS id"))).rows[0].id;
+  await asUser(otherAdmin, c => c.query("INSERT INTO public.commune_members (commune_id,user_id,role) VALUES ($1,$2,'admin')", [removalCommune, removedAdmin]));
+  const removedToken = (await asUser(removedAdmin, c => c.query("SELECT public.create_commune_invite($1,'removal@example.test','member') AS token", [removalCommune]))).rows[0].token;
+  const holder = await pool.connect(), remover = await pool.connect();
+  try {
+    await holder.query("BEGIN");
+    await holder.query("SELECT 1 FROM public.commune_members WHERE commune_id=$1 AND user_id=$2 FOR UPDATE", [removalCommune, removedAdmin]);
+    await remover.query("BEGIN");
+    await remover.query("SELECT set_config('request.jwt.claims',json_build_object('sub',$1::text)::text,true)", [otherAdmin]);
+    const { pid } = (await remover.query("SELECT pg_backend_pid() AS pid")).rows[0];
+    const removal = settle(remover.query("SELECT public.manage_commune_member($1,$2,'remove')", [removalCommune, removedAdmin]));
+    await waitUntil("SELECT wait_event_type='Lock' AS ok FROM pg_stat_activity WHERE pid=$1", [pid]);
+    before = calls.length;
+    const sending = settle(send(removedAdmin, removedToken));
+    await waitUntil("SELECT count(*)>0 AS ok FROM pg_stat_activity WHERE $1=ANY(pg_blocking_pids(pid))", [pid]);
+    await holder.query("COMMIT");
+    const [removed, sent] = await Promise.all([
+      removal.then(async (outcome) => { await remover.query(outcome.error ? "ROLLBACK" : "COMMIT"); return outcome; }), sending]);
+    for (const outcome of [removed, sent]) assert.notEqual(outcome.error?.code, "40P01", `the removal and the sender deadlocked: ${outcome.error?.message}`);
+    assert.equal(removed.error, undefined, removed.error?.message);
+    assert.equal(sent.error?.status, 404, sent.error?.message ?? "the invitation was emailed");
+    assert.equal(sent.error.message, UNAVAILABLE);
+  } finally { holder.release(true); remover.release(true); }
+  assert.equal((await pool.query("SELECT active FROM public.commune_members WHERE commune_id=$1 AND user_id=$2", [removalCommune, removedAdmin])).rows[0].active, false);
+  assert.equal(+(await pool.query("SELECT count(*) FROM public.invitation_email_deliveries WHERE commune_id=$1", [removalCommune])).rows[0].count, 0, "no delivery is left");
+  assert.equal(calls.length, before, "the provider is never called");
+  console.log("Invitation email: authorization, inviter and commune naming, report link, contact-like names, suppressed addresses, new-account and Commune-wide limits, index use, provider failure/retry, deduplication, rate limit, expiry, token privacy, HTML escaping and an admin removed while sending passed. No real email sent.");
 } finally {
-  await pool.query("DELETE FROM public.communes WHERE id = ANY($1::uuid[])", [[commune, bulkCommune].filter(Boolean)]);
+  await pool.query("DELETE FROM public.communes WHERE id = ANY($1::uuid[])", [[commune, bulkCommune, removalCommune].filter(Boolean)]);
   await pool.query("DELETE FROM public.email_suppressions WHERE provider_event_id LIKE 'msg_test_%'");
-  await pool.query('DELETE FROM neon_auth."user" WHERE id = ANY($1::text[])', [[actor, bulkSender]]);
+  await pool.query('DELETE FROM neon_auth."user" WHERE id = ANY($1::text[])', [[actor, bulkSender, otherAdmin, removedAdmin]]);
   await pool.end();
   await senderPool.end();
 }

docs/HANDOFF.md

@@ -65,6 +65,10 @@ lifecycle pack. Work is paused for a handoff, not a production release.
   (`apps/web/src/data/covers.ts`), on a calm tone for the commune type while
   they load or if they fail; bundling them needs the three original files from
   Edem. Production builds preload the data client chunk (`vite.config.ts`).
+- Also from 9 October, for review: completing, skipping, handing over and
+  undoing a task now lock the commune first (migration 0037), and the
+  invitation email sender does the same, so either meeting a member leaving
+  waits for it instead of failing as a deadlock.
 - The website is `web/` (the editorial homepage); the earlier v2 site is in
   `web-backup/`. Launch names `website` (3004) and `website-backup` (3003).
 - No push, deployment, hosted migration or provider change has been made. Do
@@ -178,7 +182,7 @@ no further copy-cleanup pass is needed for the handoff.
 Read [the product-depth roadmap](PRODUCT-DEPTH-ROADMAP-2026-10-02.md) and the
 Remaining section of [money lifecycle](MONEY-LIFECYCLE-2026-10-02.md).
 
-1. **Hosted parity.** Migrations 0013 to 0036, schema-cache refresh, the
+1. **Hosted parity.** Migrations 0013 to 0037, schema-cache refresh, the
    matching web client and sender, auth, invitations and real email delivery
    are all unverified on Neon. Apply migrations before deploying the client.
    Measure cold/warm/idle loading with separate accounts. Do not declare

infra/local/README.md

@@ -41,7 +41,8 @@ files in order using a database owner connection:
 35. `infra/neon/migrations/0034_database_hardening.sql`
 36. `infra/neon/migrations/0035_currency_rule.sql`
 37. `infra/neon/migrations/0036_lighter_loading.sql`
-38. `infra/local/after-core.sql`
+38. `infra/neon/migrations/0037_task_command_lock_order.sql`
+39. `infra/local/after-core.sql`
 
 `after-core.sql` is idempotent. Re-run it after applying a newer migration to
 an existing local database so the local-only reminder helpers stay current.

infra/neon/README.md

@@ -14,8 +14,8 @@ this is not a replacement of Neon Auth or a switch back to Supabase.
 - [Current design language and preserved layouts](../../docs/DESIGN-LANGUAGE.md)
 - [Local build-out from 5 October: migrations 0020 onwards](../../docs/LOCAL-BUILDOUT-2026-10-05.md)
 
-Migrations **0015 to 0036 are local only**. They have not been applied to any
-Neon branch. Release order when they are: apply 0013 to 0036 in numeric order
+Migrations **0015 to 0037 are local only**. They have not been applied to any
+Neon branch. Release order when they are: apply 0013 to 0037 in numeric order
 to the preview branch, refresh that branch's Data API schema cache, then
 deploy the matching web client. The new client calls `send_reminder_command`,
 `record_share_receipt`, `reverse_share_receipt`, `correct_expense`,
@@ -89,6 +89,16 @@ Six of the 9 October migrations change that rule:
 `set_recurring_bill_archived`; production has no recurring bills yet, so only
 the new client uses them.
 
+0037 makes completing, skipping, reopening, handing over and undoing a task
+lock the commune before the task, as edits and removals have since 0027, so
+one of them meeting a member's departure waits for it instead of failing as a
+deadlock. It needs no client or worker change. Separately, the invitation
+email sender in `apps/reminder-worker` now locks the invitation's commune
+before the invitation, so emailing an invitation while its admin leaves, is
+removed or is made a member waits and is then refused as unavailable instead
+of failing as a deadlock. That is a worker change only, released with the
+worker deploys above.
+
 ## Last recorded production evidence — 27 September 2026
 
 The app is served at `https://app.ourcommune.io` from `main`. The latest
@@ -252,7 +262,7 @@ rate limits) are recorded in [`docs/AUTH-SETTINGS.md`](../../docs/AUTH-SETTINGS.
    Broad grants would let direct writes bypass ledger invariants. Keep preview
    branches separate from production.
 3. Apply `migrations/0001_commune_core.sql` through
-   `migrations/0036_lighter_loading.sql` in numeric order to the
+   `migrations/0037_task_command_lock_order.sql` in numeric order to the
    development branch via an **unpooled** connection. Run `tests/grants.sql`,
    review RLS, and run two-account signed-JWT tests before applying the same
    reviewed migrations to production. After a migration adds or changes a

infra/neon/migrations/0037_task_command_lock_order.sql

@@ -0,0 +1,86 @@
+-- Completing, skipping, reopening, handing over and undoing a task now lock
+-- the commune before the task, as editing and removing have since 0027. A
+-- member leaving locks the commune and then each of their tasks. These
+-- commands locked the task first and needed the commune later, when the
+-- occurrence or activity line they add takes a key-share lock on it through
+-- its foreign key. So when a member left while someone acted on one of their
+-- tasks, each waited on the other until PostgreSQL cancelled one as a
+-- deadlock, and the action or the departure failed. Now the command waits
+-- for the departure to finish, then finds the task as the departure left it
+-- and goes ahead or is refused with its usual message.
+--
+-- Only the start of each function changes, so each is patched in place: the
+-- task's commune is read without a lock and locked, then the task (for undo,
+-- the task through its occurrence, then the occurrence) is locked and read
+-- as before. A task never moves to another commune, so that commune is still
+-- the task's once the locks are held, and every check runs on the task as it
+-- is then. Nothing else locks a task or an occurrence before its commune:
+-- adding a task locks the commune first, and the snapshot and search only
+-- read. No client change. Apply after 0036.
+BEGIN;
+
+-- Each patch must apply, and leave the commune as the first row its
+-- function locks.
+DO $$
+DECLARE v_original text; v_source text;
+BEGIN
+  v_original := pg_get_functiondef('public.record_operation_occurrence(uuid,text,date,date,uuid,text)'::regprocedure);
+  v_source := replace(v_original, '  v_op public.commune_operations%ROWTYPE;',
+    '  v_commune_id uuid;
+  v_op public.commune_operations%ROWTYPE;');
+  v_source := replace(v_source,
+    '  -- The row lock serialises two people acting on the same task.
+  SELECT * INTO v_op FROM public.commune_operations WHERE id = p_operation_id FOR UPDATE;',
+    '  -- The commune first, then the task, as in update_operation: the order a
+  -- member leaving takes them in. The other order lets the two wait on each
+  -- other until one fails as a deadlock.
+  SELECT commune_id INTO v_commune_id FROM public.commune_operations WHERE id = p_operation_id;
+  PERFORM 1 FROM public.communes WHERE id = v_commune_id FOR UPDATE;
+  -- The row lock serialises two people acting on the same task.
+  SELECT * INTO v_op FROM public.commune_operations WHERE id = p_operation_id FOR UPDATE;');
+  IF position('  v_commune_id uuid;' IN v_source) = 0
+     OR split_part(v_source, ' FOR UPDATE', 1) !~ 'PERFORM 1 FROM public\.communes WHERE id = v_commune_id$' THEN
+    RAISE EXCEPTION 'Unsupported occurrence definition';
+  END IF;
+  EXECUTE v_source;
+
+  v_original := pg_get_functiondef('public.hand_over_operation(uuid,text,date,text,date,text,uuid)'::regprocedure);
+  v_source := replace(v_original, '  v_op public.commune_operations%ROWTYPE;',
+    '  v_commune_id uuid;
+  v_op public.commune_operations%ROWTYPE;');
+  v_source := replace(v_source,
+    '  SELECT * INTO v_op FROM public.commune_operations WHERE id = p_operation_id FOR UPDATE;',
+    '  -- The commune first, then the task, as in update_operation.
+  SELECT commune_id INTO v_commune_id FROM public.commune_operations WHERE id = p_operation_id;
+  PERFORM 1 FROM public.communes WHERE id = v_commune_id FOR UPDATE;
+  SELECT * INTO v_op FROM public.commune_operations WHERE id = p_operation_id FOR UPDATE;');
+  IF position('  v_commune_id uuid;' IN v_source) = 0
+     OR split_part(v_source, ' FOR UPDATE', 1) !~ 'PERFORM 1 FROM public\.communes WHERE id = v_commune_id$' THEN
+    RAISE EXCEPTION 'Unsupported handover definition';
+  END IF;
+  EXECUTE v_source;
+
+  v_original := pg_get_functiondef('public.undo_operation_occurrence(uuid)'::regprocedure);
+  v_source := replace(v_original, '  v_op public.commune_operations%ROWTYPE;',
+    '  v_commune_id uuid;
+  v_op public.commune_operations%ROWTYPE;');
+  v_source := replace(v_source,
+    '  SELECT o.* INTO v_op FROM public.commune_operations o
+    JOIN public.operation_occurrences h ON h.operation_id = o.id WHERE h.id = p_occurrence_id FOR UPDATE OF o;',
+    '  -- The commune first, then the task, then the occurrence, as in
+  -- update_operation.
+  SELECT o.commune_id INTO v_commune_id FROM public.commune_operations o
+    JOIN public.operation_occurrences h ON h.operation_id = o.id WHERE h.id = p_occurrence_id;
+  PERFORM 1 FROM public.communes WHERE id = v_commune_id FOR UPDATE;
+  SELECT o.* INTO v_op FROM public.commune_operations o
+    JOIN public.operation_occurrences h ON h.operation_id = o.id WHERE h.id = p_occurrence_id FOR UPDATE OF o;');
+  IF position('  v_commune_id uuid;' IN v_source) = 0
+     OR split_part(v_source, ' FOR UPDATE', 1) !~ 'PERFORM 1 FROM public\.communes WHERE id = v_commune_id$' THEN
+    RAISE EXCEPTION 'Unsupported undo definition';
+  END IF;
+  EXECUTE v_source;
+END;
+$$;
+
+NOTIFY pgrst, 'reload schema';
+COMMIT;