Committing exception-repair results: matches, unmatched_*, summary¶

For: implementing "after transactions are corrected, flip is_reconciled for the ones that resolved, and recalculate the summaries."

Short version: you can UPDATE and DELETE these tables directly. Nothing needs to be worked around, and no BigQuery behaviour has been disabled to make that true.

1. Why the streaming-buffer error happened, and what changed¶

BigQuery's streaming buffer is internal machinery — it can't be turned off. Rows written by the legacy streaming API (insert_rows_json, i.e. tabledata.insertAll) sit in it for ~30-90 minutes, and any UPDATE / DELETE / MERGE touching them fails with:

UPDATE or DELETE statement over table ... would affect rows in the streaming buffer,
which is not supported

As of commit f887ffec, matches / summary / unmatched_gl / unmatched_bank are written through the Storage Write API (gRPC) instead (shared/core/infrastructure/storage_write.py). Those rows still go through a buffer — get_table().streaming_buffer is still populated for them — but BigQuery permits DML against them immediately. The buffer was not removed; only the write API changed.

Verified against the dev dataset, on rows written seconds earlier:

PASS  UPDATE summary (recalculated metrics, in place)  -> 1 row
PASS  DELETE resolved rows from unmatched_gl           -> 2 rows
summary rows for session: 1        (no duplicate)
unmatched rows left:      1        (3 written, 2 resolved -> 1)

Which tables are mutable¶

Write path	Tables	DML on fresh rows
Load job	`bank_statements`, `gl_transactions`	always legal (buffer never involved)
Storage Write API (gRPC)	`matches`, `summary`, `unmatched_gl`, `unmatched_bank`	allowed
Legacy `insertAll`	`session_name_results`, `narration_name_cache`	blocked ~30-90 min

The last two are the only tables still on the legacy path, and nothing ever mutates them.

2. The commit sequence¶

Run these in this order. The order is load-bearing — see §3. If the resolution runs after the session has completed (it does), also read §7 — the BigQuery writes below are only half the job.

Step 1 — INSERT the new matches¶

New matches are always an INSERT, never an update. Go through the normal store path (ReconciliationRepository.store(...), reconciliation_repository.py:177) so the rows get their composite keys and the Storage Write routing.

This alone makes the reported numbers correct: get_reconciliation_totals (reconciliation_repository.py:936) does not read stored metrics, it derives them — matched = COUNT(DISTINCT m.gl_record_id / m.bank_record_id) from matches, unmatched = COUNT(DISTINCT record_id) from unmatched_* anti-joined against matches.

Step 2 — DELETE the resolved rows from `unmatched_gl` / `unmatched_bank`¶

One statement per table per session. BigQuery serializes mutating DML per table, so do not issue one statement per record.

DELETE FROM `{dataset}.unmatched_gl`
WHERE id         = @session_id
  AND partner_id = @partner_id
  AND tenant_id  = @tenant_id
  AND record_id IN UNNEST(@resolved_gl_ids)

Step 3 — UPDATE the `summary` row in place¶

UPDATE `{dataset}.summary`
SET total_matches = @total_matches,
    gl_match_rate = @gl_rate, bank_match_rate = @bank_rate,
    unmatched_gl  = @u_gl,    unmatched_bank  = @u_bank,
    updated_at    = CURRENT_TIMESTAMP()
WHERE id = @session_id AND composite_key = @composite_key

Never insert a second summary row for the same session+period. See §4.

Step 4 — flip the transaction flags¶

await repo.mark_transactions_as_processed(
    gl_record_ids, bank_record_ids,          # is_processed  — the pass covered these
    partner_id, tenant_id,
    matched_gl_ids=…, matched_bank_ids=…,    # is_reconciled — only the resolved ones
)

repository.py:3501. gl_transactions / bank_statements are load-job written, so this has never been affected by the buffer — it already runs right after ingestion. Note the two flags are deliberately distinct: is_processed = a pass ran over this row, is_reconciled = it genuinely matched.

3. Why insert-before-delete¶

Every intermediate state has to be consistent, because a crash between steps is possible.

Crash after step 1: the record has a match row, and every accounting query anti-joins matches, so the stale unmatched_* row is invisible. Numbers correct, cleanup merely deferred.
If you deleted first and crashed: the record is neither matched nor unmatched. It disappears from the accounting identity — silently wrong totals.

Same reasoning applies to the existing end-of-session recovery pass, which promotes late matches with INSERT only and no DELETE.

4. Do not write a second `summary` row¶

summary already holds one row per batch per session — same id, different composite_key per period range. A recalculated summary for the same session and period carries the same composite_key, so it is a true duplicate, and get_reconciliation_totals sums across a session's summary rows:

SUM(COALESCE(s.gl_transactions,   0)) AS gl_transactions,
SUM(COALESCE(s.bank_transactions, 0)) AS bank_transactions,
SUM(COALESCE(s.withheld_count,    0)) AS rolling_withheld_count,

Those are the denominators. Duplicate the row and they double, so the match rate halves. get_summaries (:521) would also return both rows, and get_unmatched_gl_for_window / get_unmatched_bank_for_window (:622, :689) join summary s ON s.id = u.id with no DISTINCT, which would multiply carried-forward records.

Update in place, or don't write at all.

5. Handling a DML failure¶

Three cases can still bounce a DELETE/UPDATE off the buffer:

