Sources/orchestration: validate ordered delta-chain continuity before advancing state

[LouisDeconinck]

Problem

Some bulk-data feeds consist of independent objects:

filing-123.pdf
filing-456.pdf
filing-789.pdf

Those can be processed independently.

Other feeds form an ordered delta chain where every predecessor is required.

For example:

full snapshot
    ↓
D20260901.zip
    ↓
D20260902.zip
    ↓
D20260903.zip

A national registry may define the current state as:

latest full snapshot
+ every subsequent delta in publication order

For such a source, discovering and successfully parsing D20260903.zip is not sufficient if D20260902.zip is missing or failed.

Applying later deltas across a gap can produce a dataset that is structurally valid and whose pipeline reports success, but whose state is incorrect.

Relationship to #324

#324 correctly handles:

Which new or changed remote objects exist and have not been successfully processed?

This issue asks a different question:

Is this sequence complete enough that the next object may safely be applied?

I would keep these separate.

Remote collection discovery should not need to assume that every collection has sequence semantics.

Proposed solution

Add an optional ordered-sequence / continuity contract that can operate on discovered objects or partitions.

Conceptually:

sequence:
  keyFrom: filename
  pattern: "D{date:YYYYMMDD}.KBO.zip"
  orderBy: date
  requireContinuity: true

or more generally:

sequence:
  valueColumn: publication_date
  cadence: day
  requireContinuity: true

The exact configuration shape is less important than the semantics.

Expected behavior

Given:

2026-09-01 succeeded
2026-09-02 missing
2026-09-03 discovered

Duckle should report something like:

sequence status: blocked
expected: 2026-09-02
next observed: 2026-09-03
reason: sequence_gap

and must not advance the authoritative sequence state past the missing item.

Failure semantics

A discovered delta should only advance the sequence when its associated processing succeeds.

For example:

D01 succeeds
    -> sequence position = D01

D02 parse/publish fails
    -> sequence remains D01

D03 exists
    -> visible as discovered
    -> blocked by D02

Retrying D02 successfully should then make D03 eligible.

This should compose with Duckle's durable item/slice state rather than create another job system.

Missing versus late

A gap may simply mean the publisher has not released the file yet.

Useful states could therefore include:

waiting
blocked
complete
gap

and an optional grace period could distinguish:

expected item not published yet

from:

sequence gap persisted beyond expected publication window

Freshness alerting from #304 can remain responsible for notifying operators.

Sequence identity

The sequence position should be based on an explicit deterministic key such as:

publication date
monotonic sequence number
version number

Do not infer order solely from mtime unless the user explicitly chooses that semantics.

Initial baseline

The initial state should be explicit.

Examples:

start from latest successful full snapshot
start from sequence 123456
start from date 2026-01-01

This lets a monthly/full snapshot reset the required delta chain.

Full snapshot reset

Some feeds periodically publish a new full snapshot.

It should be possible to model:

full snapshot F2 succeeds
    ↓
sequence baseline becomes F2
    ↓
only deltas after F2 are required

The older chain can remain in provenance without blocking the new generation.

Out-of-order arrival

Discovery and application order should be distinct.

For example:

D03 arrives
D02 arrives ten minutes later

Duckle may discover both immediately, but should process/apply them according to the configured sequence where continuity is required.

Provenance

A run applying a delta should record enough information to answer:

sequence key
previous accepted sequence key
source object/version
producer run
resulting sequence position

so the reconstructed state can be audited.

Scope

This is not a general stream-processing framework.

The first implementation can support deterministic ordered feeds with:

date cadence
integer sequence

and block on gaps.

That already covers many registry and incremental bulk-data feeds.

Acceptance criteria

