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>
This commit is contained in:
Gregory SchierandClaude Opus 5.5 authored and Stetta task tk_mvsazf4rsaotrdpf committed 2026-10-28 10:34:14 -07:00
1 parent 39a7397ebb
commit e33aeee5ff
2 files changed
+46 -16

No files matched your search

+18 -1
View File
@@ -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;
+28 -15
View File
@@ -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}");
}
}