[worker] Change background appender channel to be memory-bound instead of record count bound (#5076)

* [MemoryPool] Add support for overdrafts in MemoryPool

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).

* [worker] Change background appender channel to be memory-bound instead 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.

Author: MohamedBassem

Cargo.lock

@@ -25,6 +25,7 @@ use restate_types::live::LiveLoadExt;
 use restate_core::MetadataWriter;
 use restate_core::my_node_id;
 use restate_core::{Metadata, ShutdownError};
+use restate_memory::NonZeroByteCount;
 use restate_types::config::Configuration;
 use restate_types::logs::metadata::SealMetadata;
 use restate_types::logs::metadata::{LogletParams, Logs, SegmentIndex};
@@ -206,16 +207,17 @@ impl Bifrost {
         ))
     }
 
+    /// See [`BackgroundAppender::new`] for the semantics of `memory_limit`.
     pub fn create_background_appender<T: StorageEncode>(
         &self,
         log_id: LogId,
         error_recovery_strategy: ErrorRecoveryStrategy,
-        queue_capacity: usize,
+        memory_limit: Option<NonZeroByteCount>,
         max_batch_size: usize,
     ) -> Result<BackgroundAppender<T>> {
         Ok(BackgroundAppender::new(
             self.create_appender(log_id, error_recovery_strategy)?,
-            queue_capacity,
+            memory_limit,
             max_batch_size,
         ))
     }
@@ -1359,16 +1361,16 @@ mod tests {
         let bifrost = Bifrost::init_in_memory(env.metadata_writer).await;
 
         let background_appender: crate::BackgroundAppender<String> = bifrost
-            .create_background_appender(LogId::new(0), ErrorRecoveryStrategy::Wait, 10, 10)?;
+            .create_background_appender(LogId::new(0), ErrorRecoveryStrategy::Wait, None, 10)?;
 
         let mut handle = background_appender.start("test-appender")?;
         let sender = handle.sender();
 
         // A string with 100 bytes
         let payload = String::from_utf8(vec![b't'; 100]).unwrap();
 
-        // try_enqueue should fail with RecordTooLarge
-        let result = sender.try_enqueue(payload.clone());
+        // enqueue should fail with RecordTooLarge
+        let result = sender.enqueue(payload.clone());
         assert_that!(
             result,
             pat!(Err(pat!(EnqueueError::RecordTooLarge {
@@ -1378,7 +1380,7 @@ mod tests {
         );
 
         // enqueue (async) should also fail with RecordTooLarge
-        let result = sender.enqueue(payload.clone()).await;
+        let result = sender.enqueue(payload.clone());
         assert_that!(
             result,
             pat!(Err(pat!(EnqueueError::RecordTooLarge {
@@ -1387,8 +1389,8 @@ mod tests {
             })))
         );
 
-        // try_enqueue_with_notification should also fail
-        let result = sender.try_enqueue_with_notification(payload.clone());
+        // enqueue_with_notification should also fail
+        let result = sender.enqueue_with_notification(payload.clone());
         assert!(matches!(
             result,
             Err(EnqueueError::RecordTooLarge {
@@ -1415,14 +1417,14 @@ mod tests {
         let bifrost = Bifrost::init_in_memory(env.metadata_writer).await;
 
         let background_appender: crate::BackgroundAppender<String> = bifrost
-            .create_background_appender(LogId::new(0), ErrorRecoveryStrategy::Wait, 10, 10)?;
+            .create_background_appender(LogId::new(0), ErrorRecoveryStrategy::Wait, None, 10)?;
 
         let mut handle = background_appender.start("test-appender")?;
         let sender = handle.sender();
 
         // With a 10KB limit, the ~2KB estimated record should succeed
         let payload = "test".to_string();
-        sender.enqueue(payload).await?;
+        sender.enqueue(payload)?;
 
         // Drain and wait for commit
         handle.drain().await?;
@@ -1442,29 +1444,24 @@ mod tests {
         let bifrost = Bifrost::init_in_memory(env.metadata_writer).await;
 
         let background_appender: crate::BackgroundAppender<String> = bifrost
-            .create_background_appender(LogId::new(0), ErrorRecoveryStrategy::Wait, 1000, 100)?;
+            .create_background_appender(LogId::new(0), ErrorRecoveryStrategy::Wait, None, 100)?;
 
         let mut handle = background_appender.start("test-appender")?;
         let sender = handle.sender();
 
-        // Rapidly enqueue many records using try_enqueue (non-blocking)
+        // Rapidly enqueue many records using enqueue
         let mut enqueued = 0;
         for i in 0..100 {
-            match sender.try_enqueue(format!("rapid-record-{i}")) {
-                Ok(()) => enqueued += 1,
-                Err(EnqueueError::Full(_)) => {
-                    // Queue is full, use async enqueue
-                    sender.enqueue(format!("rapid-record-{i}")).await?;
-                    enqueued += 1;
-                }
+            match sender.enqueue(format!("rapid-record-{i}")) {
+                Ok(_) => enqueued += 1,
                 Err(e) => return Err(e.into()),
             }
         }
 
         assert_that!(enqueued, eq(100));
 
         // Wait for all to be committed
-        let token = sender.notify_committed().await?;
+        let token = sender.notify_committed()?;
         token.await?;
 
         handle.drain().await?;

crates/bifrost/src/background_appender.rs

@@ -27,7 +27,9 @@ mod types;
 mod watchdog;
 
 pub use appender::Appender;
-pub use background_appender::{AppenderHandle, BackgroundAppender, CommitToken, LogSender};
+pub use background_appender::{
+    AppenderHandle, BackgroundAppender, CommitToken, EnqueueWithNotificationResult, LogSender,
+};
 pub use bifrost::{Bifrost, ErrorRecoveryStrategy};
 pub use bifrost_admin::{BifrostAdmin, MaybeSealedSegment};
 pub use data_record::{DataRecord, DataRecordError};

crates/bifrost/src/bifrost.rs

@@ -154,6 +154,19 @@ pub struct WorkerOptions {
     #[cfg_attr(feature = "schemars", schemars(skip))]
     #[serde(skip_serializing_if = "std::ops::Not::not", default)]
     pub use_multi_db_layout: bool,
+
+    /// # Self-proposal queue memory limit
+    ///
+    /// The amount of memory a partition leader may use to buffer commands it proposes to its
+    /// own log (timers, invoker effects, RPCs, etc.) before pushing back on their producers.
+    /// Larger values improve append batching and throughput at the cost of memory usage and
+    /// commit tail latency. The limit applies to each partition individually.
+    ///
+    /// Default: 64 MiB
+    ///
+    /// Since v1.7.3
+    #[cfg_attr(feature = "schemars", schemars(skip))]
+    pub self_proposal_queue_memory_limit: NonZeroByteCount,
 }
 
 impl WorkerOptions {
@@ -217,6 +230,9 @@ impl Default for WorkerOptions {
             ),
             rule_book_poll_interval: NonZeroFriendlyDuration::from_secs_unchecked(30),
             use_multi_db_layout: false,
+            self_proposal_queue_memory_limit: NonZeroByteCount::new(
+                NonZeroUsize::new(64 * 1024 * 1024).expect("non zero"),
+            ),
         }
     }
 }