SS-492 - storage: let PostgreSQL exports progress independently during a snapshot

[martykulma]

Motivation

Every export's frontier stays at the minimum until the last table's COPY finishes. Adding one large table to a source therefore still stalls ingestion for every table already hydrated in that source.

Description

With storage_persist_sink_description_lookahead nonzero, each export's port holds the minimum capability only while that export still has snapshot data to emit or a pending rewind.

• Exports that are not being snapshotted drop their snapshot capabilities up front, so they advance with the replication stream.
• A snapshotted export releases its snapshot capability as soon as its own copy finishes on each worker.
• The replication operator downgrades every port without a pending rewind at each yield point.

With the lookahead at zero, all ports advance together, so behavior matches #39419.

The replication operator now reads the setting as well. The two operators read it independently at startup, and parallel-workload rolls the lookahead through zero and nonzero values, so a change can land between those two reads and leave the operators in different modes within one dataflow. That is benign in both directions. With only the snapshot operator concurrent, the replication operator holds every port until the rewinds resolve and then releases them all. With only the replication operator concurrent, it still waits for the rewind input to close, which the serial snapshot operator does after the copy.

Verification

The mzcompose default sets the lookahead, so the existing mzcompose-based PostgreSQL source tests exercise per-export progress. #39420 adds targeted coverage.

🤖 Generated with Claude Code

[linear-code]

SS-492

[peterdukelarsen]

Looks good to me overall, some more nitty small requests that don't change the logic.

[peterdukelarsen]

Is there a risk of a race here where one part of the full rendered source saw a different value for this flag? Wondering if we should pull the config value up higher and wire down the flag lower.

[martykulma]

this is addressed, moved this out of the async operator

[peterdukelarsen]

nit: stylistically I'd pref just wiring down a shared boolean if it's less than lets say three layers of functions between the nearest common ancestor. I buy that this is correct, but it feels complex.

[martykulma]

Fair. I avoid adding parameters when I can derive from the existing inputs cheaply. The shared boolean pushes render to 9 parameters, and it sets a pattern for future flags that just keeps growing that input list.

I'm down with refactoring this down the road since it definitely seems like we can group some things together.

[peterdukelarsen]

Is there a way we can collapse these repetitive pre-continue data_cap_sets[output_index] = CapabilitySet::new();? One Idea I have would be moving the piece inside the for (&output_index, info) in outputs.iter() { into a function and then wrapping that in another function that looks structurally like:

fn handle_output_wrapping() {
    handle_output_existing();
    if concurrent_replication {
        data_cap_sets[output_index] = CapabilitySet::new();
    }
}

fn handle_output_existing() {
    // all existing logic (replace continues with returns).
}

[martykulma]

If I understand correctly, wouldn't handle_output_existing be different in each case? Wouldn't we have to break them all up in to function (or pass them as FnMut)?

Replacing:

    if concurrent_replication {
        data_cap_sets[output_index] = CapabilitySet::new();
    }

with this is simpler to read, but not by much:

drop_caps_if_concurrent(data_cap_sets, output_index);

[peterdukelarsen]

I'm not sure we're on the same page. I was recommending putting the entirety of the existing code in the for loop into a function. Then slotting in the new code for capability dropping in a wrapping function.

Definitely wasn't trying to recommend something so narrow as changing:

    if concurrent_replication {
        data_cap_sets[output_index] = CapabilitySet::new();
    }

to:

drop_caps_if_concurrent(data_cap_sets, output_index);

[martykulma]

ah, gotcha! hmm.. there's probably close to 10 local variables that then need to be passed into the function, and that's a mess on its own.

An alternative is to do an async block, but this is a little weird because you have to return a result from it.

let copied = async {
    if let Err(err) = verify_schema(oid, info, &upstream_info) {
        // give the error
        return Ok(());
    }
    let Some(ctid_range) = worker_ctid_range(&config, block_count, oid) else {
        return Ok(());
    };
    // COPY
    Ok::<_, TransientError>(())
}.await;
copied?;
if concurrent_replication {
    data_cap_sets[output_index] = CapabilitySet::new();
}

That's worth it if we're planning to add more exit paths, but honestly, I don't think that's the case.

[martykulma]

I tried to rewrite this morning (11 parameters as is). A few of them were derived from existing structs, so I adjusted. Should be ok now (at least no clippy exception).

[peterdukelarsen]

Looks great to me now, thank you!!

[martykulma]

thanks for the review!

[martykulma]

Fair. I avoid adding parameters when I can derive from the existing inputs cheaply. The shared boolean pushes render to 9 parameters, and it sets a pattern for future flags that just keeps growing that input list.

I'm down with refactoring this down the road since it definitely seems like we can group some things together.

[martykulma]

If I understand correctly, wouldn't handle_output_existing be different in each case? Wouldn't we have to break them all up in to function (or pass them as FnMut)?

Replacing:

    if concurrent_replication {
        data_cap_sets[output_index] = CapabilitySet::new();
    }

with this is simpler to read, but not by much:

drop_caps_if_concurrent(data_cap_sets, output_index);