1. an ordered daily/integer delta feed can declare a continuity requirement;
2. a later item cannot advance state while an earlier required item is missing or failed;
3. out-of-order discovery does not cause out-of-order application;
4. successfully retrying the missing item unblocks subsequent items;
5. a new accepted full snapshot can reset the sequence baseline;
6. sequence position and predecessor identity are recorded in provenance;
7. continuity uses existing durable run/item/slice primitives rather than a separate executor.

[SouravRoy-ETL]

This one is the cleanest of the three to state and the easiest to get subtly wrong, so let me be precise about where we are.

None of it exists. Every ingredient does. I grepped for continuity|contiguous|sequence_gap|monotonic across all crates: five hits, and not one is a check.

What does exist is every position-advancing mechanism you would need to gate:

• persist-on-success-only for every resumable position, with an optimistic-concurrency check against a mid-run edit (lib.rs:2660, gated at :2677)
• DuckLake CDC advancing last -> current snapshot (connectors.rs:16302, :16311-16316)
• xf.incremental with col > saved and a new mark of MAX(col) (connectors.rs:16191-16211, :16227)
• Kafka resume offsets, topic and partition scoped (connectors.rs:13925, :18388)
• src.changed listing mode with its per-URI ledger (connectors.rs:5394)
• and remote listings already returned oldest-first, so a capped run works a backlog in order

The specific hole your issue names, in the code: xf.incremental steps over holes silently. The new mark is MAX over the rows that arrived (connectors.rs:16227), so a day the publisher never released is skipped permanently and nothing reports it. That is exactly your "advance past a missing predecessor", and it is one line.

The shape I would build - a sixth state kind plus a gate node, and no scheduler:

1. A sequence key parser mirroring partition::key_for, with Key as a date or an integer, parsed from a filename pattern or a column. Reuse partition::Cadence rather than inventing a second cadence enum - partition::generate already proves the calendar arithmetic and the DST-safe advance. expected_after(key, cadence) is the whole continuity primitive. Explicit key only, never mtime: today's listings sort by leaf name (connectors.rs:5680, :7218), which is close to a key and is not one.

2. A sixth state shape registered in watermark.rs, carrying accepted / cadence / baseline / previous. It has to be added to kind_of before the value branch, per that function's own ordering note, and it should be editable - "start from sequence 123456" and "reset the baseline" are exactly the operator edits that module exists for, and they are your acceptance criterion 5.

3. A gate node rather than a change to src.changed, taking the rows that source already emits and passing through only the contiguous prefix starting at expected_after(accepted), adding sequence_key and previous_key. Keeping it separate honours your own "I would keep these separate", and leaves the discovery source doing one job.

On your AC 7 - existing durable primitives, not a separate executor - that is already satisfiable rather than aspirational. backfill.rs is a generic slice ledger by design; #306 just landed on it, where a chunk is a slice with a different generator. A sequence is a third generator, not a third ledger.

Two honest caveats. Advance granularity today is the whole run, not the item: every queued position lands or none does (lib.rs:2677). And the slice executor has no dependency edges - is_claimable is per-slice and workers claim independently - so your AC 2 and AC 3 both need ordering that the executor cannot currently express. That is the part I would design in this issue before writing any of it, because "block until the predecessor is accepted" is a scheduling constraint and I would rather add it as a claim predicate than as a second executor.

[LouisDeconinck]

I agree with the key architectural direction:

sequence = another slice generator
not another executor

and I think predecessor ordering belongs in the normal claimability rule rather than in a separate scheduler.

I would go one step further and try to make the durable slice ledger the authoritative progression state instead of maintaining two independent truths:

sequence watermark / accepted pointer
+
slice ledger

Conceptually, each delta becomes its own slice:

epoch / baseline
sequence_key
predecessor_key
status
run_id

and:

slice D03 is claimable
iff
D02 succeeded

or, for the first item in an epoch:

predecessor == accepted baseline

That gives the core behavior directly:

D01 succeeded
D02 missing
D03 discovered

D03 exists in the plan
but is not claimable
reason = waiting_for_predecessor(D02)

