Empty is not an error: reconciling derived data across stores

A reviewer opens the dashboard of reviewed conversations, filters to “first-tier reviewed” and gets an empty table. The API answered 200 OK, the search index behind it answered quickly, and the nightly job that feeds the index ran on schedule. Nothing raised an error, so nobody was paged, and the table stayed empty for 33 days.

The reviews live in a task store, a database with one row per completed review: a task id, the reviewer’s label, and usually a key pointing at the reviewed conversation. It is the source of truth (when two places disagree, it wins), but it is poor at the dashboard’s filtered, sorted query, so a search index keeps a copy arranged for it. The copy is derived data (elsewhere a cache or an analytics table): delete it and you can rebuild it from the source. A scheduled sync job keeps it in step. Call the copy the sink and the API that queries it the reader.

A toy night: Monday has five completed reviews, tasks 1 to 5. For each, the job finds the conversation’s document in the index by its key and writes the label onto it. Tasks 4 and 5 came from an upstream service that changed its id format that afternoon, so their lookups match nothing. The job counts a no-match as a skip (often right: some conversations were never indexed), logs status=success, updated=3, no_docs_matched=2, and the reader returns three rows with 200 OK. Every component is green. On Tuesday that service’s tasks miss again.

The gap between what the source says the reader should show and what it does show is drift. Monitoring watches for errors, and a lookup that matches nothing is a well-formed answer. Each component judges only its own step; nothing compares the end of the pipeline with its start.

Reconciliation does: on a schedule, per bucket (a day of reviews, say), it counts what the source says the reader should show and what the reader does show, compares a fingerprint of the contents, and alarms when they disagree. A run ledger records what each run of the job was handed and where every row ended up (written, no match, skipped for a stated reason, or failed). In a toy pipeline run for 40 simulated days with nine faults switched on one at a time, the three alarms most teams already have catch 4 of the 9. Reconciliation catches 8 and the ledger 6; together they catch all 9.

The figure is a pipe with a flow meter at each end and a status lamp wired only to the pump. With the leak valve a quarter open, the lamp stays green while the sink meter falls behind (click to close or reopen the leak), ending at source 635, sink 479, drift 156: day 20 of the bench’s vocabulary drift fault, where labels carry a tier value the reader’s filter does not know.

The model

Derived data has a conservation law

Kleppmann’s Designing Data-Intensive Applications argues for auditing derived data against its system of record (the source of truth) rather than trusting each component. For each bucket, let expected be the source rows that should reach the reader, computed from the source alone:

expected(day) = visible(day) + excluded(day)

Visible is what the reader’s own query returns; excluded is rows dropped on purpose, each with a counted reason (a task whose conversation was never indexed, for instance). Anything else is drift, in one of three directions: missing (in the source, not the reader), extra (in the reader, not the source) or mismatched (in both, with different content: a wrong label, or three copies of one).

Huang et al. call this a gray failure and name the property behind it differential observability: “the system’s failure detectors may not notice problems even when applications are afflicted by them”. In the incident below, the job’s status line is the failure detector and the dashboard is the afflicted application.

Reproduce it

A fault bench

The bench, experiments/empty-is-not-an-error/fault_matrix.py, simulates 40 days with 900 to 1,099 completed tasks a day. 39% of tasks have no join key, the (conversation, run) pair the job uses to find a document, so they can never reach the index. That is the field ratio: one backfill window (a one-off catch-up run) matched 78,602 tasks, of which 47,767 grouped into a pair, and the script computes 30,835 / 78,602 = 39.2%. Another 1.5% of keys have no document because another writer never indexed that conversation. The sync runs daily at 01:00, takes 60 ms per write, has a 900 s time limit and merges labels by task id. One fault switches on at the start of day 20.

The faults are the rows of the matrix below; a lost update is a second writer overwriting documents and erasing their labels. The first three detectors are the ones most teams already have.

detector	fires when
error rate	the job crashes or times out, or the reader returns 5xx
status alarm	the job’s summary line says failed or partial_failure
freshness	no successful index write for 36 h
canary write	a sentinel write and read-back, as the job’s identity, fails (daily at 12:00 and after a deploy)
run ledger	a day’s window has other than one summary, any failed write, updated=0 with input, or a no-match rate over 5%
reconciliation	in any day bucket: more than 5% of expected task ids missing from the reader’s view, any extra, any content mismatch