rows written before f887ffec reached that environment (within their ~90 min window),
an insertAll fallback — try_append_rows returns None on transport failure and the row goes out the legacy way; it logs at ERROR,
BQ_STORAGE_WRITE_ENABLED = false (kill switch, shared/config/appconfig.py).

Catch these with the existing detector rather than a string match of your own:

from shared.core.repositories.banking.repository import is_streaming_buffer_error   # :23

A failed cleanup is cosmetic, not incorrect — the anti-join keeps the reported numbers right until a retry succeeds. So the correct handling is: log, leave the row, retry later. Never let it abort the run.

6. Gotchas in the current code¶

a) Readers that will still see a resolved record. Several unmatched_* readers neither anti-join matches nor filter is_reconciled, so between step 1 and step 2 — and permanently, if step 2 fails — they will re-read a record that is already resolved. Notably the exception service's own: fetch_unmatched_record_ids (exception_service/banking/repository.py:228), rules/bank_fee.py:100,129, rules/extraction_issue.py:41,57, signals/signal_builder.py:210,236. Consequence is re-clustering an exception you already fixed. Worth adding the anti-join to any query that means currently unmatched.

b) The carried-forward fetchers use is_reconciled, not the anti-join. get_unmatched_gl_for_window / ..._bank_... filter COALESCE(u.is_reconciled, FALSE) = FALSE and do not anti-join matches. If you rely on step 2's DELETE this is moot; if a delete fails, the record gets re-offered next period. Adding NOT EXISTS (SELECT 1 FROM matches …) there makes it robust either way.

c) is_streaming_buffered() is currently a false positive. exception_service/banking/repository.py:207 gates the whole Gemini repair pre-pass on get_table().streaming_buffer is not None. That field is still populated for Storage Write rows even though the UPDATE on them succeeds — measured. So the guard will skip the repair pass and log unmatched_gl still in streaming buffer — skipping repair pre-pass, on a write that would now go through. It is also table-level, so one unrelated legacy write vetoes repair for every session on that table. Left as-is for now, deliberately — but if the repair pre-pass appears never to run, this is why.

7. This runs AFTER the session completes — what that adds¶

The exception pass is deliberately the last step of a session, and it runs after the results have already gone downstream:

routes/reconciliation.py:1006   _push_to_loader(...)                  ← software team gets the numbers
routes/reconciliation.py:1011   create_task(summary_task_id, "completed")
routes/reconciliation.py:1029   ExceptionService().run(...)           ← your repair happens here

The Dataflow poller path publishes the same way (:1358). So the BigQuery sequence in §2 works unchanged, but BigQuery and the software side will disagree until you re-publish. Three things to add.

a) Re-send the aggregate stats¶

_push_to_loader pushes ai_statistics via loader_client.update_reconciliation(...) — totalMatchedGlRecords, totalMatchedStatementRecords, totalGlRecords, totalStatementRecords (routes/reconciliation.py:1191). Those are now stale. Recompute with get_reconciliation_totals (reconciliation_repository.py:936 — it derives live from matches, so it already reflects your inserts) and call update_reconciliation again.

b) Re-publish the newly matched records¶

_publisher.publish_matches(...) for the resolved pairs only. This is safe to repeat: the loader upserts by bank_record_id (see the comment at reconciliation_repository.py:1051), so re-publishing updates rather than duplicates.

c) Confirm the `session_end` sentinel semantics — do not skip this¶

publish_session_end has already fired for this session. Its contract is explicit:

Publish the terminal isLastRecord=True sentinel … Must be called once after ALL batches are published.

So the software consumer has already been told the session is over. Before shipping, confirm with whoever owns that consumer which of these applies:

it accepts records arriving after the sentinel → publish the late matches and you are done; or
it does not → the late batch needs its own sentinel re-emitted, with a fresh last_record from the newly published set.

This is the difference between the corrections reaching the software team and being silently dropped. Everything else in this document is verifiable from our side; this one is not.

d) Unmatched needs no re-publish¶

The loader reads unmatched straight from BigQuery — "unmatched_gl": [] # Loader reads from BQ directly (routes/reconciliation.py:1353). So the §2 step-2 DELETE is visible downstream on its own.

e) `summary` is per-batch, not per-session¶

A fanned-out session has one summary row per batch — same id, different composite_key. So "update the summary row" means attributing each resolved record back to its batch's composite_key. Unless you need the stored copy fresh for the UI, the simpler answer is to leave the metric columns alone: get_reconciliation_totals derives them, and only get_summaries / get_summary serve the stored values.

f) Do not count on the buffer having flushed¶

Running "after the session" does not mean waiting ~90 minutes — the exception pass runs seconds after publish. Storage Write is what makes the DML legal here, not elapsed time.

g) A later OVERWRITE re-run discards these matches¶

Re-running the period with on_existing_action=overwrite wipes that generation, including exception-added matches. That is correct — the fresh run recomputes from the corrected data — but worth knowing before you debug a "disappearing match".

8. Verifying it locally¶

Local ADC expires; build the client from the service account in the root .env:

from google.cloud import bigquery as bq
from shared.config.appconfig import settings as env_config
from shared.config.gcp_credentials import resolve_credentials
client = bq.Client(project=env_config.PROJECT_ID, credentials=resolve_credentials())
# DATASET_ID is already project-qualified

Run with PYTHONPATH=shared/src — shared is installed non-editable in each service's .venv, so make install-shared is required for local edits to take effect at runtime.

Post-change, assert exactly this: one summary row per (session, composite_key), zero unmatched_* rows for records that have a matches row, and get_reconciliation_totals returning the same figures the UI shows.