feat(sp27): unify 835 ingest + reconciliation in handle_835 (atomic)
Move 'reconcile.run(s, record.id)' inside CycloneStore.add()'s ingest session, before s.commit(). The placeholder adjustment_amount set by _remittance_835_row is overwritten by reconcile's CAS-aggregate pass in the same transaction — readers never see a half-reconciled Remittance row. If reconcile raises, the entire 835 ingest rolls back via the session's __exit__. The previous flow committed batch + remittance rows in session-1, then opened session-2 to run reconcile.fail-soft. Two visible problems closed: a race window where readers fetched the placeholder adjustment_amount, and a half-reconciled state left visible if reconcile crashed. Deviation from the plan (N4 in autoreview): the reconcile call lives in cycl_store.add() rather than handle_835 calling a new reconcile.run_now(batch_id) helper after add(). Same end state, one fewer module surface, handler stays a thin wrapper over the store.
This commit is contained in:
@@ -519,12 +519,13 @@ def _build_and_persist_ack(batch_id: str) -> dict | None:
|
|||||||
def _reconciliation_summary_for_batch(batch_id: str) -> dict:
|
def _reconciliation_summary_for_batch(batch_id: str) -> dict:
|
||||||
"""Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch.
|
"""Return ``{matched, unmatched_claims, unmatched_remittances, skipped}`` for a batch.
|
||||||
|
|
||||||
Reads from the DB after ``store.add()`` has already triggered T10
|
Reads from the DB after ``store.add()`` has already run reconciliation
|
||||||
reconciliation synchronously (via ``_run_reconcile``). Counts are
|
synchronously (SP27 Task 10: ``reconcile.run(s, batch_id)`` inside the
|
||||||
observed at this moment; a subsequent manual match/unmatch will not
|
ingest session, before commit). Counts are observed at this moment;
|
||||||
be reflected until the next request.
|
a subsequent manual match/unmatch will not be reflected until the
|
||||||
|
next request.
|
||||||
|
|
||||||
``skipped`` is reserved for future use — the T10 orchestrator tracks
|
``skipped`` is reserved for future use — the orchestrator tracks
|
||||||
skipped claims internally but does not surface a queryable count.
|
skipped claims internally but does not surface a queryable count.
|
||||||
"""
|
"""
|
||||||
from sqlalchemy import func, select
|
from sqlalchemy import func, select
|
||||||
|
|||||||
@@ -15,10 +15,11 @@ but moving it is out of Task 5 scope). We import it lazily inside
|
|||||||
acceptable because api.py also imports scheduler lazily (inside its
|
acceptable because api.py also imports scheduler lazily (inside its
|
||||||
lifespan handler).
|
lifespan handler).
|
||||||
|
|
||||||
The two-phase ingest problem (batch row first, then a separate
|
Two-phase ingest closed in SP27 Task 10: the placeholder
|
||||||
``reconcile`` pass that overwrites ``adjustment_amount``) is a known
|
``adjustment_amount`` is overwritten by reconcile in the same DB
|
||||||
gap; atomic unification happens in SP27 Task 10. For Task 5 we lock
|
session as the insert (before commit), so a reader never sees a
|
||||||
current behavior.
|
half-reconciled Remittance row. If reconcile raises, the whole
|
||||||
|
ingest rolls back and the scheduler records the per-file error.
|
||||||
|
|
||||||
``event_bus`` is best-effort and follows the same TODO(sp27-task-6)
|
``event_bus`` is best-effort and follows the same TODO(sp27-task-6)
|
||||||
sync/async gap as handle_999 / handle_ta1 / handle_277ca.
|
sync/async gap as handle_999 / handle_ta1 / handle_277ca.
|
||||||
|
|||||||
@@ -224,10 +224,15 @@ class ReconcileResult:
|
|||||||
def run(session, batch_id: str) -> ReconcileResult:
|
def run(session, batch_id: str) -> ReconcileResult:
|
||||||
"""Orchestrate reconciliation for an 835 batch.
|
"""Orchestrate reconciliation for an 835 batch.
|
||||||
|
|
||||||
Expects Batch + Remittance + CasAdjustment rows already persisted
|
Expects Batch + Remittance + CasAdjustment rows already added to
|
||||||
(this is called from store.add() AFTER commit). Loads the new
|
``session`` (not yet committed). SP27 Task 10 calls this from
|
||||||
remittances + all currently-unmatched Claims, calls match(), then
|
inside ``CycloneStore.add``'s ingest session, BEFORE
|
||||||
applies each match's intent inside the caller's session.
|
``s.commit()`` — so the whole 835 ingest (rows + reconcile) is
|
||||||
|
a single transaction. If anything here raises, the ingest rolls
|
||||||
|
back; no half-reconciled batch ever lands.
|
||||||
|
|
||||||
|
Loads the new remittances + all currently-unmatched Claims, calls
|
||||||
|
match(), then applies each match's intent inside the caller's session.
|
||||||
|
|
||||||
Returns counts. Does NOT commit; caller controls transaction.
|
Returns counts. Does NOT commit; caller controls transaction.
|
||||||
Raises on failure (caller catches and writes activity event).
|
Raises on failure (caller catches and writes activity event).
|
||||||
|
|||||||
@@ -912,9 +912,12 @@ class CycloneStore:
|
|||||||
|
|
||||||
For 835 batches: inserts the Batch row, one Remittance row per
|
For 835 batches: inserts the Batch row, one Remittance row per
|
||||||
ClaimPayment, and a ``remit_received`` ActivityEvent per
|
ClaimPayment, and a ``remit_received`` ActivityEvent per
|
||||||
ClaimPayment. After commit, calls ``_run_reconcile`` (T10 stub)
|
ClaimPayment. Reconciliation (auto-match + per-pair CAS
|
||||||
in a fail-soft manner — reconciliation errors are logged but
|
aggregate) runs IN THE SAME SESSION before commit (SP27
|
||||||
do not roll back the persisted batch.
|
Task 10) — a single ``s.commit()`` covers both ingest and
|
||||||
|
reconciliation. If reconcile raises, the whole 835 ingest
|
||||||
|
rolls back; the batch never appears half-reconciled with
|
||||||
|
placeholder ``adjustment_amount`` values.
|
||||||
|
|
||||||
Idempotency: ``Claim.id`` and ``Remittance.id`` are PRIMARY KEYS,
|
Idempotency: ``Claim.id`` and ``Remittance.id`` are PRIMARY KEYS,
|
||||||
so a re-ingest of the same fixture (e.g. ``/api/parse-837`` called
|
so a re-ingest of the same fixture (e.g. ``/api/parse-837`` called
|
||||||
@@ -1019,19 +1022,17 @@ class CycloneStore:
|
|||||||
f"Unsupported BatchRecord subclass: {type(record).__name__}"
|
f"Unsupported BatchRecord subclass: {type(record).__name__}"
|
||||||
)
|
)
|
||||||
|
|
||||||
s.commit()
|
# SP27 Task 10: run reconcile INSIDE the same session, before
|
||||||
|
# commit. The placeholder ``adjustment_amount`` set by
|
||||||
|
# ``_remittance_835_row`` is overwritten by reconcile's
|
||||||
|
# ``_reconcile_pair`` SUM-of-CAS-aggregate pass before the
|
||||||
|
# rows are ever visible. If reconcile raises, the whole
|
||||||
|
# 835 ingest rolls back and the batch never lands.
|
||||||
|
if record.kind == "835":
|
||||||
|
from cyclone import reconcile as _reconcile
|
||||||
|
_reconcile.run(s, record.id)
|
||||||
|
|
||||||
# Reconcile 835 batches after the batch is durably persisted.
|
s.commit()
|
||||||
# Fail-soft: errors are logged, not raised, so an 835 parse that
|
|
||||||
# crashes in reconciliation still shows up in /api/batches.
|
|
||||||
if record.kind == "835":
|
|
||||||
try:
|
|
||||||
self._run_reconcile(record.id)
|
|
||||||
except Exception: # pragma: no cover - logged via default handler
|
|
||||||
import logging
|
|
||||||
logging.getLogger(__name__).exception(
|
|
||||||
"reconcile.run failed for batch %s", record.id,
|
|
||||||
)
|
|
||||||
|
|
||||||
# Publish live-tail events synchronously. EventBus.publish is async
|
# Publish live-tail events synchronously. EventBus.publish is async
|
||||||
# but its body is purely synchronous ``put_nowait`` enqueues; we
|
# but its body is purely synchronous ``put_nowait`` enqueues; we
|
||||||
@@ -1113,26 +1114,6 @@ class CycloneStore:
|
|||||||
for queue in list(event_bus._subscribers.get(kind, ())):
|
for queue in list(event_bus._subscribers.get(kind, ())):
|
||||||
event_bus._enqueue_or_drop_oldest(queue, event)
|
event_bus._enqueue_or_drop_oldest(queue, event)
|
||||||
|
|
||||||
def _run_reconcile(self, batch_id: str) -> None:
|
|
||||||
"""T10 stub: invoke the reconcile orchestrator for a batch.
|
|
||||||
|
|
||||||
The actual reconciliation is implemented in T10 (this method
|
|
||||||
will import ``cyclone.reconcile`` and call ``reconcile.run``
|
|
||||||
with the current session). For T9 we keep the import lazy and
|
|
||||||
fail-soft so a missing or NotImplementedError reconcile module
|
|
||||||
never breaks the 835 ingest path.
|
|
||||||
"""
|
|
||||||
# T10 will replace this with the real implementation. Until then
|
|
||||||
# we accept the failure modes the spec lists: ImportError
|
|
||||||
# (module not yet wired), NotImplementedError (stub in place),
|
|
||||||
# or any other transient reconcile error — all swallowed.
|
|
||||||
try:
|
|
||||||
from cyclone import reconcile as _reconcile
|
|
||||||
with db.SessionLocal()() as s:
|
|
||||||
_reconcile.run(s, batch_id)
|
|
||||||
except (ImportError, NotImplementedError):
|
|
||||||
pass
|
|
||||||
|
|
||||||
# -- read path ------------------------------------------------------
|
# -- read path ------------------------------------------------------
|
||||||
|
|
||||||
def _row_to_record(self, row: Batch) -> BatchRecord:
|
def _row_to_record(self, row: Batch) -> BatchRecord:
|
||||||
|
|||||||
@@ -1,14 +1,17 @@
|
|||||||
"""Direct tests for the ``handle_835`` handler (SP27 Task 5).
|
"""Direct tests for the ``handle_835`` handler (SP27 Task 5 + Task 10).
|
||||||
|
|
||||||
Locks the handler's contract independent of the scheduler lifecycle
|
Locks the handler's contract independent of the scheduler lifecycle
|
||||||
so a regression in the scheduler wiring doesn't hide a regression
|
so a regression in the scheduler wiring doesn't hide a regression
|
||||||
in the handler.
|
in the handler.
|
||||||
|
|
||||||
Note: the 835 handler is the largest of the four because the schema
|
Note: the 835 handler is the largest of the four because the schema
|
||||||
covers per-claim remittances + CAS adjustments + validation. The
|
covers per-claim remittances + CAS adjustments + validation. SP27
|
||||||
two-phase ingestion problem (batch row first, then a separate
|
Task 10 unified the two-phase ingest path (batch row first, then a
|
||||||
``reconcile`` pass) is a known gap; atomic unification happens in
|
separate ``reconcile`` pass that overwrote ``adjustment_amount``)
|
||||||
SP27 Task 10. For Task 5 we lock current behavior (no atomic).
|
into one critical section. These tests pin the atomicity invariants:
|
||||||
|
after ``handle()`` returns, the persisted ``adjustment_amount`` is
|
||||||
|
already the authoritative CAS sum; if reconcile raises, no batch
|
||||||
|
rows land.
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
@@ -102,3 +105,148 @@ def test_handle_835_raises_on_completely_unparseable_input():
|
|||||||
bad = "this is not even EDI"
|
bad = "this is not even EDI"
|
||||||
with pytest.raises((CycloneParseError, ValueError)):
|
with pytest.raises((CycloneParseError, ValueError)):
|
||||||
handle(bad, source_file="bad.835")
|
handle(bad, source_file="bad.835")
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# SP27 Task 10: atomicity invariants
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_handle_835_adjustment_amount_matches_cas_sum():
|
||||||
|
"""After ``handle()`` returns, every persisted Remittance's
|
||||||
|
``adjustment_amount`` equals the SUM of its CasAdjustment rows.
|
||||||
|
|
||||||
|
Pins the Task 10 atomicity fix: ingest + reconcile live in the
|
||||||
|
same DB session, so reconcile's CAS-aggregate write happens
|
||||||
|
BEFORE commit and the placeholder value never escapes. Before
|
||||||
|
Task 10, a reader could observe the placeholder
|
||||||
|
(``_remittance_835_row`` sums only the first service line's CAS)
|
||||||
|
while the second-phase ``reconcile.run`` was still pending.
|
||||||
|
"""
|
||||||
|
from decimal import Decimal
|
||||||
|
|
||||||
|
from sqlalchemy import func, select
|
||||||
|
|
||||||
|
from cyclone.db import CasAdjustment, Remittance
|
||||||
|
|
||||||
|
text = MINIMAL.read_text()
|
||||||
|
handle(text, source_file=MINIMAL.name)
|
||||||
|
|
||||||
|
with db.SessionLocal()() as session:
|
||||||
|
remits = session.execute(select(Remittance)).scalars().all()
|
||||||
|
assert len(remits) >= 1
|
||||||
|
for r in remits:
|
||||||
|
cas_sum = session.execute(
|
||||||
|
select(func.coalesce(func.sum(CasAdjustment.amount), 0))
|
||||||
|
.where(CasAdjustment.remittance_id == r.id)
|
||||||
|
).scalar_one()
|
||||||
|
assert r.adjustment_amount == Decimal(str(cas_sum)), (
|
||||||
|
f"remit {r.id}: adjustment_amount={r.adjustment_amount} "
|
||||||
|
f"!= CAS sum={cas_sum}"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_handle_835_reconcile_failure_rolls_back_ingest(monkeypatch):
|
||||||
|
"""If reconcile raises mid-ingest, the batch + remittance rows
|
||||||
|
don't land at all (atomic rollback, SP27 Task 10).
|
||||||
|
|
||||||
|
Before Task 10, ``store.add`` committed the batch first and then
|
||||||
|
ran reconcile in a separate session with fail-soft semantics —
|
||||||
|
a reconcile crash left the half-reconciled batch visible. After
|
||||||
|
Task 10, ``reconcile.run`` runs inside the ingest session, so
|
||||||
|
any reconcile exception rolls the whole transaction back.
|
||||||
|
"""
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
from cyclone import reconcile
|
||||||
|
from cyclone.db import Batch, Remittance
|
||||||
|
|
||||||
|
def boom(*a, **kw):
|
||||||
|
raise RuntimeError("simulated reconcile outage")
|
||||||
|
|
||||||
|
monkeypatch.setattr(reconcile, "match", boom)
|
||||||
|
|
||||||
|
text = MINIMAL.read_text()
|
||||||
|
with pytest.raises(RuntimeError, match="simulated reconcile outage"):
|
||||||
|
handle(text, source_file=MINIMAL.name)
|
||||||
|
|
||||||
|
with db.SessionLocal()() as session:
|
||||||
|
# No batch, no remittance — atomic rollback worked.
|
||||||
|
batches = session.execute(select(Batch)).scalars().all()
|
||||||
|
remits = session.execute(select(Remittance)).scalars().all()
|
||||||
|
assert batches == [], (
|
||||||
|
f"reconcile raised but {len(batches)} batch rows landed"
|
||||||
|
)
|
||||||
|
assert remits == [], (
|
||||||
|
f"reconcile raised but {len(remits)} remittance rows landed"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_handle_835_late_reconcile_failure_rolls_back_match_loop(monkeypatch):
|
||||||
|
"""A crash in the SECOND pipeline (``_reconcile_pair``) still
|
||||||
|
rolls back the FIRST pipeline's mutations — Claim.state,
|
||||||
|
Claim.matched_remittance_id, Remittance.claim_id, the new
|
||||||
|
ActivityEvent row, the new Match row.
|
||||||
|
|
||||||
|
Pins the atomic rollback across the whole reconcile pipeline,
|
||||||
|
not just the listing-error path. Without this, a future change
|
||||||
|
that splits ``reconcile.run`` into two sessions would quietly
|
||||||
|
re-introduce the same race Task 10 closed.
|
||||||
|
"""
|
||||||
|
from datetime import date
|
||||||
|
from decimal import Decimal
|
||||||
|
|
||||||
|
from sqlalchemy import select
|
||||||
|
|
||||||
|
from cyclone import reconcile
|
||||||
|
from cyclone.db import (
|
||||||
|
ActivityEvent, Batch, Claim, ClaimState, Match, Remittance,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Pre-seed a Claim that will auto-match the minimal_835 remit
|
||||||
|
# (``payer_claim_control_number = "CLM001"``). Without this,
|
||||||
|
# ``reconcile.match`` returns no matches and ``_reconcile_pair``
|
||||||
|
# is never called — the test would silently become a no-op.
|
||||||
|
with db.SessionLocal()() as s:
|
||||||
|
s.add(Claim(
|
||||||
|
id="CLM001",
|
||||||
|
batch_id="seed",
|
||||||
|
patient_control_number="CLM001",
|
||||||
|
service_date_from=date(2026, 6, 2),
|
||||||
|
charge_amount=Decimal("100.00"),
|
||||||
|
state=ClaimState.SUBMITTED,
|
||||||
|
))
|
||||||
|
s.commit()
|
||||||
|
|
||||||
|
def boom_pair(session, claim, remittance):
|
||||||
|
raise RuntimeError("simulated late-stage reconcile outage")
|
||||||
|
|
||||||
|
monkeypatch.setattr(reconcile, "_reconcile_pair", boom_pair)
|
||||||
|
|
||||||
|
text = MINIMAL.read_text()
|
||||||
|
with pytest.raises(RuntimeError, match="simulated late-stage reconcile outage"):
|
||||||
|
handle(text, source_file=MINIMAL.name)
|
||||||
|
|
||||||
|
with db.SessionLocal()() as session:
|
||||||
|
# Zero-state invariants — same as the early-failure test,
|
||||||
|
# but also pinning that the match-loop's side effects (Match
|
||||||
|
# row, ActivityEvent) didn't slip through the rollback.
|
||||||
|
batches = session.execute(select(Batch)).scalars().all()
|
||||||
|
remits = session.execute(select(Remittance)).scalars().all()
|
||||||
|
match_rows = session.execute(select(Match)).scalars().all()
|
||||||
|
activity_rows = session.execute(select(ActivityEvent)).scalars().all()
|
||||||
|
assert batches == []
|
||||||
|
assert remits == []
|
||||||
|
assert match_rows == [], (
|
||||||
|
f"_reconcile_pair raised but {len(match_rows)} Match rows landed"
|
||||||
|
)
|
||||||
|
assert activity_rows == [], (
|
||||||
|
f"_reconcile_pair raised but {len(activity_rows)} ActivityEvent rows landed"
|
||||||
|
)
|
||||||
|
|
||||||
|
# And the pre-seeded claim is unchanged — no matched_remittance_id,
|
||||||
|
# still in SUBMITTED state. The match-loop's mutations against
|
||||||
|
# this row were rolled back too.
|
||||||
|
claim = session.get(Claim, "CLM001")
|
||||||
|
assert claim.matched_remittance_id is None
|
||||||
|
assert claim.state == ClaimState.SUBMITTED
|
||||||
|
|||||||
@@ -330,8 +330,20 @@ def test_run_reversal_flips_paid_to_reversed(fixture_835):
|
|||||||
assert reversal_match.prior_claim_state == ClaimState.PAID
|
assert reversal_match.prior_claim_state == ClaimState.PAID
|
||||||
|
|
||||||
|
|
||||||
def test_run_failed_reconcile_writes_activity_event(fixture_835):
|
def test_run_reconcile_raise_in_session_leaves_prior_commits_alone(fixture_835):
|
||||||
"""If reconcile crashes, the batch + remittances stay; activity event records failure."""
|
"""A ``reconcile.run`` raise inside an open session does not damage
|
||||||
|
previously-committed rows.
|
||||||
|
|
||||||
|
The test seeds a batch + remit, commits them in one session, then
|
||||||
|
calls ``reconcile.run`` in a fresh session with ``match`` monkey-
|
||||||
|
patched to raise. The pre-existing rows must survive — they're on
|
||||||
|
disk from the prior commit, separate from the rolling-back
|
||||||
|
session. This pins the unit-level invariant that ``reconcile.run``
|
||||||
|
itself never commits and never silently mutates rows outside the
|
||||||
|
session it was given; it is the *caller's* responsibility (now
|
||||||
|
``CycloneStore.add`` per SP27 Task 10) to control the transaction
|
||||||
|
boundary.
|
||||||
|
"""
|
||||||
with db.SessionLocal()() as s:
|
with db.SessionLocal()() as s:
|
||||||
_make_batch(s)
|
_make_batch(s)
|
||||||
_make_remit(s, "CLP-1", "PCN-A", "1", "100.00", "100.00")
|
_make_remit(s, "CLP-1", "PCN-A", "1", "100.00", "100.00")
|
||||||
|
|||||||
Reference in New Issue
Block a user