NodeQueue

BlockSender(timeout_ms): the caller BLOCKS until the frame is admitted or the deadline passes (then Err(Backpressure), and the frame is withdrawn). Rides Actor.call on the writer -- no runtime primitive. The writer keeps a FIFO line of waiting frames and a FIFO line of blocked callers, one entry each per call; every admission from the line answers the oldest caller. So Ok means: the writer has admitted a frame at or ahead of yours, and yours is admitted in order -- the same promise DropNew's Ok makes (queued, not delivered). A call that times out withdraws its own frame; if admission raced the deadline the frame may still go out. Under the interpreter Actor.call cannot park the caller across later credit, so a blocking send that cannot be admitted at once returns Err(Backpressure) there (as Actor.set_queue_limit policy 3 is refused); the compiled backend blocks for real.

NotAuthorized is ClusterNode's (step 11b): a raw send refused because this node's or the peer's certificate lacks raw_send. A queue itself never returns it.

sent is a total; granted = the receiver's last announced consumed total + the budget, so at most one budget of bytes is ever in flight (written but not yet consumed). A frame may go when sent + len <= granted, with one exception for a frame the window can never cover (may_write). dead: a write failed (the connection is gone). Nothing further is written and nothing already queued is charged off the budget: a failed write is not a sent frame.

Actor.call sentinels: constructor index = handler index (see Writer).

An ACTOR_MSG for message [seq] to [to] through the queue [q] under [policy]: NodeSend's frame, NetFrame-encoded, as Bytes (enqueue). The remote-send path a program should use once it has a queue for the peer; NodeSend.cast writes straight to a socket. No certificate-mode raw-send check is made here (ClusterNode makes it before queueing).

cast with the message type's schema hash [schema] in the frame (build step 9; `\

The connection is gone (its owner shut it down): [q]'s writer stops writing, drops what is queued, and answers every caller blocked in BlockSender with NoConnection now. Later frames are never written. Does not close the fd (see close_fd). Asynchronous: returns at once.

close, and close the connection's fd through its writer, in mailbox order: every frame the writer has written before this message is on the wire, and nothing is written after. Frames still waiting for credit are dropped. For a local queue (start_local) the sink is dropped instead: nothing is handed to it afterwards. Asynchronous: returns at once.

The receiver's half: account [len] consumed bytes from the peer whose control connection is [control_fd]. Whenever at least budget/4 bytes have been consumed since the last announcement, send the running total in a CREDIT frame on [control_fd] (a failed write is ignored). Returns the total consumed so far on that connection.

Apply a CREDIT frame's [total_consumed] to [q]'s writer: it may now have up to total_consumed + budget bytes written in all, and writes what that allows. A total lower than one already applied changes nothing. Asynchronous: returns at once.

match NodeQueue.decode_credit(frame) do
  Ok(total) -> NodeQueue.credit(q, total)
  Err(_) -> ()
end

Decode a CREDIT frame payload into its consumed total. Err for bytes that are not MessagePack or not a CREDIT.

Bytes queued on [q] and not yet written (the monitoring probe); 0 for a queue never started.

A CREDIT frame payload (MessagePack, not yet length-framed) announcing [total_consumed]: the receiver's TOTAL consumed byte count, not a delta, so a duplicated or replayed CREDIT never over-credits.

Queue one already-framed [frame] (message [seq]) on [q] under [policy]. Returns as soon as the frame is admitted or refused (a BlockSender call waits up to its timeout); the write happens in the writer's turn, when credit allows. Ok(()) means admitted, not written.

The frame is admitted outright when it fits the budget, or when nothing is queued. Otherwise: DropNew returns Err(Backpressure); DropOld admits it and evicts the oldest queued frames until it fits (see take_evicted); Unbounded admits it past the budget; BlockSender(ms) waits for room, then Err(Backpressure) on timeout (the frame is withdrawn) or Err(NoConnection) when the queue is closed meanwhile. On a dead or closed connection a frame can still be admitted (Ok) but is never written.

match NodeQueue.enqueue(q, 7, frame, NodeQueue.BlockSender(500)) do
  Ok(_) -> "queued"
  Err(NodeQueue.Backpressure) -> "peer too slow"
  Err(_) -> "connection gone"
end

Start a writer actor for the data connection [fd] with a byte [budget], and return the queue handle (writer, fd) the other functions take. At most [budget] bytes are queued (see the policies) and at most one budget is in flight before the receiver's first CREDIT. One queue per fd: a second start on the same fd resets the shared byte counters. (close_fd(q) later closes the fd through the writer.)

let q = NodeQueue.start(peer.data_fd, 1048576)

A LOCAL queue: the same writer and admission as start, but each frame is handed to [sink] (in the writer's turn, in queue order) instead of written to a socket. There is no receiver to grant credit, so it is bypassed: the budget is effectively unbounded (2^60 bytes). The cluster node's loopback link (a node sending to itself, ClusterNode.queue_for(h, own_id)) is one. Each call allocates a new local id (-2, -3, ...); close_fd drops the sink.

let q = NodeQueue.start_local(fn frame -> handle(frame))

The frame tag of a CREDIT: 11.

The seqs DropOld has evicted from [q] since the last call, oldest first, and forget them. Asks the writer (waits up to 5 s); [] when it does not answer in time.