From e33aeee5ffb482ff11d7e5d256478fb805d02e1d Mon Sep 17 00:00:00 2001
From: Gregory Schier <gschier1990@gmail.com>
Date: Wed, 28 Oct 2026 10:34:14 -0700
Subject: [PATCH] Close the clear and count races the second review found

- Send a flow's write event inside its transaction, so a clear's event,
  sent after its own commit, always comes after it.
- Record a saved request only on a flow that still exists, so a save
  that finishes after a clear does not bring the flow back.
- Count the total again from the database a second after new flows,
  so a flow both in a page and in the events around it settles to the
  right count.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 apps/yaak-proxy/lib/store.ts           | 19 +++++++++++-
 crates-proxy/yaak-proxy-lib/src/lib.rs | 43 +++++++++++++++++---------
 2 files changed, 46 insertions(+), 16 deletions(-)

diff --git a/apps/yaak-proxy/lib/store.ts b/apps/yaak-proxy/lib/store.ts
index aef76401a..ff317dac9 100644
--- a/apps/yaak-proxy/lib/store.ts
+++ b/apps/yaak-proxy/lib/store.ts
@@ -15,6 +15,8 @@ import { listen, rpc } from "./rpc";
 
 /** How long new flows gather before they reach the list, so a burst renders once */
 const FLUSH_MS = 100;
+/** How long after new flows the total is counted again from the database */
+const RECOUNT_MS = 1000;
 
 /** The flows in memory: a window onto the database, which keeps them all */
 export const flowWindowAtom = atom<FlowWindow>(emptyWindow);
@@ -81,6 +83,7 @@ export function syncFlows(store: ReturnType<typeof createStore>) {
   let pending: ModelPayload[] = [];
   let timer: ReturnType<typeof setTimeout> | null = null;
   let loaded = false;
+  let loads = 0;
 
   function flush() {
     timer = null;
@@ -97,6 +100,21 @@ export function syncFlows(store: ReturnType<typeof createStore>) {
     if (change != null) {
       store.set(selectedFlowAtom, change.change.type === "delete" ? null : change.model);
     }
+    if (recountTimer == null) recountTimer = setTimeout(recount, RECOUNT_MS);
+  }
+
+  // The total counts each new flow as it comes, which a flow both in a page and in the events
+  // around it can count twice or not at all; a count from the database a moment later settles it.
+  let recountTimer: ReturnType<typeof setTimeout> | null = null;
+  function recount() {
+    recountTimer = null;
+    const thisLoad = loads;
+    void rpc("list_flows", { limit: 0 }).then((res) => {
+      if (thisLoad !== loads) return;
+      store.set(flowWindowAtom, (win) =>
+        win.total === res.total ? win : { ...win, total: res.total },
+      );
+    });
   }
 
   listen("model_write", (payload) => {
@@ -113,7 +131,6 @@ export function syncFlows(store: ReturnType<typeof createStore>) {
     load();
   });
 
-  let loads = 0;
   function load() {
     loaded = false;
     const thisLoad = ++loads;
diff --git a/crates-proxy/yaak-proxy-lib/src/lib.rs b/crates-proxy/yaak-proxy-lib/src/lib.rs
index 2c5b04585..efe5e1038 100644
--- a/crates-proxy/yaak-proxy-lib/src/lib.rs
+++ b/crates-proxy/yaak-proxy-lib/src/lib.rs
@@ -319,14 +319,21 @@ impl ProxyCtx {
         let saved = ClientDb::open(&self.client_data_dir)?.save(&flow, &body, req)?;
 
         let flow = Flow { saved_request_id: Some(saved.request.id.clone()), ..flow };
-        let (flow, created) = self
-            .db
-            .with_tx(|tx| tx.upsert(&flow, &UpdateSource::Background))
+        self.db
+            .with_tx(|tx| {
+                // Cleared while saving: recording it would bring the flow back without bodies.
+                if tx.find_optional::<Flow>(FlowIden::Id, flow_id).is_none() {
+                    return Ok(());
+                }
+                let (flow, created) = tx.upsert(&flow, &UpdateSource::Background)?;
+                // Sent before the commit, so it cannot reach the app after a clear that follows.
+                self.events.emit(
+                    "model_write",
+                    &ModelPayload { model: flow, change: ModelChangeEvent::Upsert { created } },
+                );
+                Ok::<_, yaak_database::Error>(())
+            })
             .map_err(|e| format!("Failed to record the saved request on the flow: {e}"))?;
-        self.events.emit(
-            "model_write",
-            &ModelPayload { model: flow, change: ModelChangeEvent::Upsert { created } },
-        );
         Ok(saved)
     }
 
@@ -334,6 +341,8 @@ impl ProxyCtx {
     /// many there were.
     pub fn clear_flows(&self) -> Result<usize, String> {
         let cleared = flows::clear(&self.db, &self.blobs)?;
+        // After the commit, so the app reloads what is left. Flows written before the clear sent
+        // their events before it began.
         self.events.emit("flows_cleared", &FlowsClearedPayload {});
         Ok(cleared)
     }
@@ -746,14 +755,18 @@ fn finish_flow(db: &ProxyQueryManager, events: &RpcEventEmitter, f: InFlight) {
 
     let mut flow = f.flow;
     flow.content_type = find_header(&flow.response_headers, "content-type").map(String::from);
-    match db.with_tx(|ctx| ctx.upsert(&flow, &UpdateSource::Background)) {
-        Ok((saved, created)) => {
-            events.emit(
-                "model_write",
-                &ModelPayload { model: saved, change: ModelChangeEvent::Upsert { created } },
-            );
-        }
-        Err(e) => warn!("Failed to write proxy flow: {e}"),
+    let written = db.with_tx(|ctx| {
+        let (saved, created) = ctx.upsert(&flow, &UpdateSource::Background)?;
+        // Sent while this holds the writer, before the commit, so a clear's event, sent after
+        // its own commit, always comes after it.
+        events.emit(
+            "model_write",
+            &ModelPayload { model: saved, change: ModelChangeEvent::Upsert { created } },
+        );
+        Ok::<_, yaak_database::Error>(())
+    });
+    if let Err(e) = written {
+        warn!("Failed to write proxy flow: {e}");
     }
 }