On one engagement, operations and finance relied on a SQL Server data warehouse that brought manufacturing and inventory data across from SAP every night – production volumes, inventory movements and material consumption for several manufacturing plants. The loads ran in SQL Server Integration Services, and every morning the dashboard showed every job as completed successfully. Then a plant controller noticed something odd. According to the warehouse, raw material consumption had stayed almost perfectly flat for four days in a row, while production volumes had gone up substantially.
We picked one goods issue that had been posted in SAP the evening before, and traced it from SAP toward the warehouse. It was there in SAP's material documents. It wasn't in the warehouse. The extraction log said the job had completed successfully – and transferred zero records. As it turned out, the SAP team had recently moved the extraction to a new technical user as part of a security cleanup. That account could still log in and call the RFC function the integration used, but it was no longer authorized to read material documents for two of the plants. The custom RFC wrapper caught the authorization error, wrote it to an application log, and handed SSIS an empty result. As far as SSIS knew, it had called the function, received a valid answer and loaded zero rows. Monitoring only checked whether the package had run, so everything stayed green.
Monitoring on most integrations answers one question: did the job run? Finance wants the answer to a different question – do the two systems agree? – and those are not the same thing at all. A sync can finish successfully every night for a month while a validation rule in the target quietly rejects one record in five hundred, and nothing in the logs will look wrong. To be honest, the only way I know to be sure is to compare the systems directly, on a schedule, and to treat every difference as something somebody has to explain.
In this article, I'll walk through the reconciliation job I would build for a typical case: orders or payments that start life in a source system and get copied into a target, such as an ERP or a ledger. The code uses PostgreSQL 14 or later as the reconciliation store, and Python 3.11 with psycopg 3. Neither the source nor the target has to be Postgres. All they have to be is readable.
Snapshots first, comparison second
The first decision is to stop comparing live systems against each other. If you query both sides at the moment of comparison, the answer depends on whatever happened while the queries were running, and there's no way to go back and re-examine a run afterward. So each run takes a keyed snapshot of both sides – a copy of the records, with their keys, as of that moment – into staging tables, and every step after that works only on the snapshots.
CREATE TABLE recon_run (
run_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
as_of timestamptz NOT NULL,
settle_window interval NOT NULL DEFAULT '2 hours',
finished_at timestamptz
);
CREATE TABLE recon_snapshot (
run_id bigint NOT NULL REFERENCES recon_run,
side text NOT NULL CHECK (side IN ('source', 'target')),
record_key text NOT NULL,
amount numeric(18,4) NOT NULL,
currency char(3) NOT NULL,
status text,
event_at timestamptz NOT NULL, -- source event time, carried into the target
updated_at timestamptz NOT NULL
);
CREATE INDEX ON recon_snapshot (run_id, side, record_key);
CREATE TABLE recon_tolerance (
currency char(3) PRIMARY KEY,
minor_units int NOT NULL, -- 2 for USD, 0 for JPY
amount_tol numeric(18,4) NOT NULL DEFAULT 0
);
CREATE TABLE recon_result (
run_id bigint NOT NULL REFERENCES recon_run,
record_key text NOT NULL,
outcome text NOT NULL,
diff jsonb,
PRIMARY KEY (run_id, record_key, outcome)
);
You'll notice there's no unique constraint on (run_id, side, record_key). That's deliberate. If a key appears twice on one side, I want to know about it – it's a finding in its own right – and I want the job to report it and keep going, instead of crashing on a load error.
The whole design depends on record_key being an identifier both systems share – usually the source's ID, stored on the target record as an external ID. If the integration was built without one, adding it is the first piece of work. You can make fuzzy matching on dates and amounts work, after a fashion, but nobody ever fully trusts what it produces, and a reconciliation report that nobody trusts isn't worth running.
Extraction and normalization
Extraction streams rows into the snapshot with COPY, which is far faster than inserting rows one at a time once you're past a few thousand records. This is also the place for normalization. If the source says paid and the ERP says POSTED, map both to one vocabulary on the way in. The comparison SQL then compares like with like, and the mapping lives in one place, where someone can review it.
from collections.abc import Iterable, Iterator
from datetime import datetime
import psycopg
STATUS_MAP = {"paid": "posted", "refunded": "reversed", "pending": "open"}
COLUMNS = "run_id, side, record_key, amount, currency, status, event_at, updated_at"
def extract_source(src: psycopg.Connection, start: datetime, end: datetime) -> Iterator[tuple]:
# Server-side cursor so a large window streams instead of loading into memory.
with src.transaction(), src.cursor(name="recon_extract") as cur:
cur.execute(
"""SELECT order_id::text, total, currency, status, created_at, updated_at
FROM orders WHERE created_at >= %s AND created_at < %s""",
(start, end),
)
for key, total, ccy, status, created, updated in cur:
yield key, total, ccy, STATUS_MAP.get(status, status), created, updated
def load_snapshot(conn: psycopg.Connection, run_id: int, side: str,
rows: Iterable[tuple]) -> int:
n = 0
with conn.transaction(), conn.cursor() as cur:
cur.execute("DELETE FROM recon_snapshot WHERE run_id = %s AND side = %s",
(run_id, side))
with cur.copy(f"COPY recon_snapshot ({COLUMNS}) FROM STDIN") as copy:
for row in rows:
copy.write_row((run_id, side, *row))
n += 1
return n
The target extractor has the same shape, with one important difference: it filters on the source event time that the integration carried across, and not on the target's own creation timestamp. That sounds like a small detail. It isn't. If you cut the source by created_at and the target by the time the ERP inserted the row, every record near the edge of the window will show up as missing on one side or the other, and you'll spend your mornings explaining breaks that aren't really there. Both sides have to be cut by the same clock.
I use a rolling lookback window – typically seven days – instead of reconciling all of history every night. Older periods are still covered, because each night's run overlaps the previous six, and at month-end you can run a wider window with the same code.
The FULL OUTER JOIN that classifies everything
Duplicates get handled first. If a key has two rows on one side, there's no way to match it without guessing which copy is the real one, and I don't want the job guessing.
INSERT INTO recon_result (run_id, record_key, outcome, diff)
SELECT run_id, record_key, 'duplicate_in_' || side, jsonb_build_object('copies', count(*))
FROM recon_snapshot
WHERE run_id = %(run)s
GROUP BY run_id, side, record_key
HAVING count(*) > 1;
Everything else goes through a single statement. A FULL OUTER JOIN keeps rows from both sides whether or not they find a match, so one pass produces every outcome – missing in target, missing in source, mismatched, matched, and in flight. The field-level differences are built in a lateral subquery as a JSON object that holds only the fields that disagree, each as a [source, target] pair. If you haven't used FULL OUTER JOIN much, this is the job it was made for.
INSERT INTO recon_result (run_id, record_key, outcome, diff)
SELECT r.run_id,
coalesce(s.record_key, t.record_key),
CASE
WHEN t.record_key IS NULL AND s.updated_at > r.as_of - r.settle_window THEN 'in_flight'
WHEN t.record_key IS NULL THEN 'missing_in_target'
WHEN s.record_key IS NULL THEN 'missing_in_source'
WHEN d.diff = '{}'::jsonb THEN 'matched'
WHEN s.updated_at > r.as_of - r.settle_window THEN 'in_flight'
ELSE 'mismatched'
END,
CASE WHEN s.record_key IS NOT NULL AND t.record_key IS NOT NULL
AND d.diff <> '{}'::jsonb THEN d.diff END
FROM (SELECT * FROM recon_snapshot WHERE run_id = %(run)s AND side = 'source') s
FULL OUTER JOIN
(SELECT * FROM recon_snapshot WHERE run_id = %(run)s AND side = 'target') t
ON s.record_key = t.record_key
JOIN recon_run r ON r.run_id = %(run)s
LEFT JOIN recon_tolerance tol ON tol.currency = coalesce(s.currency, t.currency)
CROSS JOIN LATERAL (
SELECT jsonb_strip_nulls(jsonb_build_object(
'amount', CASE WHEN abs(round(s.amount, coalesce(tol.minor_units, 2))
- round(t.amount, coalesce(tol.minor_units, 2)))
> coalesce(tol.amount_tol, 0)
THEN jsonb_build_array(s.amount, t.amount) END,
'currency', CASE WHEN s.currency IS DISTINCT FROM t.currency
THEN jsonb_build_array(s.currency, t.currency) END,
'status', CASE WHEN s.status IS DISTINCT FROM t.status
THEN jsonb_build_array(s.status, t.status) END,
'event_at', CASE WHEN abs(extract(epoch FROM s.event_at - t.event_at)) > 120
THEN jsonb_build_array(s.event_at, t.event_at) END
)) AS diff
) d
WHERE NOT EXISTS (
SELECT 1 FROM recon_result x
WHERE x.run_id = %(run)s AND x.record_key = coalesce(s.record_key, t.record_key)
);
Most of the judgment in this job lives in the tolerances. Amounts are compared after both sides are rounded to the currency's minor units (cents, for dollars). Without that, a source that keeps four decimal places and an ERP that keeps two will disagree on almost every line. I leave amount_tol at zero by default, and raise it to one minor unit only for a target I know rounds differently – for example, one that uses banker's rounding, where Postgres's round() rounds half away from zero. A blanket tolerance of a few cents feels harmless, right up until someone notices it has been quietly absorbing a systematic fee error. Timestamps get two minutes of leeway, because targets often truncate to whole seconds, or store local time with a lossy conversion. Anything beyond two minutes is a real difference, and I want to see it.
The settle window takes care of records that simply haven't arrived yet. If a source record was created or updated in the last two hours, and it's missing or different in the target, it's classified as in_flight and isn't treated as a break. It stays visible in the results, but nobody gets paged. If it's still wrong on the next run, after the window has passed, it becomes a real break. Of everything in this design, I'd say this distinction does the most for the credibility of the report. A report that cries wolf on every record still in transit teaches people to stop reading it.
Control totals by day
Matching record by record does most of the work, but control totals are cheap, and they catch a different kind of failure – a whole batch loaded twice, say, or posted to the wrong period.
WITH daily AS (
SELECT snap.side, (snap.event_at AT TIME ZONE 'America/New_York')::date AS day,
snap.currency, count(*) AS n,
sum(round(snap.amount, coalesce(tol.minor_units, 2))) AS total
FROM recon_snapshot snap
LEFT JOIN recon_tolerance tol ON tol.currency = snap.currency
WHERE snap.run_id = %(run)s
GROUP BY 1, 2, 3
)
SELECT coalesce(s.day, t.day) AS day, coalesce(s.currency, t.currency) AS currency,
s.n AS source_count, t.n AS target_count,
coalesce(s.total, 0) - coalesce(t.total, 0) AS difference
FROM (SELECT * FROM daily WHERE side = 'source') s
FULL OUTER JOIN (SELECT * FROM daily WHERE side = 'target') t
ON s.day = t.day AND s.currency = t.currency
WHERE s.n IS DISTINCT FROM t.n OR s.total IS DISTINCT FROM t.total
ORDER BY 1, 2;
Amounts are rounded the same way here as in the record matching, so precision differences don't turn up as breaks in the totals. Group by the business day in the time zone where finance closes the books, and not by the UTC date. Otherwise, every evening's orders will straddle two days. And where the target can produce its own totals – a ledger balance by day, for instance – compare against those, and not against your own extract. A total calculated from your own snapshot can't catch a bug in your own extraction.
Breaks that persist until someone works them
A list of differences rebuilt from scratch every night has no memory. You can't tell whether a break appeared last night or three weeks ago, and there's nowhere to record who's responsible for it. So the results feed a recon_break table that persists from one run to the next.
CREATE TABLE recon_break (
break_id bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
record_key text NOT NULL,
outcome text NOT NULL,
diff jsonb,
first_seen timestamptz NOT NULL,
last_run_id bigint NOT NULL,
owner text,
status text NOT NULL DEFAULT 'open' CHECK (status IN ('open', 'cleared', 'resolved')),
resolution text
);
CREATE UNIQUE INDEX recon_break_open ON recon_break (record_key, outcome) WHERE status = 'open';
INSERT INTO recon_break (record_key, outcome, diff, first_seen, last_run_id, owner)
SELECT res.record_key, res.outcome, res.diff, run.as_of, res.run_id, o.owner
FROM recon_result res
JOIN recon_run run ON run.run_id = res.run_id
LEFT JOIN recon_owner o ON o.outcome = res.outcome
WHERE res.run_id = %(run)s AND res.outcome NOT IN ('matched', 'in_flight')
ON CONFLICT (record_key, outcome) WHERE status = 'open'
DO UPDATE SET diff = EXCLUDED.diff, last_run_id = EXCLUDED.last_run_id;
UPDATE recon_break b SET status = 'cleared'
WHERE b.status = 'open' AND b.last_run_id < %(run)s
AND EXISTS (SELECT 1 FROM recon_result x
WHERE x.run_id = %(run)s AND x.record_key = b.record_key
AND x.outcome <> 'in_flight');
recon_owner is a small lookup table that maps each outcome to a team – missing records go to operations, amount mismatches go to finance. The clearing rule is cautious on purpose. A break clears only when this run actually saw the key again, with a different outcome. A key that has aged out of the lookback window hasn't been fixed. It has just stopped being looked at, so it stays open. Better still, have the extractors also pull every key with an open break, however old it is.
The orchestration runs classification and break maintenance in one transaction, under an advisory lock, so two overlapping runs can't clear each other's breaks. Connections are opened with autocommit=True, so the explicit transaction blocks are the only transactions there are.
from datetime import datetime, timedelta
# The four SQL statements shown above, held as module-level strings.
RECON_LOCK = 724_001
STEPS = (DUPLICATES_SQL, CLASSIFY_SQL, UPSERT_BREAKS_SQL, CLEAR_BREAKS_SQL)
def run_recon(conn, src, tgt, as_of: datetime, lookback: timedelta) -> int:
run_id = conn.execute(
"INSERT INTO recon_run (as_of) VALUES (%s) RETURNING run_id", (as_of,)
).fetchone()[0]
start = as_of - lookback
n_src = load_snapshot(conn, run_id, "source", extract_source(src, start, as_of))
n_tgt = load_snapshot(conn, run_id, "target", extract_target(tgt, start, as_of))
if n_src == 0 or n_tgt == 0:
raise RuntimeError(f"run {run_id}: empty extract (source={n_src}, target={n_tgt})")
with conn.transaction():
conn.execute("SELECT pg_advisory_xact_lock(%s)", (RECON_LOCK,))
for sql in STEPS:
conn.execute(sql, {"run": run_id})
conn.execute("UPDATE recon_run SET finished_at = now() WHERE run_id = %s", (run_id,))
return run_id
The empty-extract guard looks almost too simple to bother with. To be honest, it's the check I'd least want to lose. Two empty snapshots reconcile perfectly! An expired credential, a renamed table or a service account that has quietly lost an authorization can produce a clean, green report for a system that hasn't been read at all.
That last one is exactly what happened in the SAP case at the top of this article. Restoring the authorization, re-extracting the missing movements and reconciling the warehouse against SAP by plant and posting date fixed the data. It didn't fix the reason nobody had noticed. So we changed the integration contract: an authorization failure could never again come back as an empty, successful extract – the wrapper had to return a structured error, or fail the call. We also split monitoring into three separate questions. Did the job run? Is the data fresh, meaning have we received transactions with recent posting dates? And is it complete, meaning do counts and quantities agree with SAP by plant and posting date? Finally, we added an alert for abnormal zero-record extracts. One plant can have a quiet day. Several active plants with no material movements at all is something a person should look at.
Reporting so breaks get worked
The alert should be about what changed and what's getting old, and not about how many breaks there are. A message listing four hundred open breaks every morning will be ignored by the second week. Compare that with a message saying three new amount mismatches appeared overnight, and two missing-in-target breaks owned by operations are now past their 24-hour target. That one gives a specific person a specific job.
import httpx
def alert(conn, run_id: int, webhook_url: str, sla_hours: int = 24) -> None:
rows = conn.execute(
"""WITH run AS (SELECT as_of FROM recon_run WHERE run_id = %(run)s),
summary AS (
SELECT b.outcome, coalesce(b.owner, 'unassigned') AS owner,
count(*) FILTER (WHERE b.first_seen = run.as_of) AS new,
count(*) FILTER (WHERE b.first_seen
< run.as_of - %(sla)s * interval '1 hour') AS aged
FROM recon_break b CROSS JOIN run
WHERE b.status = 'open'
GROUP BY 1, 2)
SELECT outcome, owner, new, aged FROM summary
WHERE new > 0 OR aged > 0 ORDER BY aged DESC""",
{"run": run_id, "sla": sla_hours},
).fetchall()
if not rows:
return
lines = [f"{o} ({owner}): {new} new, {aged} past {sla_hours}h" for o, owner, new, aged in rows]
httpx.post(webhook_url, json={"text": f"Reconciliation run {run_id}\n" + "\n".join(lines)},
timeout=10).raise_for_status()
The payload is the plain text form that Slack incoming webhooks accept, and any chat or ticketing endpoint works the same way. When someone resolves a break, they set status = 'resolved' and write what they did in resolution. There's one rule I hold teams to: a break gets fixed where it started, by correcting the source or the integration – never by editing the target until the numbers line up. Hand-editing the target turns tonight's report green, and it destroys the evidence of why it was red in the first place.
Over time, the resolution column turns into the most useful data in the whole system. If the same cause shows up every week, that's a defect in the integration, and the reconciliation job has just told you exactly where to look. The job proves agreement record by record. The process around it – who signs off, what goes in the period report, how timing items are presented – is a separate design problem, and I've written about it in more depth in a piece on designing reconciliation reports finance will sign off against.
I'm Jayant Chaudhary, founder of eProxim, where we build integrations that carry money and inventory between systems, and reconciliation is part of every one.
Top comments (1)
The "job succeeded, zero records transferred" story is painfully familiar. One cheap addition I've found useful next to the key-level diff: a control-totals check per natural slice (here, per plant per day) comparing COUNT(*) and SUM(amount) on both snapshots. It runs in seconds, flags the "one plant silently returns nothing" case before the detailed comparison even finishes, and gives finance a one-line summary they can actually read. I also liked the deliberate choice not to put a unique constraint on (run_id, side, record_key). Duplicate keys on the target side are often the first visible sign of a retry that wasn't idempotent, so surfacing them as findings instead of load errors is the right call.