fix(gossip): bound broadcast queue to prevent memory leaks

- Add max_gossip_queue_depth to config (default 5000)
- Enforce limit in Dissemination.enqueue by dropping oldest items (ring buffer)
- Prevents unbounded memory growth when broadcast rate exceeds gossip bandwidth

lib/dissemination.ml

···
 
 let create () = { queue = Kcas_data.Queue.create (); depth = Kcas.Loc.make 0 }
 
-let enqueue t msg ~transmits ~created =
+let enqueue t msg ~transmits ~created ~limit =
   let item = { msg; transmits = Kcas.Loc.make transmits; created } in
   Kcas.Xt.commit
     {
       tx =
         (fun ~xt ->
-          Kcas_data.Queue.Xt.add ~xt item t.queue;
-          Kcas.Xt.modify ~xt t.depth succ);
+          let d = Kcas.Xt.get ~xt t.depth in
+          if d >= limit then ignore (Kcas_data.Queue.Xt.take_opt ~xt t.queue)
+          else Kcas.Xt.set ~xt t.depth (d + 1);
+          Kcas_data.Queue.Xt.add ~xt item t.queue);
     }
 
 let depth t = Kcas.Xt.commit { tx = (fun ~xt -> Kcas.Xt.get ~xt t.depth) }

lib/dissemination.mli

···
 type t
 
 val create : unit -> t
-val enqueue : t -> protocol_msg -> transmits:int -> created:Mtime.span -> unit
+
+val enqueue :
+  t -> protocol_msg -> transmits:int -> created:Mtime.span -> limit:int -> unit
+
 val depth : t -> int
 
 val drain :

lib/protocol.ml

···
     Protocol_pure.retransmit_limit t.config
       ~node_count:(Membership.count t.members)
   in
-  Dissemination.enqueue t.broadcast_queue msg ~transmits ~created:(now_mtime t);
+  Dissemination.enqueue t.broadcast_queue msg ~transmits ~created:(now_mtime t)
+    ~limit:t.config.max_gossip_queue_depth;
   Dissemination.invalidate t.broadcast_queue
     ~invalidates:Protocol_pure.invalidates msg
 

lib/types.ml

···
   encryption_enabled : bool;
   gossip_verify_incoming : bool;
   gossip_verify_outgoing : bool;
+  max_gossip_queue_depth : int;
 }
 
 let default_config =
···
     encryption_enabled = false;
     gossip_verify_incoming = true;
     gossip_verify_outgoing = true;
+    max_gossip_queue_depth = 5000;
   }
 
 type 'a env = {

lib/types.mli

···
   encryption_enabled : bool;
   gossip_verify_incoming : bool;
   gossip_verify_outgoing : bool;
+  max_gossip_queue_depth : int;
 }
 
 val default_config : config

test/generators.ml

···
   and+ label = oneof [ return ""; gen_topic ]
   and+ encryption_enabled = bool
   and+ gossip_verify_incoming = bool
-  and+ gossip_verify_outgoing = bool in
+  and+ gossip_verify_outgoing = bool
+  and+ max_gossip_queue_depth = int_range 10 10000 in
   {
     bind_addr;
     bind_port;
···
     encryption_enabled;
     gossip_verify_incoming;
     gossip_verify_outgoing;
+    max_gossip_queue_depth;
   }
 
 let gen_decode_error : decode_error QCheck.Gen.t =