When D02 later appears:

D02 runs
-> succeeds
-> D03 becomes claimable

This also makes retry and provenance ordinary slice behavior rather than special continuity behavior.

I think we still need a small explicit baseline record for:

start from 2026-01-01
start from sequence 123456
accepted full snapshot F2

but I would treat that as the sequence epoch/anchor rather than continuously duplicating the highest successful position into a second mutable watermark.

A full snapshot reset could create a new epoch:

epoch F1
  D01
  D02 failed
  D03 blocked

new full snapshot F2 accepted

epoch F2
  baseline = F2
  old D02 no longer blocks new work

That keeps old provenance without making obsolete gaps block the new generation.

I also agree that continuity should remain explicitly opt-in. A normal monotonically increasing watermark is allowed to have gaps; an ordered registry delta chain is not. So I would not globally change xf.incremental semantics merely because its current MAX behavior demonstrates the failure mode.

The useful common primitive seems to be:

is_claimable(slice)
    -> normal eligibility
    -> resource/concurrency rules
    -> optional predecessor-success requirement

That can support ordered deltas without creating either another executor or another scheduler.

[SouravRoy-ETL]

Your version is better than mine and it removes a piece rather than adding one.

I proposed a sixth watermark state kind plus a gate node, which means two records of how far the chain has got: an accepted pointer in state/<pipeline>/<node>.json and the slices themselves. Two truths that must agree, and the interesting failures are exactly the ones where they do not - a retry that moves the ledger and not the pointer, or an operator edit that moves the pointer past a slice that never ran. Making the ledger authoritative deletes that whole class.

slice D03 is claimable
iff D02 succeeded, or predecessor == accepted baseline

That is one rule in one place, and it composes with what is already there: is_claimable today is "requested only", and this makes it "requested, and its predecessor is done". The gate node then does not need to exist, because the ledger will not hand out a slice whose predecessor has not landed - which is a stronger guarantee than filtering rows, since filtering happens after something already decided to run.

Epochs. F2 creating a new generation, with old D02 no longer blocking, is the part I had not thought about and it is the difference between an operator-recoverable system and one that needs a state file edited by hand after every missed delta. Keeping the old epoch's slices for provenance rather than deleting them is right for the same reason a backfill keeps its failures.

Not touching xf.incremental. Agreed, and I want to be careful about how I described it. Its MAX-over-arrived-rows behaviour demonstrates the failure mode, but a plain watermark over an append-only stream is allowed to have gaps - that is what a watermark means. The bug is only a bug when the thing being consumed is an ordered chain that promises no gaps, and the source cannot know which it is. So continuity stays opt-in and declared, and xf.incremental keeps its semantics.

So the primitive is the one you wrote:

is_claimable(slice)
    -> normal eligibility
    -> resource / concurrency rules
    -> optional predecessor-success requirement

Two things I would still need to settle before building, both about the generator rather than the rule:

• where the sequence key comes from. Parsed from the object name is the common case, but the parse has to be declared and total - a filename that does not match the pattern must be a refusal, not a skip, or a mis-named delta silently drops out of the chain and the chain still looks contiguous.
• what happens at the head. A chain is always missing its next element; "D03 is blocked because D02 has not arrived yet" and "D02 will never arrive" are the same state to the ledger and very different to an operator. That probably wants a declared expectation - a cadence, or a maximum wait - so the gap can be reported rather than waited on forever.

Neither changes the shape you described; they are the two places where an ordered chain differs from a set of independent slices, and I would rather name them now than discover them in the executor.

[LouisDeconinck]

I agree these are the two remaining generator-level questions.

1. Sequence parsing should be total over the selected input set

I would make this invariant explicit:

Every object that reaches the sequence generator must either produce exactly one valid sequence key or cause a structured refusal.

So:

collection listing
    ↓
include/exclude filtering
    ↓
sequence-key parser

