Fix concurrent bootstrap output progress - #6849

Claude claims that this re-introduces the same bug, just in enqueue_latest_snapshot, because that function does this:

        let processed_records = processed_records.or(Some(ProcessedRecords {
            total_processed_input_records: self.status.num_total_processed_records(),
            total_processed_steps: self.status.global_metrics.total_completed_steps(),
        }));

and that it should be more like this:

let processed_records = if owed_reemission {
    None // withheld — must not be backfilled; leave the counter untouched
} else {
    processed_records.or(Some(ProcessedRecords {
        total_processed_input_records: self.status.num_total_processed_records(),
        total_processed_steps: self.status.global_metrics.total_completed_steps(),
    }))
};

It says:

Two distinct meanings of None are colliding.

• Pre-existing meaning (mid-transaction): "no override supplied; the last fully-committed count is still accurate, so stamp the endpoint with the current global count." That's what .or(Some(ProcessedRecords { total_processed_input_records: self.status.num_total_processed_records(), .. })) at controller.rs:7972 correctly does — and it's a case enqueue_latest_snapshot can genuinely see safely, because it's only used mid-transaction, when the global count truly hasn't moved past reality.
• New meaning (owed re-emission): "a real batch was produced, but reporting it would claim output the endpoint hasn't received yet — leave the counter alone." For the ordinary delta path, output_batch/OutputBuffer::insert already do exactly that (skip the store on None). But enqueue_latest_snapshot doesn't know this None means "skip" — it only knows the old meaning, so it invents a fresh stamp instead of skipping.

What correct behavior looks like: when enqueue_latest_snapshot delivers the stale, pre-cutover snapshot to a freshly-pending endpoint on a relation still owed a re-emission, the progress counter should stay wherever it was (0, for a brand-new endpoint) — not jump to the pipeline's current input count. The stale snapshot is fine to send (the cutover's delta will correct it later, at Finalizing, via the normal delta path); it's only the counter that must wait, exactly as it does for every other endpoint under owed_reemission. Once the real cutover delta lands at Finalizing, that ordinary delta path already stamps the correct, by-then-accurate count — no extra logic needed there.

Frankly, I don't understand the different kinds of bootstrapping well enough to judge whether it is right.

Claude claims that this re-introduces the same bug, just in enqueue_latest_snapshot, because that function does this:

        let processed_records = processed_records.or(Some(ProcessedRecords {
            total_processed_input_records: self.status.num_total_processed_records(),
            total_processed_steps: self.status.global_metrics.total_completed_steps(),
        }));

and that it should be more like this:

let processed_records = if owed_reemission {
    None // withheld — must not be backfilled; leave the counter untouched
} else {
    processed_records.or(Some(ProcessedRecords {
        total_processed_input_records: self.status.num_total_processed_records(),
        total_processed_steps: self.status.global_metrics.total_completed_steps(),
    }))
};

It says:

Two distinct meanings of None are colliding.

• Pre-existing meaning (mid-transaction): "no override supplied; the last fully-committed count is still accurate, so stamp the endpoint with the current global count." That's what .or(Some(ProcessedRecords { total_processed_input_records: self.status.num_total_processed_records(), .. })) at controller.rs:7972 correctly does — and it's a case enqueue_latest_snapshot can genuinely see safely, because it's only used mid-transaction, when the global count truly hasn't moved past reality.
• New meaning (owed re-emission): "a real batch was produced, but reporting it would claim output the endpoint hasn't received yet — leave the counter alone." For the ordinary delta path, output_batch/OutputBuffer::insert already do exactly that (skip the store on None). But enqueue_latest_snapshot doesn't know this None means "skip" — it only knows the old meaning, so it invents a fresh stamp instead of skipping.

What correct behavior looks like: when enqueue_latest_snapshot delivers the stale, pre-cutover snapshot to a freshly-pending endpoint on a relation still owed a re-emission, the progress counter should stay wherever it was (0, for a brand-new endpoint) — not jump to the pipeline's current input count. The stale snapshot is fine to send (the cutover's delta will correct it later, at Finalizing, via the normal delta path); it's only the counter that must wait, exactly as it does for every other endpoint under owed_reemission. Once the real cutover delta lands at Finalizing, that ordinary delta path already stamps the correct, by-then-accurate count — no extra logic needed there.

Frankly, I don't understand the different kinds of bootstrapping well enough to judge whether it is right.

This was a good catch, thank you.

The CI failure is unrelated; I will work on it separately.