This walkthrough follows application bytes through an established TransportStream. The forward path begins when the session initiator writes to the stream, divides the byte sequence into Sphinx-sized transport frames and sends those frames to the session recipient. The recipient restores the ordered byte stream, exposes the bytes to the mounted libp2p protocol and acknowledges the received sequences. The walkthrough then follows the reverse path, where the recipient forms a temporary redundancy batch from individual SURBs supplied by the initiator and sends the same Data or ACK frame through every SURB in that batch. If the recipient does not have enough SURBs for that batch, the reverse send waits while the initiator-driven supply mechanism restores the queue.
The implementation is divided across four modules:
wire.nimdefinesData,Ackand the SURB supply snapshot carried by reverse frames. The module also defines the fixed-width stream and sequence types, derives one maximum Data payload size and validates both fixed-size acknowledgement bitmaps.streams.nimowns sequence numbers, retained outbound chunks, the receive window, its bitmap and the events used by the flow tasks.sessions.nimowns the recipient’s bounded queue of individual SURBs, numbered supply state, liveness-probe timing and the lock that serializes return sends within one session.transport.nimconnects the standard libp2pConnectionmethods to those state machines. The module selects forward Mix delivery for frames sent by the session initiator and SURB delivery for frames sent by the session recipient.
The current implementation bounds memory, propagates application backpressure, retransmits unacknowledged Data and replenishes recipient SURBs through an initiator-driven push mechanism. Data retransmission is enabled by default and can be disabled when constructing MixTransport. SURB replenishment is always active. The transport does not probe a stalled Data receive window when every ACK carrying the advanced window has been lost.
The State Behind One Virtual Connection
TransportStream inherits from libp2p’s BufferStream. The same object therefore contains the transport state and the connection read buffer passed to the application protocol.
The receiver accepts Data sequence numbers only within a fixed-size receive window. The receive base is the lowest sequence number in that window and, equivalently, the next sequence that has not yet been passed in order into BufferStream. Sequence numbers below the receive base have entered BufferStream, although the application may not have read all of those bytes yet. The local field receiveBase stores this value and starts at 1. SequenceNumber.high is reserved as the terminal receive-base value, so the last valid Data sequence is MaxDataSequenceNumber = SequenceNumber.high - 1. After passing that final Data sequence to BufferStream, the receiver can still represent the next expected position without wrapping to zero.
Each ACK reports the receiver’s current receive base to the sender. The sender stores the most recently reported value in remoteReceiveBase. The local receiveBase therefore describes this endpoint’s incoming Data, while remoteReceiveBase records how far the other endpoint has progressed when passing Data sent by this endpoint into its BufferStream. The sender combines remoteReceiveBase with the receive-window size to determine the highest sequence number that the receiver currently permits.
The outbound part of a stream records nextOutboundSequence, remoteReceiveBase, and the chunks that have been submitted but not acknowledged. Each retained OutboundChunk contains the payload and an optional retransmission deadline. The deadline is absent while the initial transmission or a retransmission is being submitted. After submission completes, the transport sets the next deadline unless an ACK has already removed the chunk.
The outbound part owns two events. An application writer waits on sendStateChanged when the stream cannot allocate another sequence. Applying a received ACK signals sendStateChanged after removing acknowledged chunks or advancing remoteReceiveBase. Removing a chunk that could not be submitted and closing the stream also signal sendStateChanged. The retransmission task waits on retransmissionStateChanged when no deadline exists or when the next deadline is still in the future. Scheduling a deadline, removing an acknowledged chunk and closing the stream signal retransmissionStateChanged, causing the task to inspect the current table state again.
The inbound part records receiveBase, a bitmap describing received positions at and above receiveBase, and the out-of-order payloads retained inside the receive window. Retaining a payload means inserting the payload into pendingInbound under its sequence number and setting the corresponding bit in the acknowledgement bitmap. The payload remains in pendingInbound until all preceding sequences have been passed to BufferStream and the payload becomes the next one eligible for ordered delivery.
The inbound part also owns two events. dataAvailable tells the ordered-delivery task to check whether the payload for receiveBase is available. The transport signals dataAvailable after receiveData accepts and retains any in-window payload, including a payload that arrived out of order. After advancing receiveBase, the transport signals dataAvailable again when pendingInbound already contains the payload for the new base. Closing the stream also signals dataAvailable so that the delivery task can terminate. A dataAvailable signal therefore does not assert that the required payload is present - meaning it does not assert that payload at the receiveBase has been received - it signals that some payload within the receive window has been received.
shouldSendAck tells the ACK task to capture and send the current receiveBase and acknowledgement bitmap. The transport signals shouldSendAck after accepting an in-window payload, after receiving duplicate Data, and after advancing receiveBase. Closing the stream also signals shouldSendAck; the awakened ACK task checks stream.closed and terminates without sending another ACK.
The fields are declared together on TransportStream:
TransportStream* = ref object of BufferStream
sessionId: PeerId
streamId: StreamId
codec: string
direction: StreamDirection
state: StreamState
writeHandler: StreamWriteHandler
writeLock: AsyncLock
nextOutboundSequence: SequenceNumber
remoteReceiveBase: SequenceNumber
pendingOutbound: Table[SequenceNumber, OutboundChunk]
sendStateChanged: AsyncEvent
retransmissionStateChanged: AsyncEvent
receiveBase: SequenceNumber
acknowledgementBitmap: seq[byte]
pendingInbound: Table[SequenceNumber, seq[byte]]
dataAvailable: AsyncEvent
shouldSendAck: AsyncEventThe outbound and inbound state describe different directions of the same bidirectional stream. pendingOutbound contains chunks sent by this endpoint and awaiting acknowledgement. pendingInbound contains chunks received from the remote endpoint but not yet passed in order into this endpoint’s BufferStream.
The four events carry notifications rather than the associated state. After waking, a task checks the corresponding counters, tables or bitmap again. A notification means “check again,” not “capacity definitely exists” or “a particular chunk is definitely available.” The waiting loops therefore re-evaluate their conditions after every wake-up.
Construction starts both sequence spaces at 1 and allocates the fixed bitmap:
nextOutboundSequence: 1,
remoteReceiveBase: 1,
pendingOutbound: initTable[SequenceNumber, OutboundChunk](),
sendStateChanged: newAsyncEvent(),
retransmissionStateChanged: newAsyncEvent(),
receiveBase: 1,
acknowledgementBitmap: newSeq[byte](AckBitmapBytes),
pendingInbound: initTable[SequenceNumber, seq[byte]](),
dataAvailable: newAsyncEvent(),
shouldSendAck: newAsyncEvent(),configureStream installs the callback used by TransportStream.write and starts one ordered-delivery task and one acknowledgement task. When enableDataRetransmissions is true, configureStream also starts the stream’s retransmission task:
proc configureStream(
self: MixTransport, session: TransportSession, stream: TransportStream
) =
let writeHandler: StreamWriteHandler = proc(
data: sink seq[byte]
): Future[void] {.async: (raw: true, raises: [CancelledError, LPStreamError]).} =
self.writeStream(session, stream, move(data))
stream.setWriteHandler(writeHandler)
stream.trackStreamTask(runInboundDelivery(stream))
stream.trackStreamTask(runAcknowledgements(self, session, stream))
if self.dataRetransmissionsEnabled:
stream.trackStreamTask(runRetransmissions(self, session, stream))An outbound stream is configured after its StreamAck arrives. At the recipient, an inbound stream is configured immediately before it is established and passed to the mounted protocol. A rejected stream never starts these tasks. The tasks are stored on the stream whose state and events they use; they are not stored in a transport-wide task collection.
1. An Application Write Enters MixTransport
The application uses a normal libp2p connection. In this example it calls the inherited LPStream.writeLp method:
await initiatorStream.writeLp(TestRequest)writeLp constructs one byte sequence containing the varint length prefix followed by TestRequest, then passes that sequence to the virtual write method:
method writeLp*(
s: LPStream, msg: openArray[byte]
): Future[void] {.base, async: (raises: [CancelledError, LPStreamError], raw: true).} =
let vbytes = PB.toBytes(msg.len().uint64)
var buf = newSeqUninit[byte](msg.len() + vbytes.len)
buf[0 ..< vbytes.len] = vbytes.toOpenArray()
buf[vbytes.len ..< buf.len] = msg
s.write(move(buf))Because s is a TransportStream, dynamic dispatch selects TransportStream.write for the final call. The application therefore invokes writeLp, while the transport implements the lower-level write operation that receives the complete length-prefixed byte sequence.
Before the stream is exposed to the application, configureStream assigns stream.writeHandler a closure that captures the MixTransport, TransportSession and TransportStream objects associated with this virtual connection. Calling that closure invokes self.writeStream(session, stream, ...). The write handler is therefore the adapter between the application-facing TransportStream.write method and the transport’s frame-producing MixTransport.writeStream procedure.
TransportStream.write checks that configureStream installed this handler, serializes complete application writes with writeLock, and invokes the handler with the application bytes:
method write*(
stream: TransportStream, msg: sink seq[byte]
): Future[void] {.async: (raises: [CancelledError, LPStreamError]).} =
if stream.writeHandler.isNil:
raise newException(LPStreamError, "MixTransport stream is not writable")
await stream.writeLock.acquire()
defer:
stream.writeLock.release()
await stream.writeHandler(move(msg))The lock covers one complete application write. Two application tasks can call write concurrently on the same stream; for example, a protocol can produce a response while another task writes a control message. Without the lock, writeStream could alternate between those calls while assigning sequence numbers, so chunks from the two byte sequences could be interleaved on the receiving side. Applying received ACKs and delivering inbound Data do not use writeLock; both operations continue while one writer waits for capacity or Mix delivery.
Application write boundaries are not encoded on the wire. The receiver reconstructs one ordered byte stream, which is what readOnce, readExactly and readLp expect.
2. The Writer Waits for Capacity
Before taking bytes from the application write, writeStream waits until the stream may allocate another outbound sequence:
proc writeStream(
self: MixTransport,
session: TransportSession,
stream: TransportStream,
data: sink seq[byte],
): Future[void] {.async: (raises: [CancelledError, LPStreamError]).} =
if stream.state != StreamState.Established:
raise newException(LPStreamError, "MixTransport stream is not established")
var offset = 0
while offset < data.len:
await stream.waitForOutboundCapacity()
let chunkLength = min(MaxDataPayloadBytes, data.len - offset)
# Chunk reservation and frame submission continue below.To avoid polling while capacity is unavailable, waitForOutboundCapacity suspends the application writer on sendStateChanged. The waitForOutboundCapacity procedure is defined in streams.nim and repeatedly evaluates canReserveOutbound, the stream’s predicate for whether another chunk may be allocated:
proc waitForOutboundCapacity*(
stream: TransportStream
): Future[void] {.async: (raises: [CancelledError, LPStreamError]).} =
while true:
if stream.closed:
raise newLPStreamClosedError()
if stream.nextOutboundSequence > MaxDataSequenceNumber:
raise newException(LPStreamError, "stream sequence space is exhausted")
if stream.canReserveOutbound:
return
stream.sendStateChanged.clear()
await stream.sendStateChanged.wait()Chronos schedules these tasks cooperatively. Clearing sendStateChanged and registering the wait happen without an intervening await, so another task cannot change capacity between those operations. After sendStateChanged wakes the writer, the loop checks closure, sequence exhaustion and capacity again. Checking stream.closed before capacity prevents a closed stream from returning successfully merely because its outbound table still has room. Reporting sequence exhaustion before waiting prevents a writer from sleeping forever after the final sequence has been allocated.
The predicate used by that loop combines two limits:
func canReserveOutbound*(stream: TransportStream): bool =
stream.pendingOutbound.len < MaxInflightChunks and
stream.nextOutboundSequence <= MaxDataSequenceNumber and
stream.nextOutboundSequence < stream.remoteReceiveLimitMaxInflightChunks is currently 64. It bounds the sent payloads retained while awaiting acknowledgement. The remote receive limit is derived from the latest base reported by the other endpoint:
func remoteReceiveLimit*(stream: TransportStream): SequenceNumber =
let window = SequenceNumber(ReceiveWindowChunks)
if stream.remoteReceiveBase > SequenceNumber.high - window:
SequenceNumber.high
else:
stream.remoteReceiveBase + windowIf the remote base is 20, its 256-position window admits sequences 20 through 275; 276 is the exclusive limit. The sender can stop because it already has 64 unacknowledged chunks or because its next sequence falls outside that remote window.
3. Every Data Frame Uses One Payload Bound
wire.nim names the two numeric domains instead of spreading primitive integer types through the transport:
type
StreamId* = uint32
SequenceNumber* = uint32
SurbSupplySequence* = uint32
const MaxDataSequenceNumber* = SequenceNumber.high - 1The frame encodes stream IDs, Data sequence numbers, ACK receive bases and SURB supply sequence values as Protobuf fixed32 fields:
streamId* {.fieldNumber: 4, fixed.}: Opt[StreamId]
sequence* {.fieldNumber: 5, fixed.}: Opt[SequenceNumber]
receiveBase* {.fieldNumber: 8, fixed.}: Opt[SequenceNumber]
firstSurbSequence* {.fieldNumber: 10, fixed.}: Opt[SurbSupplySequence]
surbSupplyReceiveBase* {.fieldNumber: 11, fixed.}: Opt[SurbSupplySequence]
surbSupplyLimit* {.fieldNumber: 13, fixed.}: Opt[SurbSupplySequence]Each fixed numeric field occupies four value bytes regardless of the current identifier or sequence. Consequently, a stream that has just opened and a stream approaching sequence exhaustion use the same Data payload bound. The transport can calculate one payload bound for every Data frame; it does not need to encode candidate frames repeatedly to determine how many application bytes fit at a particular stream or sequence number.
The remaining variable-size Data field is sessionId. connect generates session pseudonyms with PeerId.random, which currently produces a 39-byte binary Peer ID. MaxSessionIdBytes records that wire constraint, and both session registration and frame validation reject a longer identifier.
MaxDataFrameOverheadBytes accounts for the Protobuf tags and encoded values of version, the maximum-size sessionId, kind, streamId, sequence, the payload length prefix and the complete SURB supply snapshot. The snapshot is included because Data sent from the session recipient reports its latest supply receipt and credit. Data sent from the initiator does not carry the snapshot, but using one conservative payload bound in both directions keeps chunking independent of session role.
MaxDataPayloadBytes subtracts that overhead from the Sphinx payload space available after Mix service framing:
const
MaxSessionIdBytes* = 39
# Every Data-frame field number fits in a one-byte Protobuf field tag.
ProtobufFieldTagBytes = 1
SingleByteVarintBytes = 1
MaxDataPayloadLengthPrefixBytes = 2
VersionFieldBytes = ProtobufFieldTagBytes + SingleByteVarintBytes
SessionIdFieldBytes =
ProtobufFieldTagBytes + SingleByteVarintBytes + MaxSessionIdBytes
FrameKindFieldBytes = ProtobufFieldTagBytes + SingleByteVarintBytes
StreamIdFieldBytes = ProtobufFieldTagBytes + sizeof(StreamId)
SequenceFieldBytes = ProtobufFieldTagBytes + sizeof(SequenceNumber)
DataPayloadHeaderBytes = ProtobufFieldTagBytes + MaxDataPayloadLengthPrefixBytes
SurbSupplyReceiveBaseFieldBytes =
ProtobufFieldTagBytes + sizeof(SurbSupplySequence)
SurbSupplyAckBitmapFieldBytes =
ProtobufFieldTagBytes + SingleByteVarintBytes + SurbSupplyAckBitmapBytes
SurbSupplyLimitFieldBytes =
ProtobufFieldTagBytes + sizeof(SurbSupplySequence)
MaxDataFrameOverheadBytes =
VersionFieldBytes + SessionIdFieldBytes + FrameKindFieldBytes +
StreamIdFieldBytes + SequenceFieldBytes + DataPayloadHeaderBytes +
SurbSupplyReceiveBaseFieldBytes + SurbSupplyAckBitmapFieldBytes +
SurbSupplyLimitFieldBytes
let MaxDataPayloadBytes* = MaxTransportFrameBytes - MaxDataFrameOverheadBytesThe expression uses sizeof(StreamId), sizeof(SequenceNumber) and sizeof(SurbSupplySequence) to account for their fixed-width representation. The frame validator rejects a larger payload, and encode independently checks the complete encoded frame size.
4. A Sequence Is Assigned and the Chunk Is Retained
Before sending a chunk, writeStream must assign the chunk a sequence number and retain its payload until an ACK confirms receipt. The implementation calls this combined operation a reservation. A successful reservation increments nextOutboundSequence and adds the payload to pendingOutbound under the assigned sequence number.
writeStream copies the next slice, calls reserveOutbound to perform that operation, and builds the wire frame:
proc writeStream(
self: MixTransport,
session: TransportSession,
stream: TransportStream,
data: sink seq[byte],
): Future[void] {.async: (raises: [CancelledError, LPStreamError]).} =
# Establishment check and capacity wait precede this excerpt.
let chunkLength = min(MaxDataPayloadBytes, data.len - offset)
var chunk = data[offset ..< offset + chunkLength]
let reservedSequence = stream.reserveOutbound(chunk).valueOr:
raise newException(LPStreamError, error)
let frame = MixTransportFrame(
version: MixTransportVersion,
sessionId: session.sessionId,
kind: FrameKind.Data,
streamId: Opt.some(stream.streamId),
sequence: Opt.some(reservedSequence),
payload: Opt.some(move(chunk)),
)The reserveOutbound implementation first repeats the capacity check, then increments the sequence and retains the payload. The new chunk has no retransmission deadline because its initial submission has not completed:
proc reserveOutbound*(
stream: TransportStream, payload: seq[byte]
): Result[SequenceNumber, string] =
if stream.nextOutboundSequence > MaxDataSequenceNumber:
return err("stream sequence space is exhausted")
if not stream.canReserveOutbound:
return err("stream has no outbound capacity")
let sequence = stream.nextOutboundSequence
inc stream.nextOutboundSequence
stream.pendingOutbound[sequence] =
OutboundChunk(payload: payload, nextRetransmissionAt: Opt.none(Moment))
ok(sequence)The frame receives move(chunk), while pendingOutbound retains its own payload for possible retransmission. An ACK removes the retained OutboundChunk; immediate submission failure calls cancelOutbound.
The wire validator requires exactly the fields appropriate for Data:
proc validateFrame(
frame: MixTransportFrame, requireValidSurbEncoding: bool
): Result[void, string] =
# Validation shared by all frame kinds precedes these Data checks.
require frame.sequence.isSome == (frame.kind == FrameKind.Data),
"sequence does not match the frame kind"
require frame.payload.isSome == (frame.kind == FrameKind.Data),
"payload does not match the frame kind"
# Validation for the remaining frame-specific fields follows.After sendStreamFrame accepts the frame for Mix submission, writeStream schedules the first retransmission deadline when Data retransmission is enabled:
(await self.sendStreamFrame(session, frame)).isOkOr:
stream.cancelOutbound(reservedSequence)
raise newException(LPStreamError, error)
if self.dataRetransmissionsEnabled:
stream.scheduleOutboundRetransmission(
reservedSequence, self.dataRetransmissionTimeout
)Submission does not mean that the remote endpoint received the Data. The chunk remains in pendingOutbound until an ACK removes it. If an ACK arrives before sendStreamFrame returns, the ACK removes the chunk first and scheduleOutboundRetransmission finds no matching entry, so the completed initial submission does not recreate acknowledged state.
5. Session Role Selects the Delivery Path
sendStreamFrame is the common submission procedure for an already constructed MixTransportFrame. The procedure is not specific to Data frames. writeStream passes the Data frame constructed in Section 4 to sendStreamFrame. The runAcknowledgements task described in Section 10 uses the same procedure for each Ack frame. Centralizing this decision ensures that Data and ACK traffic follow the same delivery rule for a given session role.
sendStreamFrame selects the delivery path from the session role. The initiator encodes the frame and sends it through the ordinary forward Mix path. The recipient first selects a temporary SURB redundancy batch, attaches the current SURB supply snapshot, and then encodes the resulting frame for reverse delivery:
proc sendStreamFrame(
self: MixTransport, session: TransportSession, frame: MixTransportFrame
): Future[Result[void, string]] {.async: (raises: [CancelledError]).} =
case session.role
of SessionRole.Initiator:
let payload = frame.encode().valueOr:
return err("could not encode " & $frame.kind & " frame: " & error)
let destination = session.destination.valueOr:
return err("initiator session has no destination")
(await self.sendToDestination(destination, session.sessionId, payload)).isOkOr:
return err("could not send " & $frame.kind & " frame: " & error)
of SessionRole.Recipient:
await session.acquireReplySend()
defer:
session.releaseReplySend()
(await session.waitForReplySurbs(DefaultReplySurbRedundancy)).isOkOr:
return err(error)
var replyBatch = session.takeReceivedSurbs(DefaultReplySurbRedundancy).valueOr:
return err(error)
var replyFrame = frame
session.attachSurbSupplySnapshot(replyFrame)
let payload = replyFrame.encode().valueOr:
return err("could not encode " & $frame.kind & " frame: " & error)
(await self.sendWithSurbRedundancyBatch(replyBatch, payload)).isOkOr:
return err("could not send " & $frame.kind & " frame: " & error)
ok()When the local endpoint is the session initiator, the initiator knows the destination and submits the encoded frame through the forward Mix path. This path carries both Data written by the initiator and ACKs produced after the initiator receives reverse Data from the recipient.
When the local endpoint is the session recipient, the recipient does not have a forward destination for the anonymous initiator. The recipient removes DefaultReplySurbRedundancy individual SURBs from the session queue, treats those SURBs as a temporary redundancy batch, and submits the same encoded frame through every SURB in that batch. This path carries both response Data written by the recipient’s application handler and ACKs produced after the recipient receives forward Data from the initiator.
acquireReplySend serializes all return sends belonging to the same session. The session has one shared queue of individual SURBs, so an application response and an ACK task must not concurrently inspect and consume that queue. The recipient waits until one complete temporary redundancy batch is available and then removes that batch. Removing SURBs increases the absolute supply limit; attaching the snapshot to the same reverse frame reports that replacement credit to the initiator.
The initiator-driven replenishment mechanism is documented in Mix Transport SURB Replenishment Strategy and mapped to code in Mix Transport Implementation Walk Through - SURB Replenishment. This walkthrough relies on its resulting contract: one reverse transport frame consumes DefaultReplySurbRedundancy session-owned SURBs, and Data or ACK transmission waits when the queue cannot provide that complete batch.
After the session recipient submits the same encoded frame through every SURB in the temporary batch, the explanation moves to the session initiator that receives those redundant replies. The initiator’s reply credential store contains one private credential for each SURB. Each arriving reply is recovered independently and consumes only its matching credential. Multiple recovered replies may therefore contain the same logical Data or ACK frame. Data sequence numbers and absolute ACK state make those frames idempotent: the first copy changes stream state, while later copies are recognized as duplicates and cannot deliver application bytes or remove outbound chunks twice.
6. Both Receive Paths Converge on handleData
Forward frames arrive through the registered Mix service handler and handleDelivery. SURB replies are recovered by handleRawSurbReply, decoded, checked against the session ID stored with the matching individual credential and passed to handleReplyFrame.
Both paths converge on handleData:
proc handleData(self: MixTransport, frame: MixTransportFrame) {.gcsafe, raises: [].} =
let session = self.sessions.get(frame.sessionId).valueOr:
return
if session.state != SessionState.Established:
return
let stream = session.getStream(frame.streamId.get()).valueOr:
return
if stream.state != StreamState.Established:
return
discard stream.receiveData(frame.sequence.get(), frame.payload.get())The (sessionId, streamId) pair selects the connection. Data for an unknown or non-established session or stream is ignored and cannot enter an unrelated connection.
7. receiveData Updates the Fixed Window
The receive window contains 256 positions, so the bitmap is exactly 32 bytes:
const
ReceiveWindowChunks* = 256
AckBitmapBytes* = ReceiveWindowChunks div 8
MaxInflightChunks* = 64Bit i represents receiveBase + i. With receiveBase = 10, bit zero represents sequence 10 and bit 255 represents sequence 265.
handleData passes the sequence number and payload to receiveData. The complete receiveData procedure rejects sequences outside the window, recognizes duplicates and retains every newly accepted payload:
proc receiveData*(
stream: TransportStream, sequence: SequenceNumber, payload: sink seq[byte]
): InboundDataDisposition =
if sequence > MaxDataSequenceNumber:
return InboundDataDisposition.OutsideWindow
if sequence < stream.receiveBase:
stream.fireShouldSendAckEvent()
return InboundDataDisposition.Duplicate
let offset = sequence - stream.receiveBase
if offset >= SequenceNumber(ReceiveWindowChunks):
return InboundDataDisposition.OutsideWindow
if stream.acknowledgementBitmap.bitmapContains(offset):
stream.fireShouldSendAckEvent()
return InboundDataDisposition.Duplicate
stream.pendingInbound[sequence] = move(payload)
stream.acknowledgementBitmap.setBitmapBit(offset)
stream.dataAvailable.fire()
stream.fireShouldSendAckEvent()
InboundDataDisposition.AcceptedA sequence below the base was already delivered in order. A set bit identifies a duplicate still inside the window. In both duplicate cases, the sender may be retrying because the previous ACK was lost, so the receiver signals shouldSendAck and sends its current snapshot again.
An above-window sequence cannot be produced by a compliant sender because the sender limits new sequences using the latest remoteReceiveBase reported by the receiver. The receiver classifies such a sequence as OutsideWindow, discards its payload and does not signal shouldSendAck. This prevents an invalid frame from causing unnecessary ACK traffic. If the receiver is the session recipient, sending that ACK would consume a temporary batch of SURBs. If the receiver is the session initiator, sending that ACK would consume a forward Mix delivery.
An accepted payload is inserted into pendingInbound and its acknowledgement-bitmap bit is set. If the chunk at receiveBase has not arrived, the receiver may still retain chunks with higher sequence numbers that arrived first. Those out-of-order chunks remain in pendingInbound until the missing earlier chunk arrives and ordered delivery can continue. The receiver accepts an out-of-order chunk only when its sequence number is lower than receiveBase + ReceiveWindowChunks, so every retained chunk remains inside the fixed receive window.
8. Ordered Delivery Feeds BufferStream
The preceding section explained how receiveData accepts an inbound Data frame, stores its payload in pendingInbound and fires the stream’s dataAvailable event. Storing the payload does not immediately expose it to the application because an earlier sequence may still be missing.
The configureStream procedure shown before Section 1 starts one runInboundDelivery task for every established stream. This task moves contiguous payloads from pendingInbound into the inherited BufferStream in sequence order. The same configureStream call also starts the ACK task described in Section 10.
runInboundDelivery starts as a background task and repeatedly calls takeNextInbound. The inner loop stops when takeNextInbound reports that the sequence currently required for ordered delivery is unavailable. The task then waits on dataAvailable. Accepting another inbound payload, or advancing receiveBase to a sequence that is already present, fires dataAvailable and wakes the task for another delivery attempt:
proc runInboundDelivery(stream: TransportStream) {.async: (raises: [CancelledError]).} =
while not stream.closed:
stream.clearInboundDataAvailable()
while true:
var inbound = stream.takeNextInbound().valueOr:
break
try:
await stream.pushData(move(inbound.payload))
except LPStreamError:
return
stream.advanceReceiveWindow(inbound.sequence)
await stream.waitForInboundData()waitForInboundData is a raw async wrapper around AsyncEvent.wait. The wrapper returns the event’s future directly instead of creating another async future and continuation:
proc waitForInboundData*(
stream: TransportStream
): Future[void] {.async: (raw: true, raises: [CancelledError]).} =
stream.dataAvailable.wait()Returning the original future also preserves the intended cancellation path. Cancelling runInboundDelivery cancels the active AsyncEvent.wait, whose cancellation callback removes that future from dataAvailable’s waiter list. Using wait().join() here would create an observer future whose cancellation deliberately does not cancel the underlying event wait.
takeNextInbound uses receiveBase as the sequence that must be delivered next. The procedure looks up that exact sequence in pendingInbound. If no payload is stored under receiveBase, a higher sequence may still be present in the table, but the procedure returns none because delivering that higher sequence would create a gap in the application byte stream:
proc takeNextInbound*(
stream: TransportStream
): Opt[tuple[sequence: SequenceNumber, payload: seq[byte]]] =
stream.pendingInbound.withValue(stream.receiveBase, payload):
let value = (sequence: stream.receiveBase, payload: move(payload[]))
stream.pendingInbound.del(stream.receiveBase)
return Opt.some(value)
Opt.none(tuple[sequence: SequenceNumber, payload: seq[byte]])When the table contains receiveBase, takeNextInbound removes that entry and returns both the sequence and the payload to runInboundDelivery. Ownership of the payload therefore moves from pendingInbound to the local inbound value held by the delivery task. runInboundDelivery passes that payload to BufferStream.pushData and calls advanceReceiveWindow only after pushData succeeds.
The acknowledgement bitmap is not consulted when selecting the next payload, but its bit for receiveBase remains set while pushData is suspended. During that interval, pendingInbound no longer contains the payload for receiveBase; the delivery task owns the payload instead. If another Data frame carrying the same sequence arrives, receiveData sees the set bit and classifies the frame as a duplicate rather than inserting another payload into pendingInbound. After pushData succeeds, advanceReceiveWindow shifts the bitmap and advances receiveBase.
The precise invariant is therefore not that every set bit has a matching pendingInbound entry. A set bit means that the corresponding sequence has been accepted and has not yet advanced below receiveBase. The accepted payload is normally stored in pendingInbound; for the one sequence currently being passed to BufferStream, the delivery task temporarily owns the payload. This bitmap state removes the need for a separate field that identifies the sequence currently being delivered.
If sequence 2 arrives before 1, pendingInbound contains sequence 2 but does not contain the current receiveBase, sequence 1. takeNextInbound therefore returns none. When sequence 1 arrives, the task pushes sequence 1, advances receiveBase and can then push sequence 2.
The window advances only after pushData succeeds:
proc advanceReceiveWindow*(stream: TransportStream, sequence: SequenceNumber) =
doAssert sequence == stream.receiveBase
doAssert stream.acknowledgementBitmap.bitmapContains(0)
stream.shiftBitmap()
inc stream.receiveBase
stream.fireShouldSendAckEvent()
if stream.acknowledgementBitmap.bitmapContains(0):
stream.dataAvailable.fire()Shifting makes the former bit one the new bit zero. Incrementing receiveBase extends the far edge of the receive window by one sequence.
9. Application Backpressure Reaches the Sender
libp2p’s BufferStream.pushData uses a capacity-one asynchronous queue and waits when that queue is occupied. If the mounted protocol stops reading, runInboundDelivery eventually blocks in pushData and does not advance receiveBase.
application stops reading
↓
BufferStream queue fills
↓
runInboundDelivery waits in pushData
↓
receiveBase stops advancing
↓
ACKs stop extending the remote receive limit
↓
sender reaches the limit and waitForOutboundCapacity blocksSuccessful pushData means that the payload entered BufferStream, where it is available for the application to read. The application may not have read those bytes yet. The receiver can hold payloads in three places: out-of-order chunks in pendingInbound, the one chunk currently being passed from the delivery task to BufferStream, and the capacity-one queue inside BufferStream. The fixed receive window limits the number of chunks retained in pendingInbound, while the delivery task and BufferStream each add only a fixed amount of additional buffering. A slow application therefore cannot cause the receiver to retain an unlimited number of chunks.
After pushData succeeds, the receiver advances receiveBase and reports the new base in an ACK. Advancing the base permits the sender to allocate another sequence number, even though the application may not yet have read the payload from BufferStream. If the transport instead permitted another chunk only after the application completed a read, the transport would have to observe or wrap the BufferStream read operations. The current implementation does not provide that stricter form of flow control.
10. The Receiver Produces an Absolute ACK
The receiver must inform the sender when the receive state changes. Accepting a new Data chunk changes the acknowledgement bitmap. Advancing receiveBase reports additional space in the receive window. Receiving duplicate Data also requires another ACK because the duplicate may mean that an earlier ACK was lost.
These state changes do not send an ACK directly. They call fireShouldSendAckEvent, which signals the stream’s shouldSendAck event:
proc fireShouldSendAckEvent(stream: TransportStream) =
stream.shouldSendAck.fire()configureStream starts one runAcknowledgements task for each established stream. shouldSendAck is a manual-reset Chronos AsyncEvent. runAcknowledgements waits for the event, verifies that the stream is still open, clears the event and calls acknowledgementSnapshot. The task then creates an Ack frame from that snapshot and submits the frame through sendStreamFrame:
proc runAcknowledgements(
self: MixTransport, session: TransportSession, stream: TransportStream
) {.async: (raises: [CancelledError]).} =
while not stream.closed:
await stream.waitForShouldSendAck()
if stream.closed:
return
stream.clearShouldSendAck()
let snapshot = stream.acknowledgementSnapshot()
let frame = MixTransportFrame(
version: MixTransportVersion,
sessionId: session.sessionId,
kind: FrameKind.Ack,
streamId: Opt.some(stream.streamId),
receiveBase: Opt.some(snapshot.receiveBase),
acknowledgementBitmap: Opt.some(snapshot.acknowledgementBitmap),
)
if (await self.sendStreamFrame(session, frame)).isErr:
returnwaitForShouldSendAck follows the same raw-wrapper pattern as waitForInboundData:
proc waitForShouldSendAck*(
stream: TransportStream
): Future[void] {.async: (raw: true, raises: [CancelledError]).} =
stream.shouldSendAck.wait()acknowledgementSnapshot is the operation called immediately before the frame is constructed. The operation copies the current bitmap and pairs the copy with the current receiveBase:
proc acknowledgementSnapshot*(stream: TransportStream): AckSnapshot =
var bitmap = newSeq[byte](AckBitmapBytes)
for index, value in stream.acknowledgementBitmap:
bitmap[index] = value
AckSnapshot(receiveBase: stream.receiveBase, acknowledgementBitmap: move(bitmap))The copy gives the asynchronous send operation a stable representation of the receiver’s state at that moment. If another Data frame changes stream.acknowledgementBitmap while sendStreamFrame is awaiting the Mix send operation, that change does not modify the bitmap already stored in the outgoing Ack frame.
Several calls to fireShouldSendAckEvent that occur before the ACK task runs are coalesced into the event’s single signalled state. The snapshot still includes every Data chunk accepted into the receive window and every receiveBase advancement completed before acknowledgementSnapshot is called. If receiveData or advanceReceiveWindow signals shouldSendAck while sendStreamFrame is awaiting completion, the event remains signalled and causes the next loop iteration to send a newer snapshot.
The ACK is absolute. receiveBase declares that every lower sequence entered the ordered BufferStream. Set bitmap bits identify additional chunks that the receiver has accepted but cannot yet deliver because an earlier sequence is missing. One ACK can therefore report the acceptance of several chunks together with every advancement already reflected in the current receiveBase.
ACKs are currently sent immediately. There is no delayed-ACK timer, and a send error ends the ACK task rather than scheduling a retry.
11. The Sender Applies the ACK
This section moves from the endpoint that generated the ACK to the endpoint that previously sent the acknowledged Data. That endpoint is called the sender in this section. When the session initiator sends forward Data, the recipient’s ACK returns through a SURB. When the session recipient sends reverse Data through a SURB, the initiator’s ACK arrives through the forward Mix path. Both arrival paths decode the same Ack frame and dispatch it to handleAcknowledgement.
handleAcknowledgement uses the frame’s sessionId and streamId to find the outbound state to which the ACK applies. An ACK for an unknown session, a session that has not been established, an unknown stream or a stream that has not been established is ignored. For an established session and stream, the procedure passes the advertised receiveBase and bitmap to applyAcknowledgement:
proc handleAcknowledgement(
self: MixTransport, frame: MixTransportFrame
) {.gcsafe, raises: [].} =
let session = self.sessions.get(frame.sessionId).valueOr:
return
if session.state != SessionState.Established:
return
let stream = session.getStream(frame.streamId.get()).valueOr:
return
if stream.state != StreamState.Established:
return
discard stream.applyAcknowledgement(
frame.receiveBase.get(), frame.acknowledgementBitmap.get()
)The sender retains every unacknowledged Data chunk in stream.pendingOutbound, indexed by its sequence number. The current implementation uses the number of retained entries to enforce MaxInflightChunks. Each entry retains both the payload needed for retransmission and its optional next retransmission deadline.
applyAcknowledgement first validates the received snapshot. The bitmap must have exactly AckBitmapBytes bytes. receiveBase must not be lower than stream.remoteReceiveBase, because accepting an older base would move the sender’s view of the receive window backwards. receiveBase must not be higher than stream.nextOutboundSequence, because the receiver cannot have advanced beyond every sequence the sender has allocated:
proc applyAcknowledgement*(
stream: TransportStream, receiveBase: SequenceNumber, bitmap: openArray[byte]
): bool =
if bitmap.len != AckBitmapBytes or receiveBase < stream.remoteReceiveBase or
receiveBase > stream.nextOutboundSequence:
return false
var acknowledged: seq[SequenceNumber]
for sequence in stream.pendingOutbound.keys:
if sequence < receiveBase:
acknowledged.add(sequence)
else:
let offset = sequence - receiveBase
if offset < SequenceNumber(ReceiveWindowChunks) and bitmap.bitmapContains(offset):
acknowledged.add(sequence)
for sequence in acknowledged:
stream.pendingOutbound.del(sequence)
let changed = acknowledged.len > 0 or stream.remoteReceiveBase != receiveBase
stream.remoteReceiveBase = receiveBase
if changed:
stream.sendStateChanged.fire()
stream.retransmissionStateChanged.fire()
changedEvery sequence below receiveBase is cumulatively acknowledged, so its retained payload can be removed. For a retained sequence at or above receiveBase, the procedure calculates the sequence’s offset inside the advertised receive window. A set bit at that offset confirms that the receiver has accepted that specific sequence even if an earlier sequence is still missing.
After removing the acknowledged chunks, applyAcknowledgement records the new remoteReceiveBase. The procedure fires sendStateChanged when at least one chunk was removed or the base advanced. Removing a chunk can bring pendingOutbound.len below MaxInflightChunks. Advancing remoteReceiveBase moves the upper boundary of the remote receive window. Either change can make canReserveOutbound true and wake a writer waiting in waitForOutboundCapacity. The procedure also fires retransmissionStateChanged so that the retransmission task recalculates its earliest deadline after acknowledged entries have disappeared.
applyAcknowledgement returns false for an invalid snapshot and for a valid snapshot that does not change the sender’s state. handleAcknowledgement currently ignores that return value because both cases require no further action.
12. Unacknowledged Data Is Retransmitted
newMixTransport enables Data retransmission by default and uses a fixed 30-second timeout. A caller can disable this behavior explicitly without changing ACK processing or flow control:
proc newMixTransport*(
mix: MixProtocol,
connectTimeout = DefaultConnectTimeout,
streamOpenTimeout = DefaultStreamOpenTimeout,
dataRetransmissionTimeout = DefaultDataRetransmissionTimeout,
surbSupplyRetransmissionTimeout = DefaultSurbSupplyRetransmissionTimeout,
reverseActivityTimeout = DefaultReverseActivityTimeout,
surbStatusProbeRetryInterval = DefaultSurbStatusProbeRetryInterval,
maxSurbStatusProbeAttempts = DefaultMaxSurbStatusProbeAttempts,
enableDataRetransmissions = true,
recipientSurbCapacity = DefaultRecipientSurbCapacity,
): MixTransportWhen enableDataRetransmissions is false, configureStream does not start runRetransmissions, and writeStream does not assign retransmission deadlines. The sender still retains each outbound chunk until it is acknowledged because pendingOutbound also enforces the in-flight bound and prevents the sender from outrunning the receiver’s window.
When retransmission is enabled, runRetransmissions belongs to the same TransportStream as the ACK and ordered-delivery tasks. The task first clears its notification event and asks takeDueOutboundRetransmission for the due chunk with the earliest deadline:
proc runRetransmissions(
self: MixTransport, session: TransportSession, stream: TransportStream
) {.async: (raises: [CancelledError]).} =
while not stream.closed:
stream.clearRetransmissionStateChanged()
var retransmission = stream.takeDueOutboundRetransmission().valueOr:
let deadline = stream.earliestRetransmissionDeadline().valueOr:
await stream.waitForRetransmissionStateChange()
continue
let waitTime = deadline - Moment.now()
if waitTime > ZeroDuration:
discard await stream.waitForRetransmissionStateChange().withTimeout(waitTime)
continueIf no retained chunk has a deadline, the task waits until initial Data submission schedules one, an ACK changes the table, or stream closure requests cancellation. If the earliest deadline is in the future, the task waits until either that deadline expires or retransmissionStateChanged reports an earlier state change. In both cases the loop inspects pendingOutbound again instead of assuming that the state which caused the wake-up is still current.
takeDueOutboundRetransmission copies the selected payload for the asynchronous send and clears that chunk’s deadline. Clearing the deadline marks that a retransmission is currently being submitted and prevents the loop from selecting the same chunk again concurrently. The task constructs another Data frame with the original stream ID, sequence number and payload, then submits the frame through the same session-role-dependent path used by the initial send:
let frame = MixTransportFrame(
version: MixTransportVersion,
sessionId: session.sessionId,
kind: FrameKind.Data,
streamId: Opt.some(stream.streamId),
sequence: Opt.some(retransmission.sequence),
payload: Opt.some(move(retransmission.payload)),
)
let sent = await self.sendStreamFrame(session, frame)
stream.scheduleOutboundRetransmission(
retransmission.sequence, self.dataRetransmissionTimeout
)The next fixed deadline is scheduled after the submission finishes, whether that submission succeeded or returned an error. A transient local send failure therefore does not silently abandon an unacknowledged chunk. There is currently no retry-count limit: retransmission continues until an ACK removes the chunk or the stream closes.
An ACK can arrive while sendStreamFrame is suspended. In that case applyAcknowledgement removes the chunk from pendingOutbound. The later call to scheduleOutboundRetransmission changes an entry only when that entry still exists, so completion of the overlapping retransmission cannot restore an acknowledged chunk. The receiver may observe the already-submitted duplicate; receiveData suppresses the duplicate and requests another absolute ACK snapshot.
13. Stream Shutdown Cancels and Waits for Its Tasks
An established stream owns an ordered Data-delivery task and an ACK task. An accepted inbound application stream also owns the protocol-handler invocation operating on that stream. runInboundDelivery may wait on dataAvailable, runAcknowledgements may wait on shouldSendAck, and an application writer may wait on sendStateChanged because the in-flight or remote-window limit has been reached.
Firing the events is sufficient when a task is idle at one of those waits, but it does not stop a task that has moved into another asynchronous operation. For example, runAcknowledgements may be inside sendStreamFrame, waiting for a session SURB refill. The ACK task is no longer waiting on shouldSendAck, so firing that event cannot terminate it. TransportStream therefore owns explicit references to its internal streamTasks and optional handlerTask.
TransportStream.closeImpl delegates to closeTransportStream. That procedure fires the ordinary state events, requests cancellation of the internal stream tasks and closes BufferStream before invoking the callback that attempts a remote teardown notification:
proc closeTransportStream(
stream: TransportStream, reset: bool
): Future[void] {.async: (raises: []).} =
stream.dataAvailable.fire()
stream.shouldSendAck.fire()
stream.sendStateChanged.fire()
stream.retransmissionStateChanged.fire()
stream.resolved.fire()
stream.streamTasks.cancelSoon()
await procCall BufferStream(stream).closeImpl()
if not stream.suppressRemoteTeardown and not stream.teardownHandler.isNil:
await stream.teardownHandler(reset, stream.finalOutboundSequence)
if not stream.handlerTask.isNil:
stream.handlerTask.cancelSoon()The inherited LPStream.close operation marks the stream closed before dispatching to closeImpl. The fired events allow ordinary waiters to observe that state, while explicit cancellation reaches a task suspended deeper inside Mix or SURB processing. The locally installed teardown callback waits for streamTasks after attempting the notification. The protocol handler is cancelled after that callback; if the handler initiated the close while returning naturally, its cleanup clears handlerTask before the call and therefore does not cancel or wait for itself.
External teardown uses TransportStream.shutdown. The procedure retains a local reference to handlerTask, closes the stream and then waits for the internal tasks and handler to finish. The local reference remains valid if handler cleanup clears the field while cancellation is being processed:
proc shutdown*(stream: TransportStream): Future[void] {.async: (raises: []).} =
let handlerTask = stream.handlerTask
await stream.close()
await stream.cancelAndWaitForStreamTasks()
if not handlerTask.isNil:
await noCancel handlerTask.cancelAndWait()
stream.handlerTask = nilWhen runProtocolHandler finishes naturally, it clears handlerTask before closing the stream and waits only for the internal streamTasks. The handler therefore never waits for its own future.
Stopping the complete transport follows the ownership hierarchy. takeSessions synchronously detaches every session from the store. Before unregistering the Mix handlers, the transport makes one best-effort ResetSession submission for each detached session. Each session then synchronously detaches its streams through takeStreams; session shutdown starts every stream shutdown and waits for all of them. No table iterator survives across an await, and a handler completing during cancellation cannot mutate the container currently being iterated. After all detached sessions finish, the transport clears reply credentials:
proc stop*(self: MixTransport): Future[void] {.async: (raises: [CancelledError]).} =
if not self.started:
return
let sessions = self.sessions.takeSessions()
for session in sessions:
let frame = MixTransportFrame(
version: MixTransportVersion,
sessionId: session.sessionId,
kind: FrameKind.ResetSession,
)
discard await self.sendTeardownFrame(session, frame)
self.mix.unregisterRawSurbReplyHandler()
self.mix.unregisterMixDeliveryHandler(MixTransportCodec)
var shutdownTasks = newSeqOfCap[Future[void].Raising([])](sessions.len)
for session in sessions:
shutdownTasks.add(session.shutdown())
await noCancel shutdownTasks.allFutures()
self.replyCredentials.clear()
self.started = falseStream and session teardown use CloseStream, ResetStream, Disconnect and ResetSession. Graceful stream closure carries a final Data sequence so an out-of-order close cannot discard preceding Data, while reset remains immediate. Mix Transport Implementation Walk Through - Remote Teardown explains the complete send, receive, task-cancellation and readOnce reset paths. This note retains the task-ownership context because those tasks implement bounded Data flow, but the teardown walk-through is the source for lifecycle behavior.
Reliability Boundaries
Data retries use a fixed timeout with no retry-count limit. A failed ACK submission ends that stream’s ACK task; outstanding duplicate Data can prompt another ACK only while that task is active. There is no Data-window persist probe when all outstanding chunks have been acknowledged but window-advancing ACKs are lost. Teardown notifications are best effort. These limits are part of the behavior described in Mix Transport Design Specification, not guarantees of recovery from every loss pattern.