[worker] breakdown the action effect enum handling into a trait#5082
Closed
MohamedBassem wants to merge 5 commits into
Closed
[worker] breakdown the action effect enum handling into a trait#5082MohamedBassem wants to merge 5 commits into
MohamedBassem wants to merge 5 commits into
Conversation
…tead of taking actions inline As part of the work on centeralizes writes to the self proposer inside one scheduler, this PR refactors the RPC handlers to return their decisions to the main PP loop where they get executed there. This is in prep for shipping the proposals to a channel in the next PR. This PR is mostly mechanical and does the following: 1. Changes the handlers to no longer take the "Actuator" trait and instead return their `Decision` from the function. 2. The decision can be either: Propose a command (with its different modes), reply now, or the pre-vqueues invoker poking. Tests are much simpler as they no longer need to mock the leadership state. Only meaningful change was a bug in the restart_as_new test `copy_prefix_zero_with_no_version_uses_service_invocation_command` where it was using the default `patch_deployment_id` and it didn't actually hit the assertions inside the mock handlers as it was failing with `CannotPatchDeploymentId`. The tests where not setting expectations on num calls to the mock, so it didn't fail when the call to `append_and_respond_asynchronously` didn't happen. So this was fixed along the way by using `PatchDeploymentId::PinToLatest` to hit the correct assertions. This PR anyways removes `mockall` altogether from that worker crate.
This PR changes the handling of the PP RPCs and ingestion requests from happening inside the main PP loop and into the leader state run loop. This is done by introducing a new channel that gets consumed in the large stream_select of `LeaderState` run. With that all main callsites to `SelfProposer` are encapsulated inside `LeaderState` and in a lter PR will be dispatched by a new Self proposer scheduler (*). A couple of caveats to be aware of: 1. The new channel is of size one to maintain the semantics of keeping only one inflight RPC not enqueued in self proposer. I don't forsee any problems in increasing it though if needed. 2. The network service arm is now guarded by having a permit into this new channel. This means that although we could service stuff like `DedupQquery` (which doesn't propose commands), it'll currenly be blocked until `SelfProposer` is unblocked. This is not very different from now though. 3. `ActionEffects` is no longer effects, and I renamed it to `LeaderEvents` now that it carries also network service actions. Now the new `NetworkServiceEvent` is huge (~0.5KB) and it made the entire `LeaderEvent` wider. Luckily, we only keep 10 of those in memory at a time. In a later PR, I'm getting rid of this enum altogether. (*) Two other callsites to self proposer remain: the one callsite when candidate to send the `AnnounceLeader` command, and after becoming the leader to do any `VersionBarrier` command. Those two are rare and will be excluded from the SelfProposer backpressure.
This PR adds support for overdrafts in `MemoryPool` via the `force_reserve` API. This allows the memory pool to issue leases for more capacity than it holds. It also introduces a new `overdraft` API to query how much in the negative the memory pool is. Also, it introduces a new `wait_until_available` which waits until the pool is out of the overdraft mode (just as a notification without reserving anything).
This was referenced Jul 21, 2026
MohamedBassem
marked this pull request as draft
July 21, 2026 12:04
…d of record count bound
Context:
Before this PR, the background appender's channel was a bounded channel of size 50. This has two caveats:
1. Large commands (e.g. large schema upserts) can inflate the size of this channel in the worst case to 1.5GB (per partition). A coordinated event (e.g. schema change) can end accumulating 10s of GBs of unaccounted memory across all partitions causing the node to OOM.
2. On the other hands, if the append latency is high, and the records are tiny (as shown later in the benchmarks), this 50 records arbitrary limits hinders the batching efficiency quite a bit.
Solution:
The main problem is that the number of records is just hard to configure correctly across different workloads. We want reasonable batching efficiency with bounded memory growth. As such, this PR switches the background appender channel to be memory-bound instead of record-count bound. The memory bound is configurable and will default to 64MB (i.e. allowing the background appender to do up to two full batches per run, though we might want to tune how large we want a single batch to be).
Implementation wise:
1. We're switching the underlying channel to be unbounded channel with a `MemoryPool` on top.
2. Because we don't know how much to reserve from the memory pool beforehand, we're going to allow overcomitting the memory pool by at most once `enqueueXX` call.
a. To simplify reasoning about the overcomitting, I'm dropping the `Clone` support for the `LogSender`. Otherwise, we might overcomitt by more than one enqueue calls across concurrent senders. We don't currently need it to be `Clone`, so this was a no-op.
3. Once the memory pool is fully exhusted, the background appender is going to start rejecting future appends. As such, the `LeaderState` stops polling the effects stream once the pool is exhusted, and waits for it to become available again. Because leader state has an exclusive reference over the self proposer (and its underlying sender), it can be sure that if it sees that the pool has capacity, it'll be able to send without getting any reservation.
4. As a nice side effect for all the self proposer APIs becoming `sync`, we no longer need to worry about cancellation safety, so the events are handled inline inside `LeaderState::run`.
Benchmarks (Steady State):
Started a local one node cluster comparing base against this commit. The workfload is a 30s of Counter::get on a virtual object coming from 1000 concurrent connections.
```
# Base
❯❯❯ wrk -t1 -c1000 --latency -d30s -s ./scripts/wrk/counter.lua http://localhost:8080 ✘ 130 main
thread 1784573163 started
Running 30s test @ http://localhost:8080
1 threads and 1000 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 56.97ms 89.58ms 1.36s 98.26%
Req/Sec 21.23k 2.93k 29.85k 76.59%
Latency Distribution
50% 45.84ms
75% 53.83ms
90% 62.58ms
99% 581.58ms
631995 requests in 30.04s, 168.16MB read
Requests/sec: 21036.54
Transfer/sec: 5.60MB
thread 1784573163 made 632995 requests and got 631995 responses
```
```
# This commit
❯❯❯ wrk -t1 -c1000 --latency -d30s -s ./scripts/wrk/counter.lua http://localhost:8080 main
thread 1784573695 started
Running 30s test @ http://localhost:8080
1 threads and 1000 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 55.14ms 88.14ms 1.17s 98.28%
Req/Sec 22.02k 2.79k 31.03k 76.92%
Latency Distribution
50% 44.09ms
75% 51.81ms
90% 60.31ms
99% 574.92ms
655652 requests in 30.07s, 174.45MB read
Requests/sec: 21805.57
Transfer/sec: 5.80MB
thread 1784573695 made 656652 requests and got 655652 responses
```
So there's no degradation perf wise for this whole branch.
Benchmarks (higher append latency):
Where this becomes interesting is when append latency is higher because the 50recs bounded channel gets filled faster, and starts pushing back specially with tiny records, and due to the lack of pipelining, the throughput of the system collapses. So doing the same benchmarks but with a 200ms sleep in the background appender:
```
# Base
❯❯❯ wrk -t1 -c1000 --latency -d30s -s ./scripts/wrk/counter.lua http://localhost:8080 main
thread 1784574220 started
Running 30s test @ http://localhost:8080
1 threads and 1000 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 1.04s 308.07ms 2.00s 68.70%
Req/Sec 0.87k 592.52 2.87k 65.53%
Latency Distribution
50% 916.07ms
75% 1.14s
90% 1.52s
99% 1.97s
25328 requests in 30.08s, 6.74MB read
Socket errors: connect 0, read 0, write 0, timeout 1437
Requests/sec: 842.15
Transfer/sec: 229.45KB
thread 1784574220 made 26329 requests and got 25328 responses
```
vs:
```
# This commit
❯❯❯ wrk -t1 -c1000 --latency -d30s -s ./scripts/wrk/counter.lua http://localhost:8080 main
thread 1784573956 started
Running 30s test @ http://localhost:8080
1 threads and 1000 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 590.43ms 136.85ms 1.85s 81.02%
Req/Sec 1.72k 679.51 3.67k 71.75%
Latency Distribution
50% 605.69ms
75% 611.69ms
90% 617.10ms
99% 1.44s
50304 requests in 30.00s, 13.38MB read
Requests/sec: 1676.56
Transfer/sec: 456.80KB
thread 1784573956 made 51305 requests and got 50304 responses
```
This commit was able to have twice the throughput of the base branch under high latency.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to subscribe to this conversation on GitHub.
Already have an account?
Sign in.
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stack created with Sapling. Best reviewed with ReviewStack.