Pick a fault and watch which lamps light. The default generalizes the staging story: after an upstream run-id format change, the job reports status=success, updated=0, no_docs_matched=452 and the reader returns 200 OK with 0 of 635 rows.

The matrix gives hours from fault start to the first alarm (·: never fired in the 20 days after onset); on the healthy baseline none of the six fired in 40 days. Most hours are cadence (a check that runs after the daily sync sees a fault at its first evaluation, 26 hours after onset), so read the blanks.

Two of the default alarms’ catches are luck. Freshness sees the key rename only because every lookup failed; rename a fraction of the keys and writes keep flowing. Error rate catches the scroll fault only because retry backoff stretched the run past its time limit and the runtime killed it after writing 168 of 452 groups; a time-budget guard that exits cleanly with partial_failure would silence it. The status alarm misses that fault because a killed job prints no status, and AWS CloudWatch, for one, defaults to treating missing data as missing, which moves the alarm to INSUFFICIENT_DATA, not ALARM.

My hypothesis going in was that reconciliation alone catches every class, and it does not. Merging by task id makes the write idempotent (three runs leave the same result as one), so the index ends exactly right and a data check has nothing to find; the damage is triple the load. The ledger catches it because one day’s window produced three summary lines. Idempotency turned a corruption into a cost problem, and cost problems need a run-level check.

The math

What each part of the check buys

The bench also runs six variants of reconciliation, each changing one design choice from the full check:

Every variant, like the full check, also misses the idempotent retry. The last row is the one you are most likely to ship first: with unjoinable rows counted as missing it fires every day, someone mutes it, and you are back to no reconciliation. The fifth goes blind because a key rename shrinks expected along with visible, so you are reconciling the sink with itself. The second matters because extra rows mean the source lost data (a TTL, a botched migration) and the sink has become the system of record without anyone deciding so.

For content, hash each (task_id, label) pair into a 64-bit number and add the bucket’s hashes, wrapping around at 2⁶⁴, to get the bucket’s digest. Use a sum, not XOR: XOR cancels duplicate pairs, so three copies look like one (recon_sql.py asserts this).

Even a healthy pipeline misses rows. Keys without a document leave up to 3.07% of a day’s tasks missing (17 of 553 on the worst day), so the bench’s tolerance is 5%, and that is a blind spot: 1% vocabulary drift is never detected, 2% and 3% are detected after 74 hours only because noise pushes one day over the line, 4% and 5% after 50 hours, and 10% at the first run. To shrink the tolerance, count known exclusions with a reason (the job already knows which keys had no document), check visible + excluded_with_reason = expected at a tolerance near zero, and alarm on the exclusion rate separately so that a key rename, which turns everything into “no document”, still pages.

Per bucket, the full check is a full outer join on task id between a dump of the source and an export of what the reader’s query returns:

WITH expected AS (          -- from the source alone
  SELECT task_id, label, day
  FROM source_tasks
  WHERE conversation_key IS NOT NULL  -- joinable rows only
),
visible AS (                -- exported through the reader's filter
  SELECT task_id, min(label) AS label, max(label) AS label_max,
         min(day) AS day, count(*) AS copies
  FROM sink_export
  GROUP BY task_id
)
SELECT coalesce(e.day, v.day)                    AS day,
       count(e.task_id)                          AS expected,
       count(v.task_id)                          AS visible,
       count(*) FILTER (WHERE v.task_id IS NULL) AS missing,
       count(*) FILTER (WHERE e.task_id IS NULL) AS extra,
       count(*) FILTER (WHERE e.task_id IS NOT NULL
                          AND v.task_id IS NOT NULL
                          AND (v.copies > 1
                               OR v.label <> v.label_max
                               OR v.label <> e.label)) AS mismatched
FROM expected e
FULL OUTER JOIN visible v ON v.task_id = e.task_id
GROUP BY 1
ORDER BY 1;

