NodeQueue

NodeQueue: the per-peer outbound queue with credit-based flow control (distributed actors 2/4, step 3; design record: specs/progress/2026-09-15-credit-based-flow-control.md).

A remote send is a write to THIS queue, never to the socket. One writer actor per peer owns the data connection for writing: it sends a frame only while the receiver has granted credit for it, so the kernel buffer never fills behind a peer that stopped consuming and a stalled peer is visible here, as queue depth, rather than as a green thread stuck in write().

Budget: bytes queued but not yet written, bounded by the queue's budget. The check is made by the SENDER, synchronously, through an atomic Vault.incr, so drop_new can refuse at once; the writer actor only ever sees frames that fit. Credit: bytes the receiver has CONSUMED (handed to a mailbox, or answered with DELIVERY_FAILED) and announced with a CREDIT frame on the control connection; the writer holds a frame until credit covers it.

Policies at the budget, the three the local bounded mailbox has: drop_new refuse the new frame (Err(Backpressure)) drop_old admit it, evicting the oldest queued frames until it fits; the evicted seqs are reported through take_evicted block_sender the caller blocks until the frame is admitted or its timeout passes (BlockSender(timeout_ms), below)

Types

typePolicyPolicy = DropNew | DropOld | BlockSender(Int)#

typeEnqueueErrorEnqueueError = Backpressure | NoConnection#

typeWriterStateWriterState = { fd : Int, granted : Int, sent : Int, pending : Deque.Deque((Int, Bytes)), evicted : List(Int), budget : Int, dead : Bool,#

typeWriterReqWriterReq = EvictedReq | WaitReq#

Functions

fncastcast(q, seq : Int, to : Pid, type_tag : String, payload : List(Int), policy : Policy) : Result((), EnqueueError)#

fnconsumedconsumed(control_fd : Int, budget : Int, len : Int) : Int#

fncreditcredit(q, total_consumed : Int) : ()#

fndecode_creditdecode_credit(bytes : List(Int)) : Result(Int, String)#

fndepthdepth(q) : Int#

fnencode_creditencode_credit(total_consumed : Int) : List(Int)#

fnenqueueenqueue(q, seq : Int, frame : Bytes, policy : Policy) : Result((), EnqueueError)#

fnstartstart(fd : Int, budget : Int)#

fntag_credittag_credit() : Int do 11 end#

fntake_evictedtake_evicted(q) : List(Int)#