Function seal_upper 

pub(super) fn seal_upper<T>(
    persist_upper: &Antichain<T>,
    input_upper: &Antichain<T>,
) -> Antichain<T>
where
    T: Timestamp + TotalOrder + Refines<Timestamp>,

The start of the mz_repr::Timestamp after persist_upper’s, capped at input_upper so no unfinalized time is extracted.

Only entries at persist_upper are eligible in a round, and the drain returns every extracted entry past persist_upper as a resident Vec, outside the buffer pool budget, to be re-pushed through the chunker. Sealing at input_upper instead would do that for the whole backlog between the two frontiers once per persist round trip, which is the entire source tail while the operator catches up from behind.