recon_sql.py runs this exact query in SQLite and asserts that it matches the model on every bucket for six scenarios. The widget shows the result after day 23’s sync, one square per day. Untick the reader-filter box under vocabulary drift, then try lost update, which damages every day at once, a few percent at a time.

reconciliation · source task ids vs index, by completion dayafter day 23's run

no faultpermission lossvocabulary driftsource TTLretry ×3, appendlost update count what the reader’s filter returns

within tolerancemissing > 5% extra in indexcontent mismatch

day	expectedexp	visiblevis	missingmiss	extra	mismatchmism

–

Repair is a second incident

The backfill

Putting 33 days of labels back, once the job could write again, was a load test nobody planned.

First, retries stack. retry_stack.py counts 3 concurrent copies of a run invoked synchronously from a CLI (the job outlives the CLI’s read timeout while the SDK re-sends) and 1 for async invocation, where the platform retries a failed async run twice by default but sequentially. If a single write outlives its 30-second client timeout with 4 attempts per write, that reaches up to 12 concurrent operations on one document (Nobody pressed cancel has the general case).

Second, _update_by_query is a scrolled search plus bulk updates, not a single write, and not atomic: if it fails partway, the updates that completed “still stick, they are not rolled back”, so a retry must be safe to apply twice. Its scroll holds a search context open on every shard (an index is split into shards across the cluster’s nodes), for 5 minutes by default. OpenSearch caps open scroll contexts at 500 per node, counted per shard context, not per request (an OpenSearch issue walks through 36 shards × 10 slices = 360 contexts, where slices are parallel parts of one scroll). So a job that issues one update-by-query per item can exhaust a node-wide limit while the cluster’s CPU idles.

Health metrics measure the cluster, not your request pattern, so an idle cluster is consistent with every one of your calls failing. And a 429 from these engines can mean a full thread-pool queue, a tripped circuit breaker or, as here, the scroll-context limit, so log the error body.

Second-order damage

One label, whole document

A review label is about 219 bytes of JSON, but here it lived inside a document that also held the conversation text, a 1024-dimension embedding and nested evaluation children. Lucene, the library underneath Elasticsearch and OpenSearch, implements updateDocument “by first deleting the document(s) containing term and then adding the new document”, and nested objects are separate hidden Lucene documents stored with their parent, so the whole bundle is rewritten. The old copy stays on disk until Lucene merges the immutable segment files that hold it (McCandless).

Before Elasticsearch 9.2 and OpenSearch 3.0, a vector field was also kept in _source, the original JSON stored beside the indexes; newer versions leave it out by default for new indexes (Elasticsearch, OpenSearch). write_amp.py measures a 1024-dimension float32 vector serialized the way a Python client does it: 22,737 bytes of JSON against 4,096 bytes of binary, 5.55 times larger.

The ratio of bytes written to bytes you meant to change is write amplification. The widget splits one document by part and compares a label update embedded in it with one in a small sidecar index.

one label, whole document · bytes re-indexed per label updatemodel

label updates

–

backfill: 30,743 text

8 KB

vector also in _source

label inside the document–

label in a sidecar index–

–

With 8 KB of text, the vector in _source and two shard copies, each label update re-indexes 76.3 KB. The 30,743 label documents the backfill added re-index 2.40 GB, against 13.5 MB if the labels had lived in their own small index, 178 times less. Without the vector in _source the figures are 1.00 GB and 75 times. Those are lower bounds: postings, doc values, vector graph links and merge rewrites all add to the embedded side.

My hypothesis for the full disk, and it is a hypothesis: about 30,000 small label updates left gigabytes of deleted document copies behind. The at least 30,743 documents touched were roughly 1.3% of the index’s 2,431,826, far below the share of deletes that Lucene’s merge policy tolerates before it targets them: 20% by default since Lucene 9.5, 33% before. So nothing forced their reclaim during the run, and on a node that was already nearly full, that was enough. I did not measure segment sizes before and after, so treat this as the leading explanation, not a finding.

Prior art

Where this sits

