From e83538678a6de578c55d0889efa24f8acf050501 Mon Sep 17 00:00:00 2001 From: Arnas Donauskas Date: Fri, 22 May 2026 15:39:00 +0300 Subject: [PATCH] fix(flows): atomic execution_count increment via SECURITY DEFINER RPC MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- src/lib/flows/engine.ts | 20 +++++++---- .../012_flows_increment_counter.sql | 36 +++++++++++++++++++ 2 files changed, 49 insertions(+), 7 deletions(-) create mode 100644 supabase/migrations/012_flows_increment_counter.sql diff --git a/src/lib/flows/engine.ts b/src/lib/flows/engine.ts index 069c98a..99052e0 100644 --- a/src/lib/flows/engine.ts +++ b/src/lib/flows/engine.ts @@ -1004,13 +1004,19 @@ async function startNewRun( }); // Bump the flow's execution counter — used by the builder UI to // surface "X runs since activation" on the flow card. - await db - .from("flows") - .update({ - execution_count: flow.execution_count + 1, - last_executed_at: new Date().toISOString(), - }) - .eq("id", flow.id); + // + // Atomic RPC (migration 012) rather than read-modify-write: two + // concurrent webhooks starting runs for different contacts on the + // same flow would otherwise both read N and both write N+1, losing + // a count. Mirrors the automations engine's use of + // `increment_automation_execution_count` (migration 007). + 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. const outcome = await advanceFromNodeKey(db, run, flow.entry_node_id!); diff --git a/supabase/migrations/012_flows_increment_counter.sql b/supabase/migrations/012_flows_increment_counter.sql new file mode 100644 index 0000000..25eb4cf --- /dev/null +++ b/supabase/migrations/012_flows_increment_counter.sql @@ -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 = 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;