Fix performance regression with --server-memory-limit 0

## Related
* Closes #12626

## What
0.29 changed the size of the channel for the gRPC proxy channel to be
dependent on the `--server-memory-limit`.
The thinking was to better heed the total memory limit.
However, using a `server_memory_limit` of exactly `0` is common, and
even encouraged in our docs. Having zero bytes buffer on the incoming
channel leads to a lot of early backpressure, making the sender block,
degrading performance.

This PR changes the channel to always have a fixed 256 MiB of buffer
space. This trades performance for (potential) latency.

In future versions of Rerun we should consider making this value
configurable, but doing so now would be a breaking change, and I would
like to get this PR into a patch release.

Previous to 0.29, we also had a fixed size input buffer, but measured in
number of messages instead of number of bytes. This meant the server
could easily balloon in memory use when logging large things. I believe
having it defined in terms of bytes makes more sense, but there is
_also_ a fixed upper-bound on the number of messages in the new code
(4096).

Source-Ref: be1671f8c672ccf5d743b21baf8984aa20eec150

Author: emilk

crates/store/re_grpc_server/src/lib.rs

@@ -40,14 +40,23 @@ pub const DEFAULT_SERVER_PORT: u16 = 9876;
 pub const MAX_DECODING_MESSAGE_SIZE: usize = u32::MAX as usize;
 pub const MAX_ENCODING_MESSAGE_SIZE: usize = MAX_DECODING_MESSAGE_SIZE;
 
+/// Maximum number of messages in the input queue.
+const CHANNEL_SIZE_MESSAGES: usize = 1024; // TODO(emilk): move into `ServerOptions` after the patch release.
+
+/// Make sure we can handle a quick burst of messages without blocking,
+/// even if the server has a [`ServerOptions::memory_limit`] of zero.
+const CHANNEL_SIZE_BYTES: u64 = 128 * 1024 * 1024; // TODO(emilk): move into `ServerOptions` after the patch release.
+
 /// Options for the gRPC Proxy Server
 #[derive(Clone, Copy, Debug)]
 pub struct ServerOptions {
     /// When a client connect, should they be sent the oldest data first, or the newest?
     pub playback_behavior: PlaybackBehavior,
 
-    /// Start garbage collecting old data when we reach this.
-    pub memory_limit: MemoryLimit,
+    /// Limit on how much history the server saves.
+    ///
+    /// It will start garbage collecting old data when we reach this.
+    pub memory_limit: MemoryLimit, // TODO(emilk): rename `history_limit`
 }
 
 impl Default for ServerOptions {
@@ -871,26 +880,17 @@ impl MessageProxy {
     }
 
     fn new_with_recv(
-        mut options: ServerOptions,
+        options: ServerOptions,
     ) -> (Self, async_broadcast_channel::Receiver<LogOrTableMsgProto>) {
-        // Divide up the memory budget:
-        let (broadcast_channel_memory_limit, rest_memory_limit) = options.memory_limit.split(0.25);
-        options.memory_limit = rest_memory_limit;
-
         let (broadcast_log_tx, broadcast_log_rx) = async_broadcast_channel::channel(
             "re_grpc_server broadcast",
-            4096,
-            broadcast_channel_memory_limit.as_bytes(),
+            CHANNEL_SIZE_MESSAGES,
+            CHANNEL_SIZE_BYTES,
         );
 
         let (event_tx, event_rx) = {
-            let message_queue_capacity = if options.memory_limit == MemoryLimit::ZERO {
-                1
-            } else {
-                16 // Apply backpressure early
-            };
             // TODO(emilk): this could also use a size-based backpressure mechanism.
-
+            let message_queue_capacity = 32; // Apply backpressure early
             async_mpsc_channel::channel("re_grpc_server events", message_queue_capacity)
         };
 

crates/utils/re_quota_channel/src/async_broadcast_channel.rs

@@ -404,12 +404,24 @@ pub fn channel<T: Clone>(
     max_messages: usize,
     max_bytes: u64,
 ) -> (Sender<T>, Receiver<T>) {
+    let debug_name = debug_name.into();
     let max_messages = max_messages.max(1); // Ensure we allow at least 1 message
 
+    if max_bytes == 0 {
+        let msg = format!(
+            "Channel '{debug_name}' has a memory limit of 0 bytes. Consider giving it at least a few MiB so that it can handle a quick burst of messages without blocking."
+        );
+        if cfg!(debug_assertions) {
+            re_log::warn_once!("DEBUG WARNING: {msg}");
+        } else {
+            re_log::debug_once!("{msg}");
+        }
+    }
+
     let (inner, inner_rx) = broadcast::channel(max_messages);
 
     let state = Arc::new(ChannelState {
-        debug_name: debug_name.into(),
+        debug_name,
         locked: Mutex::new(Locked {
             bytes_in_flight: 0,
             messages_in_flight: 0,