Module cleanup 

Scoped cleanup — delete destination rows a run did not write (#478).

An incremental sync into an upsert sink cannot remove records deleted at the source: a record that disappears simply stops appearing in the “updated since X” feed, so write_mode: upsert keeps it in the destination forever. The destination looks healthy, the run reports success, and stale rows accumulate indefinitely with nothing surfacing the divergence.

Scoped cleanup closes that hole for the case where the source can make a completeness claim: “for scope S, these are all the records”. The canonical shape is a parent/child incremental sync — a child row fetching one contact’s associations is authoritative for contact_id = <that contact>.

Why the claim comes from the source

A sink cannot make it. It observes a page of records and cannot distinguish a complete set from page 1 of 3, or from a partial page preceding a failure. Two further reasons the scope is declared upstream rather than on the sink:

1. Drift safety. The predicate already exists in the source config (/contacts/${contacts.id}/associations). Declaring it a second time on the sink means two copies of one predicate, and when they diverge the pipeline deletes the wrong rows.
2. The empty-result case, which is decisive. A contact that had five associations and now has none produces a fetch returning zero records. Any design inferring scopes from observed records never learns the scope existed, so the five stale rows survive — i.e. it fails precisely the case the feature exists to fix. A scope declared by the invocation comes from the parent record and survives an empty result set.

When it runs

Once per invocation, after a fully successful run — never per page. Two ways to get this wrong, both silent data loss:

• Per page: a scope’s records spanning two pages → page 2’s “delete what I didn’t see” wipes what page 1 just wrote.
• After a partial run: the fetch dies at 40% → the delete removes the 60% that had not yet arrived.

So run_stream only invokes it when the stream reached its natural end uncancelled, and the CLI additionally attaches the policy only for real root invocations (never --dry-run / --limit / a shard).

Key accumulation and the ceiling

Deleting “what this run did not write” requires knowing what it wrote. The pipeline accumulates the key tuples only (not records) as pages are written — see SeenKeys. That is bounded by the scope’s row count, which is small for the per-parent shape this targets, so a ceiling (CleanupPolicy::max_keys) guards the pathological case: on breach the cleanup aborts with a typed error and deletes nothing, rather than issuing a partial delete that would destroy rows it simply forgot about.

Structs

CleanupPolicy

A compiled cleanup instruction for one invocation.

CleanupTracker

A sink wrapper that records the key tuples written through it (#478).

SeenKeys

The set of key tuples an invocation wrote, accumulated across pages.

Enums

CleanupMode

What to do about destination rows inside the claimed scope that this run did not write.

Constants

DEFAULT_MAX_KEYS

Default ceiling on accumulated keys for one invocation’s cleanup.