Data-quality tools such as Deequ (Schelter et al., VLDB 2018) and dbt-utils’ equal_rowcount and equality compare row counts or full rows between two tables in one warehouse. Row counts let a missing row and an extra row cancel, and the natural warehouse export of a search index is the raw index, not the reader’s view. Anti-entropy compares replicas, which share a schema and keys; a derived store’s expected set is a subset of the source, read through its own filter. The variants table measures what each of those differences costs.

Fixes, ranked

Fixes, from removing the class to mitigating it

1. Give every derived store a reconciliation job. Count and digest per bucket against the source, through the reader’s query, in both directions, with expected computed from the source alone and exclusions counted with a reason. Run it over recent buckets daily and over all history weekly, because TTLs and lost updates damage history. At scale, compare (count, digest) per bucket first and drill down, by hour and then by key prefix, only where they differ. Cassandra’s anti-entropy repair stores that drill-down as data, in Merkle trees that replicas compare “to identify mismatched data”.

2. Make the run ledger a real output of the job. Every input row ends in exactly one outcome, and the status is computed from those counts (the run-ledger rule in the detector table), not chosen by whichever code path happened to run. Treat a missing summary as a breach (in CloudWatch, TreatMissingData: breaching) and alarm when a window has more than one run.

3. Treat identity as configuration with a test. After every deploy, write a sentinel document as the job’s role and read it back through the reader, using a real key from yesterday’s source so the canary also fails on a key rename. Know how your access layer matches roles: AWS IAM converts a role ARN in a trust policy to the role’s unique principal ID, so a recreated role does not inherit it, but a search cluster’s backend-role mapping matches the ARN as written, so a replaced role loses access and, by the same logic (untested), a recreated name would regain it. In CloudFormation, changing an IAM role’s RoleName requires replacement.

4. Keep mutable labels out of big AI documents. A sidecar index keyed by (conversation_id, run_id, task_id) makes a label update cost the label, and because no two writers share a document, a full replace by the conversation indexer can no longer erase labels. Reads join at query time, in two steps or with a terms lookup, which fetches its list of ids from another document (capped at 65,536 terms by default). If labels must stay embedded, batch the updates and budget disk for a deleted copy of every document you touch, times the number of shard copies, until merges catch up.

5. Use one retry layer, idempotent writes and bulk APIs. Merge by task id so replays are harmless, retry at exactly one layer, and replace per-item update-by-query with a plain search to resolve document ids (no scroll context) plus one _bulk request of scripted updates, each addressed by _id. That version has traps of its own. A size cap on the id search silently drops documents beyond it, so compare total hits with returned hits. Concurrent partial updates to one document conflict on its version number, so set retry_on_conflict. And with the bulk helper’s raise_on_error=False, a flood-stage block comes back as per-item failures inside a successful response, so count them, per group rather than per document, or the ledger stops adding up.

6. Match retention, or decide who the system of record is. If the source has a TTL and the sink does not, the sink is your only copy of old data, and a rebuild from the source loses it. Give the sink the same retention, or declare it the system of record and back it up like one.

Detect it

The check that catches it on day one

Write down each derived store’s invariant: which rows of which source should be visible, through which query, keyed by what. Then add a vocabulary contract test in CI: generate the writer’s allowed values (tier labels, visibility values) and the reader’s filter values from one shared definition, and fail the build if the writer can emit something the reader does not query; it would have caught the unknown tier values and the unqueried visibility value in the field notes. Fixes 1 and 2 supply the other alarms. Finally, inject a fault into staging the way the bench does and confirm that the alarm you expect pages. A detector that has never fired is another well-formed empty result.

Try it

Run the bench

Everything is stdlib Python 3.9 plus Node for the parity check, seeded and deterministic. The whole set took 84 seconds on my laptop:

cd experiments/empty-is-not-an-error
python3 -I fault_matrix.py  # matrix, ablations, drift sweep
python3 -I recon_sql.py     # the post's SQL in SQLite vs the model
python3 -I write_amp.py     # bytes per label update, embedded vs sidecar
python3 -I retry_stack.py   # concurrent copies per retry layer
node parity.mjs             # widget JS model == results.json

Change P in fault_matrix.py to match your own pipeline (the unjoinable share, the no-document share, your schedule and tolerances), then add the fault you are most worried about and see which of your detectors would notice.