If a directory legitimately contains unrelated files, they should be removed by the normal discovery filter before they reach the sequence contract.

Once an object is in the selected sequence set:

filename does not match pattern
invalid date
invalid integer
duplicate sequence key with conflicting object

should be visible errors, not skips.

Otherwise:

D20260901.zip
D202609O2.zip    # typo: O instead of 0
D20260903.zip

could silently become:

D01
D03

and the system would report a gap without revealing that the supposedly missing publication is actually present under a malformed name.

A machine-readable result such as:

code = invalid_sequence_key
uri = ...
expected_pattern = ...

would also make this straightforward for an agent to diagnose.

2. Distinguish an open head from a proven internal gap

I would not call the next expected value a gap merely because it has not arrived yet.

There are three useful cases.

Normal head

last succeeded = D01
D02 not observed
no later item observed

This is simply:

waiting_for_next

The sequence is healthy as far as continuity can determine.

If the sequence declares an expected cadence, #304/freshness-style logic can eventually say it is late:

expected D02 by 04:00
grace until 08:00
now 12:00
-> overdue / stale

but that is an availability/freshness concern, not evidence of an internal sequence gap.

Proven gap

last succeeded = D01
D03 observed
D02 not observed

Now the gap is known:

expected predecessor = D02
observed later item  = D03

D03 status = blocked
reason = missing_predecessor

This should happen immediately; no grace period is needed to prove the chain is incomplete because a later item already exists.

Integer/irregular sequence without cadence

For something like:

101
102
103

with no publication schedule, Duckle can still prove:

101 succeeded
103 observed
102 absent

is a gap.

But if nothing after 101 has appeared, there is no basis for claiming that 102 is late unless the user separately declares an expectation.

That keeps the two concepts clean:

#326 continuity
    -> can later observed work safely proceed?

#304 freshness / expectation
    -> should new work have appeared by now?

Claimability

With that distinction, the common rule can stay small:

is_claimable(slice)
    -> requested?
    -> normal concurrency/resource rules?
    -> if ordered:
         predecessor succeeded
         OR predecessor == epoch baseline

A later slice with a missing predecessor exists durably but is not claimable.

The current head does not need a synthetic missing slice unless a later observed key proves the gap or an explicit cadence/expectation mechanism decides to materialize one for observability.

That keeps the executor simple while still making both late feeds and real holes explainable.

[SouravRoy-ETL]

Built, on your shape rather than mine. ee6ef08, d2ad82f, 860d817.

The ledger is the only progression state. No accepted pointer anywhere - the position is derived by walking the slices, so the class of failure I was worried about (a retry that moves one and not the other) cannot be expressed. applied is take_while(succeeded), not a count of successes, which is the difference between contiguity and arithmetic.

The claim rule, as you wrote it:

is_claimable(slice)
    -> requested?
    -> if ordered: predecessor succeeded, or predecessor == baseline

The part worth naming is which way round the requirement is stored. A missing delta has no slice - there is nothing to run - so a ledger asked "did my predecessor succeed" about a row that does not exist has to answer no. Keeping the requirement on the successor, naming a key, makes an absent predecessor block exactly as a failed one does, while the baseline satisfies it by carrying no requirement at all. A chain is therefore serial by construction: the executor sizes its worker pool from claimable slices, so three requested links start one worker.

Totality. Four codes rather than your single invalid_sequence_key - pattern_mismatch, invalid_date, invalid_integer, duplicate_sequence_key - because the operator response differs and only the last is a correctness emergency. Say if you would rather have the one code; it is a rename. Your D202609O2.zip case is a test, and a refusal refuses the plan, not just a log line: a chain that looks contiguous because a mis-named file was skipped is precisely what the contract exists to prevent, so it cannot be a warning a scheduled run steps past.

Your open-head distinction changed the design after I ran it. I had the verdict taking a boolean "has this key succeeded", and that cannot tell a link that FAILED from one that has not been asked to run. A freshly planned chain reported blocked / predecessor_failed against its own first link, which was published and simply pending. It now reads the ledger state and the three cases separate properly:

