fix(flows): atomic execution_count increment via SECURITY DEFINER RPC

startNewRun read flows.execution_count from a cached row and wrote
`cached + 1`. Two concurrent webhooks starting runs on the same flow
for different contacts both read N and both write N+1 — one count
permanently lost.

Mirrors the automations engine's solution (migration 007): a tiny
SECURITY DEFINER function that does the increment server-side, locked
to the service role. The engine's call becomes a single RPC.

The counter is best-effort: if the RPC fails we log + continue rather
than failing the run for a stat error.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Arnas Donauskas 2026-05-22 15:39:00 +03:00
parent 9ff4a8d1ba
commit e83538678a
No known key found for this signature in database
2 changed files with 49 additions and 7 deletions

View File

@ -1004,13 +1004,19 @@ async function startNewRun(
}); });
// Bump the flow's execution counter — used by the builder UI to // Bump the flow's execution counter — used by the builder UI to
// surface "X runs since activation" on the flow card. // surface "X runs since activation" on the flow card.
await db //
.from("flows") // Atomic RPC (migration 012) rather than read-modify-write: two
.update({ // concurrent webhooks starting runs for different contacts on the
execution_count: flow.execution_count + 1, // same flow would otherwise both read N and both write N+1, losing
last_executed_at: new Date().toISOString(), // a count. Mirrors the automations engine's use of
}) // `increment_automation_execution_count` (migration 007).
.eq("id", flow.id); const { error: incErr } = await db.rpc("increment_flow_execution_count", {
p_flow_id: flow.id,
});
if (incErr) {
// Non-fatal — the run itself succeeded; only the counter is off.
console.error("[flows] execution_count rpc error:", incErr.message);
}
// Run the advance loop starting from the entry node. // Run the advance loop starting from the entry node.
const outcome = await advanceFromNodeKey(db, run, flow.entry_node_id!); const outcome = await advanceFromNodeKey(db, run, flow.entry_node_id!);

View File

@ -0,0 +1,36 @@
-- ============================================================
-- 012_flows_increment_counter.sql
--
-- Atomic increment of flows.execution_count + refresh of
-- last_executed_at. Called via PostgREST RPC from the engine.
--
-- Before this, startNewRun did a read-modify-write:
-- UPDATE flows SET execution_count = <cached + 1> WHERE id = ...
-- so two concurrent dispatches (e.g. two webhooks for the same flow
-- starting runs for different contacts in the same second) could both
-- read N and both write N+1, permanently losing one count.
--
-- Mirrors migration 007 for automations — same shape, same security
-- posture. Idempotent: safe to re-run.
-- ============================================================
CREATE OR REPLACE FUNCTION increment_flow_execution_count(p_flow_id UUID)
RETURNS VOID
LANGUAGE sql
SECURITY DEFINER
SET search_path = public
AS $$
UPDATE flows
SET
execution_count = execution_count + 1,
last_executed_at = NOW()
WHERE id = p_flow_id;
$$;
-- Only the service role needs to call this (engine uses the
-- service-role client). Explicitly lock anon / authenticated out so
-- an authenticated user can't juice someone else's counter via RPC.
REVOKE ALL ON FUNCTION increment_flow_execution_count(UUID) FROM PUBLIC;
REVOKE ALL ON FUNCTION increment_flow_execution_count(UUID) FROM anon;
REVOKE ALL ON FUNCTION increment_flow_execution_count(UUID) FROM authenticated;
GRANT EXECUTE ON FUNCTION increment_flow_execution_count(UUID) TO service_role;