[shuffle] Only report latest outbox truncation idx from the shuffle - #5091
Conversation
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2e801d9b26
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| /// | ||
| /// Since v1.7.3 | ||
| #[cfg_attr(feature = "schemars", schemars(skip))] | ||
| pub self_proposal_queue_memory_limit: NonZeroByteCount, |
There was a problem hiding this comment.
The new worker.self-proposal-queue-memory-limit field is documented in crates/types/src/config/worker.rs:158-169 as a user-visible option with a 64 MiB default, and this change also alters self-proposal back-pressure/batching behavior, but the change set does not include a release-notes/unreleased entry. release-notes/README.md:113-117 says behavioral changes and new user-facing capabilities need release notes, so users upgrading would miss the new per-partition memory limit and its batching/latency tradeoff unless a note is added.
AGENTS.md reference: AGENTS.md:L51-L51
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7a628fd99d
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| #[cfg_attr(feature = "schemars", schemars(skip))] | ||
| pub self_proposal_queue_memory_limit: NonZeroByteCount, |
There was a problem hiding this comment.
Marking this runtime-parsed worker option with schemars(skip) means cargo xtask generate-config-schema (which builds the schema from Configuration in tools/xtask/src/main.rs:53-57) will omit worker.self-proposal-queue-memory-limit. For users and tooling that rely on the generated config schema for documentation, validation, or completion, the new self-proposal back-pressure limit is invisible even though serde accepts it; remove the skip like the other NonZeroByteCount worker options so the option is discoverable/configurable through the schema.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7cb07082e4
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| self.refresh_status(&mut durable_lsn_watch)?; | ||
| } | ||
| Some(msg) = self.network_leader_svc_rx.next(), if self.leadership_state.should_process_rpc() => { | ||
| Some(msg) = self.network_leader_svc_rx.next(), if network_processing_permit.is_some() => { |
There was a problem hiding this comment.
When the single leader NetworkServiceEvent slot is occupied, network_processing_permit is None, so this guard stops reading any network_leader_svc_rx message. That also blocks message types that do not enqueue a leader event, such as the DedupSequenceNrQueryRequest branch in on_rpc that only calls wait_for_tail_then; Kafka's legacy dedup query path uses that RPC with a 5s timeout, so self-proposal backpressure can make unrelated dedup queries stall or time out. Consider reading the message first and requiring the permit only for proposal/ingest requests that actually buffer into network_events_tx.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Thanks a lot for improving the outbox truncation logic @MohamedBassem. The changes look good to me. One slightly unrelated note, I was wondering whether we should pull the semantics of outbox_head_seq in the Outbox straight. I think it's current behavior is not entirely consistent.
I've only reviewed this commit and not the preceding one. So if you merge it, then please merge it w/o the scheduler changes if it wasn't reviewed until then.
Before this PR, the shuffle had a 1000 sized channel between it and the PP. This meant it can accumulate up to 1k outbox truncation messages in that channel. However, outbox truncations are prefix based, so you only care about the latest one. To make things worse, the current state was dropping the "latest" truncation messages when this channel is full (using `try_send`), although it's actually the latest ones that matter. This PR changes that channel to be a watch channel instead. The PP will just use the latest messages whenever that stream gets polled. To make this work, it required another fix. On startup, the shuffle scans the outbox for the head of the queue. If the queue is empty, it stores the head as None. The head is then later used for truncation from head until N where N is the idx of the command. If at the moment of truncation, the head was still None, it uses the first command it sees as the head. So when it was writing all messages, the head correctly matched the first record written. Now that we're overriding the commands, there's no guarantee that the shuffle will get the very first truncation command it proposed. So instead, we're initializing the head during enqueue if it's not set.
|
@codex review |
|
Codex Review: Didn't find any major issues. Hooray! Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
tillrohrmann
left a comment
tillrohrmann
left a comment
There was a problem hiding this comment.
Great work @MohamedBassem. LGTM. +1 for merging :-)
Before this PR, the shuffle had a 1000 sized channel between it and the PP. This meant it can accumulate up to 1k outbox truncation messages in that channel. However, outbox truncations are prefix based, so you only care about the latest one. To make things worse, the current state was dropping the "latest" truncation messages when this channel is full (using
try_send), although it's actually the latest ones that matter.This PR changes that channel to be a watch channel instead. The PP will just use the latest messages whenever that stream gets polled.
To make this work, it required another fix. On startup, the shuffle scans the outbox for the head of the queue. If the queue is empty, it stores the head as None. The head is then later used for truncation from head until N where N is the idx of the command. If at the moment of truncation, the head was still None, it uses the first command it sees as the head. So when it was writing all messages, the head correctly matched the first record written. Now that we're overriding the commands, there's no guarantee that the shuffle will get the very first truncation command it proposed. So instead, we're initializing the head during enqueue if it's not set.