Skip to content

Commit c256406

Browse files
committed
persist: simplify fetch-result capability handling
Drop the completed-fetches capability directly in the match pattern (`[cap, _]`) and let the retained data capability fall out of scope, removing the explicit `drop` and the redundant comment. The error path still holds the data capability across the never-returning `report_and_stop`, so the freeze is unchanged. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01MR6SwYEgWVmMJUGn4ZmLMX
1 parent 516e306 commit c256406

1 file changed

Lines changed: 6 additions & 16 deletions

File tree

src/persist-client/src/operators/shard_source.rs

Lines changed: 6 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -717,24 +717,14 @@ where
717717
// Emit completed fetches first, so `in_flight` drains and we do
718718
// not hold more than `max_concurrency` parts in memory.
719719
biased;
720-
Some((caps, fetched)) = in_flight.next(), if !in_flight.is_empty() => {
720+
Some(([cap, _], fetched)) = in_flight.next(), if !in_flight.is_empty() => {
721721
match fetched {
722-
Ok(fetched) => {
723-
// Emit at the data capability, then drop both caps.
724-
// Dropping them advances the data and completed-fetches
725-
// frontiers once this time's last outstanding fetch is
726-
// done.
727-
fetched_output.give(&caps[0], fetched);
728-
drop(caps);
729-
}
722+
Ok(fetched) => fetched_output.give(&cap, fetched),
730723
Err(e) => {
731-
// Report the missing blob and freeze. `report_and_stop`
732-
// never returns, so we stop draining results and retain
733-
// every in-flight and pending capability, including this
734-
// failed part's `caps`. Crucially, a later successfully
735-
// fetched part must NOT be allowed to drop its capability
736-
// and let the frontier advance past the part we never
737-
// emitted.
724+
// `report_and_stop` never returns, freezing the
725+
// operator: `cap` (and every other in-flight and
726+
// pending capability) stays held, so the data frontier
727+
// never advances past the part we failed to emit.
738728
error_handler.report_and_stop(e).await;
739729
}
740730
}

0 commit comments

Comments
 (0)