From 45eb718dd78bb1adf84ca5c561fc03f424963603 Mon Sep 17 00:00:00 2001 From: debpalash <4178343+debpalash@users.noreply.github.com> Date: Tue, 13 Oct 2026 09:42:56 +0530 Subject: [PATCH] fix(workbook): refresh only rows created by its own sourcing Refresh detected "newly sourced" rows by snapshotting row ids before sourcing and diffing afterward, so any row inserted concurrently by another writer (CSV import, API insert, another job) was treated as sourced and had every non-output runnable column run, including paid enrichments. materialize_source (ICP, crm_import and people_search paths) now returns the ids it inserted as an additive `added_row_ids` key, and refresh uses those instead of the snapshot diff, dropping both row-id queries. Follow-up to #74. Signed-off-by: debpalash Co-Authored-By: Claude Opus 5.5 --- apps/api/services/workbook/crm_import.py | 3 +- apps/api/services/workbook/people_search.py | 1 + apps/api/services/workbook/refresh.py | 14 ++++---- apps/api/services/workbook/source_engine.py | 5 ++- ...012-refresh-dependent-and-new-row-cells.md | 22 ++++++++---- tests/test_crm_import_sync.py | 5 ++- tests/test_workbook_refresh_cell_scope.py | 35 +++++++++++++++++-- 7 files changed, 64 insertions(+), 21 deletions(-) diff --git a/apps/api/services/workbook/crm_import.py b/apps/api/services/workbook/crm_import.py index b2496276..73addd5c 100644 --- a/apps/api/services/workbook/crm_import.py +++ b/apps/api/services/workbook/crm_import.py @@ -222,4 +222,5 @@ async def materialize_crm_import( f"CRM import {workbook_id}/{column_id} ({crm}): " f"found={found} added={added} skipped={skipped}" ) - return {"found": found, "added": added, "skipped": skipped, "crm": crm} + return {"found": found, "added": added, "skipped": skipped, "crm": crm, + "added_row_ids": list(_added_row_ids)} diff --git a/apps/api/services/workbook/people_search.py b/apps/api/services/workbook/people_search.py index d81367b6..8fafbfeb 100644 --- a/apps/api/services/workbook/people_search.py +++ b/apps/api/services/workbook/people_search.py @@ -427,6 +427,7 @@ async def materialize_people_search( ) result = { "found": found, "added": added, "skipped": skipped, + "added_row_ids": list(_added_row_ids), "companies": len(companies), "searches_used": searches_used, "search_failures": search_failures, } diff --git a/apps/api/services/workbook/refresh.py b/apps/api/services/workbook/refresh.py index 6ce8e398..759a7a51 100644 --- a/apps/api/services/workbook/refresh.py +++ b/apps/api/services/workbook/refresh.py @@ -116,15 +116,16 @@ async def _refresh_workbook_impl(workbook_id: str, reason: str, workspace_id: st source_cols = [c for c in cols if c.get("type") == "source"] enrichment_cols = [c for c in cols if c.get("type") in ("enrichment", "waterfall", "ai_formula")] ttl_map = policy.get("staleness_ttl_days") or {} - previous_row_ids = {row.id for row in db.query(WorkbookRow.id).filter( - WorkbookRow.workbook_id == workbook_id - ).all()} - # 1) Re-source (append-only via entity dedup) + # 1) Re-source (append-only via entity dedup). Track exactly the rows our + # own sourcing inserted: rows added concurrently by other writers (CSV + # import, API insert, another job) must not get every column run here. sourced = 0 + added_row_ids: list = [] for col in source_cols: res = await materialize_source(workbook_id, col["id"], workspace_id) sourced += res.get("added", 0) + added_row_ids.extend(res.get("added_row_ids") or []) # 2) Refresh stale cells and their reactive downstream dependencies. def is_reactive(column: dict) -> bool: @@ -140,14 +141,11 @@ def is_reactive(column: dict) -> bool: reenriched = 0 with SessionLocal() as db: stale = _stale_row_columns(db, workbook_id, enrichment_cols, ttl_map) if enrichment_cols else {} - added_row_ids = {row.id for row in db.query(WorkbookRow.id).filter( - WorkbookRow.workbook_id == workbook_id - ).all()} - previous_row_ids for row_id, column_ids in stale.items(): dependents = downstream_columns(cols, set(column_ids), eligible=is_reactive) stale[row_id] = list(dict.fromkeys([*column_ids, *(c["id"] for c in dependents if c.get("id"))])) if new_row_columns: - for row_id in added_row_ids: + for row_id in dict.fromkeys(added_row_ids): stale[str(row_id)] = list(new_row_columns) if stale: result = await run_workbook_enrichment( diff --git a/apps/api/services/workbook/source_engine.py b/apps/api/services/workbook/source_engine.py index 811520de..0f048cb8 100644 --- a/apps/api/services/workbook/source_engine.py +++ b/apps/api/services/workbook/source_engine.py @@ -76,7 +76,9 @@ async def materialize_source( ``workspace_id`` is the tenant this run belongs to. It arrives out-of-band (job payload / caller) — never read back off the workbook row — and binds the RLS scope for the whole run, so the wb/row reads and the lead-store read-back - are all tenant-correct. Returns a summary {found, added, skipped, query}. + are all tenant-correct. Returns a summary {found, added, skipped, query, + added_row_ids}; ``added_row_ids`` lists exactly the rows this run inserted + (absent on early-error returns, so callers should ``.get(..., [])``). """ from apps.api.core.tenancy import workspace_scope @@ -384,6 +386,7 @@ async def _materialize_source_impl( "shortfall": max(0, requested_count - total_rows_after), "found": found, "added": added, + "added_row_ids": list(_added_row_ids), "skipped": skipped, "rejected_by_reason": rejected_by_reason, "exhausted_sources": exhausted_sources, diff --git a/docs/adr/20261012-refresh-dependent-and-new-row-cells.md b/docs/adr/20261012-refresh-dependent-and-new-row-cells.md index 0fe58f47..20160ced 100644 --- a/docs/adr/20261012-refresh-dependent-and-new-row-cells.md +++ b/docs/adr/20261012-refresh-dependent-and-new-row-cells.md @@ -11,14 +11,18 @@ and research cells uninitialized. Expand each existing row's stale-cell scope through the existing transitive `downstream_columns` helper, using the row-edit path's reactive eligibility -policy. Snapshot persisted row IDs before sourcing and compare them afterward. -New rows initialize every runnable non-output column. Pass the resulting exact +policy. The source engine reports the IDs of the rows each run inserted +(`added_row_ids` in the `materialize_source` summary); those rows, and only +those, initialize every runnable non-output column. Pass the resulting exact per-row scopes through the existing enrichment runner. ## Alternatives -Returning added IDs from the source engine requires changing its return contract -and each caller. Running all columns on every stale row would restore dependent +The original version of this decision snapshotted persisted row IDs before +sourcing and diffed afterward, avoiding a source-engine contract change. That +misclassified rows inserted concurrently by other writers (CSV import, API +insert, another job) as sourced, running paid enrichments on them; it was +replaced by the additive `added_row_ids` key. Running all columns on every stale row would restore dependent values but discard this PR's cost-saving behavior for unrelated fresh cells. Running outputs during sourcing would introduce unrequested external dispatch. @@ -26,9 +30,13 @@ Running outputs during sourcing would introduce unrequested external dispatch. Changed inputs propagate through derived columns while unrelated fresh cells remain intact. Explicitly non-reactive dependents stay excluded; outputs remain -non-reactive by default and new-row initialization excludes outputs. Row-ID -comparison adds two scoped ID reads and identifies rows added during the cycle. -It is not a concurrent-writer snapshot isolation guarantee. +non-reactive by default and new-row initialization excludes outputs. + +Resolved limitation: the snapshot diff was not concurrent-writer safe. Refresh +now uses the IDs the source engine itself inserted, so rows added by another +writer during sourcing are left alone (covered by +`test_concurrently_inserted_row_is_not_treated_as_sourced`). No extra row-ID +queries are issued. ## Validation diff --git a/tests/test_crm_import_sync.py b/tests/test_crm_import_sync.py index dd75631b..17422a69 100644 --- a/tests/test_crm_import_sync.py +++ b/tests/test_crm_import_sync.py @@ -174,10 +174,12 @@ def test_hubspot_import_paging_mapping_and_sync_key(db, monkeypatch, flag_on, hu col_id = _mk_workbook(db, source={"kind": "crm_import", "crm": "hubspot"}) res = _run(se.materialize_source("wb1", col_id, WS_A)) + added_ids = res.pop("added_row_ids") assert res == {"found": 3, "added": 3, "skipped": 0, "crm": "hubspot"} assert len(calls) == 2 # two pages rows = _rows(db) + assert added_ids == [r["id"] for r in rows] # exactly the rows this run inserted assert [r["crm_external_id"] for r in rows] == ["1", "2", "3"] assert all(r["crm_type"] == "hubspot" and r["crm_object"] == "contact" for r in rows) assert rows[0]["email"] == "alice@x.com" @@ -200,7 +202,7 @@ def test_hubspot_import_rerun_dedups_on_external_id(db, monkeypatch, flag_on, hu first = _run(se.materialize_source("wb1", col_id, WS_A)) again = _run(se.materialize_source("wb1", col_id, WS_A)) assert first["added"] == 3 - assert again == {"found": 3, "added": 0, "skipped": 3, "crm": "hubspot"} + assert again == {"found": 3, "added": 0, "skipped": 3, "crm": "hubspot", "added_row_ids": []} assert len(_rows(db)) == 3 @@ -321,6 +323,7 @@ def handler(request: httpx.Request) -> httpx.Response: "kind": "crm_import", "crm": "salesforce", "filter": "Email != null", }) res = _run(se.materialize_source("wb1", col_id, WS_A)) + assert len(res.pop("added_row_ids")) == 2 assert res == {"found": 2, "added": 2, "skipped": 0, "crm": "salesforce"} assert seen["soql"].startswith( "SELECT Id, Email, FirstName, LastName, Account.Name, Phone, Title FROM Contact" diff --git a/tests/test_workbook_refresh_cell_scope.py b/tests/test_workbook_refresh_cell_scope.py index eeef88df..08321342 100644 --- a/tests/test_workbook_refresh_cell_scope.py +++ b/tests/test_workbook_refresh_cell_scope.py @@ -143,9 +143,9 @@ def test_refresh_recomputes_transitive_dependents_without_unrelated_spend(refres assert sent == [] -def test_new_source_row_runs_every_runnable_column_without_dispatch(refresh_runner, monkeypatch): +def _sourced_workbook(sessions, monkeypatch, *, concurrent_row: bool): + """Workbook whose sourcing adds row 202; optionally another writer adds 303 meanwhile.""" from apps.api.services.workbook import source_engine, http_column, research_column, agent_column - sessions, now, calls, sent = refresh_runner columns = [ {"id": "source", "type": "source"}, {"id": "email", "type": "input"}, @@ -169,8 +169,12 @@ async def source(_workbook_id, _column_id, _workspace_id): with sessions() as db: db.add(WorkbookRow(id=202, workbook_id=wid, workspace_id="refresh-scope", position=1, data={"email": "new@example.test"})) + if concurrent_row: + # A CSV import / API insert landing while sourcing is in flight. + db.add(WorkbookRow(id=303, workbook_id=wid, workspace_id="refresh-scope", position=2, + data={"email": "imported@example.test"})) db.commit() - return {"added": 1} + return {"added": 1, "added_row_ids": [202]} async def http(*args, **kwargs): return {"value": "HTTP", "error": None} async def research(*args, **kwargs): @@ -181,6 +185,12 @@ async def agent(*args, **kwargs): monkeypatch.setattr(http_column, "execute_http_column", http) monkeypatch.setattr(research_column, "execute_research_column", research) monkeypatch.setattr(agent_column, "run_agent_cell", agent) + return wid + + +def test_new_source_row_runs_every_runnable_column_without_dispatch(refresh_runner, monkeypatch): + sessions, now, calls, sent = refresh_runner + wid = _sourced_workbook(sessions, monkeypatch, concurrent_row=False) result = asyncio.run(refresh.refresh_workbook(wid, workspace_id="refresh-scope")) with sessions() as db: added = db.get(WorkbookRow, 202) @@ -190,3 +200,22 @@ async def agent(*args, **kwargs): assert "send" not in added.enrichments assert result == {"sourced": 1, "reenriched": 4, "stale_rows": 1} assert sent == calls == [] + + +def test_concurrently_inserted_row_is_not_treated_as_sourced(refresh_runner, monkeypatch): + """Only rows the refresh's own sourcing created get every column run. + + A row another writer inserts while sourcing is in flight is not ours: the + scheduled refresh must not spend paid enrichments on it. + """ + sessions, now, calls, sent = refresh_runner + wid = _sourced_workbook(sessions, monkeypatch, concurrent_row=True) + result = asyncio.run(refresh.refresh_workbook(wid, workspace_id="refresh-scope")) + with sessions() as db: + added = db.get(WorkbookRow, 202) + assert {cid: added.enrichments[cid]["value"] for cid in ("formula", "http", "agent", "research")} == { + "formula": "new@example.test", "http": "HTTP", "agent": "AGENT", "research": "RESEARCH"} + assert not db.get(WorkbookRow, 303).enrichments + assert not db.get(WorkbookRow, 101).enrichments + assert result == {"sourced": 1, "reenriched": 4, "stale_rows": 1} + assert sent == calls == []