Struct Conversations 

pub struct Conversations {
    held: Mutex<HashMap<SessionSlot, Arc<Held>>>,
    seeded: Notify,
    consumed: Mutex<HashMap<SessionSlot, Sender<u64>>>,
    fills: Mutex<HashMap<SessionSlot, Arc<Mutex<()>>>>,
    fold_held: AtomicBool,
    seed_hold: Mutex<Option<Arc<SeedHold>>>,
}

The conversations the transport is holding, one per seat it has heard of.

Every live seat, not only the watched ones, and no seat is let go. A conversation has no read behind it any more, so releasing one is not a bound - nothing could rebuild it, and the next ask for that seat would have no answer. What it holds is therefore proportional to what is RUNNING plus the sleeping seats someone has opened: a sleeping seat’s copy is seeded from its transcript and kept, which is what makes the read land before any spawn. What one seat costs is the window the ceilings keep rather than its transcript, and the window is the bound.

Fields

held: Mutex<HashMap<SessionSlot, Arc<Held>>>

seeded: Notify

Fires when a seat’s copy is first held or reseeded, so a read waiting for it to become current knows it moved.

consumed: Mutex<HashMap<SessionSlot, Sender<u64>>>

The newest number the fold loop has consumed for each seat, in a watch so an encode can wait on it.

What an answer’s copy is measured against. A subscribe’s answer carries the number its seat’s stream will continue from, and every frame numbered below that must be inside the copy it carries - because the client adopts seq - 1 and treats a later delivery of anything at or below it as already drawn. The copy is filled by the fold loop, which consumes the fanout in a task of its own, so “the frame was emitted before the counter was read” does not yet mean “the copy has it”: this is where that is known.

fills: Mutex<HashMap<SessionSlot, Arc<Mutex<()>>>>

The lock a seat’s fill runs under.

What makes “applied once” a fact rather than a hope. The fold loop and an encode’s catch-up can both apply the same frames, and the watermark between them is read and written by both; the lock spans each side’s check, applies and mark, so a frame crosses exactly once whichever path got there first.

fold_held: AtomicBool

A test’s hold on the fold loop, so the window above can be driven rather than raced: while it is set, the loop does not consume.

seed_hold: Mutex<Option<Arc<SeedHold>>>

A test’s hold on the next transcript seed, so a stream copy that lands inside the seed’s window is driven rather than raced.

Implementations

impl Conversations

pub fn new() -> Self

pub fn consumed_for(&self, slot: &SessionSlot) -> u64

The newest number the fold loop has consumed for slot.

pub fn consumed(&self, slot: &SessionSlot, seq: u64)

Say the fold loop has consumed seq for slot.

send_replace, not send. The watch is created with its receiver dropped, so consumed_for can read the watermark with nobody waiting - and send is a no-op when a channel has no receivers, which would leave the watermark at 0 forever and make every read of it wrong.

pub async fn await_consumed( &self, slot: &SessionSlot, want: u64, within: Duration, ) -> bool

Wait, up to within, for the fold to have consumed want for slot, answering whether it did.

Bounded, because the caller has a fallback and a stuck fold must not become a stall: the caller fills the gap from the seat’s own window and refuses only if that cannot reach either.

pub fn fill_lock(&self, slot: &SessionSlot) -> Arc<Mutex<()>>

The lock one seat’s fill runs under: taken by the fold loop and by an encode’s catch-up, so their watermark checks and applies cannot interleave.

pub fn hold_the_fold(&self, held: bool)

Hold the fold loop, or let it go. Test-only: it is the seam that makes the copy’s lag deterministic instead of a race.

pub fn fold_is_held(&self) -> bool

Whether a test is holding the fold loop.

pub fn hold_the_seed(&self, hold: Arc<SeedHold>)

Hold the next transcript seed before its read, so a test can land the seat’s own copy inside that window. Test-only.

pub async fn pass_the_seed_hold(&self)

Park at a test’s seed hold, when one is set: say it has arrived, then wait to be let go.

pub fn get(&self, slot: &SessionSlot) -> Option<Arc<Held>>

The seat’s conversation if one is held.

pub fn live_of(&self, slot: &SessionSlot) -> bool

Whether the seat’s turn is in flight right now, from its own frames.

false for a seat this transport has not heard of: nothing holds a turn there, and a reader folding its transcript is reading a settled conversation.

pub fn insert( &self, slot: &SessionSlot, conversation: Conversation, ) -> Arc<Held>

Hold conversation for slot, unless the stream got there first.

pub fn replace( &self, slot: &SessionSlot, conversation: Conversation, ) -> Arc<Held>

Replace a seat’s copy outright.

What a copy built from a ring stretch needs: insert keeps what is there, and the stretch replaces it so the frames after it apply to the copy the read is served from.

pub fn seeded_notice(&self) -> Notified<'_>

A future that resolves when some seat’s copy is first held or reseeded.

Enable it before looking the seat up, or a seat seeded between the lookup and the wait leaves the caller waiting for something that has already happened - and bound the wait, because a seat whose session is gone never answers at all.

A reseed counts: a seal the header has already moved on from lands on the copy the seat holds, in place, so a reader waiting for that copy to become current is woken by it rather than by the seat’s first hold.

pub fn apply(&self, update: &SessionUpdate, seq: Option<u64>)

Fold one update into the seat it names.

seq is the number the frame was minted under; the copy records it as the newest it reflects.

fn reseed( &self, slot: &SessionSlot, history: &[Message], compaction_count: u32, carried: usize, occupant: Option<&SessionId>, )

Put history on slot, materialising the seat when this is the first word about it.

Held means reseed and NOT insert: insert keeps what is there, so a connect on a seat already carrying a conversation would leave that conversation in place and the new history discarded.

The watermark goes to zero: a seal opens a new epoch, whose frame count starts at 1.

pub fn len(&self) -> usize

How many seats are held.

pub fn is_empty(&self) -> bool

Trait Implementations

impl Default for Conversations

fn default() -> Self

Returns the “default value” for a type.