hole, later object exists    -> blocked, sequence_gap
published, failed            -> blocked, predecessor_failed
published, not applied yet   -> waiting_for_next, observed: true

waiting_for_next carries observed, which is the thing a pointer cannot say: whether the publisher has not released it, or Duckle has not applied it. The first is #304's question and this deliberately does not answer it.

Live, not just tested - a directory of drops with D20260902 absent:

sequence status: blocked
position:        2026-09-01
expected:        2026-09-02
next observed:   2026-09-03
reason:          sequence_gap

and the ledger it plans:

2026-09-01 | requires: None        <- baseline
2026-09-03 | requires: 2026-09-02  <- the hole
2026-09-04 | requires: 2026-09-03

Drop D20260902 in, and status returns to waiting_for_next with exit 0.

duckle-runner sequence status|plan|apply, exit 1 when blocked so a scheduled check can gate on it. Everything after planning - retry, cancel, restart reconciliation - is backfill, unchanged.

Two things I did not do, deliberately.

Remote listing is a directory or an s3:// prefix, reusing S3Config::list. sftp:// is not wired: that spec is built inline in plan/mod.rs rather than behind a reusable parser, and extracting it is discovery's refactor rather than continuity's. This is the seam #324 plugs into - read() takes a list of URIs and does not care where it came from.

xf.incremental is untouched, per your point. Its MAX-over-arrived-rows demonstrates the failure mode but a watermark over an append-only stream is allowed to have gaps, and the source cannot know which it is looking at. Continuity stays declared.

One thing I would still ask. A column-sourced key (valueColumn / cadence in your original sketch) is not implemented - only a key parsed from the object name. Reading a key out of the rows means opening the object to find out whether it should be applied, which inverts the point of a cheap listing. Is the filename case enough for the feeds you have in mind, or is there one that only carries its sequence inside?

[LouisDeconinck]

This implementation covers the difficult part of #326 very well.

The architecture is now exactly the shape I wanted:

sequence
    -> slice generator
    -> existing durable ledger
    -> existing executor

with:

is_claimable(slice)
    -> requested?
    -> predecessor succeeded
       OR slice is the baseline

and no second mutable "accepted sequence" pointer that can disagree with the ledger.

The other important semantics also look right:

• sequence parsing is total rather than silently skipping malformed selected objects;
• a later observed key proves a real gap immediately;
• an open head is waiting_for_next, not incorrectly labelled a gap;
• failed predecessor and missing predecessor are distinguishable;
• out-of-order discovery cannot produce out-of-order execution;
• retry/cancel/reconciliation remain normal slice/backfill operations;
• status is machine-usable and can gate a scheduled workflow.

I would keep #326 open for one remaining acceptance criterion: AC5, baseline reset from a newly accepted full snapshot.

The scenario I still want to support explicitly is:

epoch F1
    D01 succeeded
    D02 failed
    D03 blocked

new full snapshot F2 succeeds

epoch F2
    F2 becomes the baseline
    deltas after F2 form the new chain
    old D02 remains in provenance
    old D02 no longer blocks current ingestion

I would avoid mutating/reinterpreting the old sequence plan.

Conceptually, accepting F2 should create a new sequence epoch / ledger generation with F2 as its anchor.

Useful provenance would then be something like:

sequence_id
epoch_id
baseline / full_snapshot identity
sequence_key
predecessor_key
run_id

The previous epoch remains historical evidence rather than state that has to be deleted.

For Signumi this matters because registry publishers commonly issue:

periodic full snapshot
+
subsequent daily deltas

and a fresh authoritative full snapshot should provide a clean recovery point after a permanently missing/broken old delta.

Once that reset/epoch behavior is present, I think all seven acceptance criteria are satisfied and #326 can close.