Struct Repos 

pub struct Repos {
    inner: Arc<Inner>,
}

The producer: the sequence and the replay buffer consumers read from.

One per server. Hand a clone to Provisioner::with_repo_sink and the same one to app_with_repos; every clone shares Inner.

Fields

inner: Arc<Inner>

Implementations

impl Repos

pub fn new(sequence: Arc<Sequence>, bounds: ReplayBounds) -> Self

Builds a producer numbering frames from sequence and holding up to bounds of them for replay.

pub fn booting( sequence: Arc<Sequence>, bounds: ReplayBounds, log: Option<Arc<Wal>>, ) -> Result<Self>

Builds a producer for a boot: it holds every frame numbered until Self::listening, and bounds more after it.

With log, the write-ahead log commits are written to, a commit is published only once log has synced past it. See the module docs.

Errors

The thread that publishes held commits would not start.

fn build( sequence: Arc<Sequence>, bounds: ReplayBounds, booting: bool, held: Option<Arc<Held>>, ) -> Self

pub fn listening(&self)

Ends the boot Self::booting began: the buffer keeps every frame the boot numbered, and holds Self::bounds more on top of them, with the same slack for frames a stream has not sent.

Called once the listener is bound. Nothing before that could have read a frame, and a relay that reconnects after a boot is owed every one the boot numbered. A producer that is not booting is left as it is.

pub fn newest(&self) -> u64

The newest sequence number assigned, or the restart floor if none has been.

pub fn subscribers(&self) -> usize

How many streams are open right now.

The per-connection state this producer holds is one receiver of published and one of closing, so this figure is the whole of it: a consumer that hung up and left its receiver behind would show up here and nowhere else. Observability rather than a bound — nothing refuses a subscriber for being the n-th — and the number a test watches to see a departed consumer actually freed.

pub fn bounds(&self) -> ReplayBounds

How much the replay buffer holds at most, once any boot is over.

fn ring(&self) -> MutexGuard<'_, Ring>

fn start_for(&self, resume: Resume, newest: u64) -> Start

Where a consumer resuming from resume starts, decided under one acquisition of the ring lock, which also registers its stream so the frames after its start are kept for it.

One acquisition for the whole decision, which is what makes the outdated answer true of the position beside it. Measuring the buffer and then reading it takes two acquisitions, and a publish landing between them evicts the very frame the cursor was just tested against: the consumer is then handed a later start with no #info OutdatedCursor in front of it, and believes it applied everything. A frame evicted after the decision is the stream’s to notice, in ReposStream::next.

pub fn publish_commit(&self, event: &RepoCommit) -> Option<Frame>

Numbers, encodes, buffers and publishes one commit.

Returns the frame so a caller — a test, mostly — can see what went out, and None when the stream could not number it: a frame numbered over a reservation that did not reach the log wears a number the next run will hand out again. See didbot_pds::Sequence::next.

pub fn publish_identity(&self, event: &RepoIdentity) -> Option<Frame>

Numbers, encodes, buffers and publishes one identity event.

pub fn publish_account(&self, event: &RepoAccount) -> Option<Frame>

Numbers, encodes, buffers and publishes one account event.

pub fn publish_sync(&self, event: &RepoSync) -> Option<Frame>

Numbers, encodes, buffers and publishes one repository reset.

fn publish( &self, kind: FrameKind, body: impl FnOnce(u64) -> Value, ) -> Option<Frame>

Numbers a frame, puts it in the buffer and hands it to the live channel, under one lock.

None when the number could not be reserved durably. The event is not buffered and not sent: a frame numbered over a reservation that did not land wears a number the next run will hand out again, and the consumer holding it as a cursor is told nothing. See didbot_pds::Sequence::next.

fn publish_locked( &self, ring: &mut Ring, kind: FrameKind, body: impl FnOnce(u64) -> Value, ) -> Option<Frame>

Self::publish, under a ring lock the caller holds.

pub fn release(&self)

Syncs the log and publishes every commit held back for it, oldest first.

A clean shutdown calls this once serving has stopped, so the commits held back go out before Self::close ends the streams.

fn release_synced(&self)

Publishes every held commit the log has synced past, oldest first.

fn release_locked(&self, ring: &mut Ring)

fn sync_held_log(&self)

Syncs the log the held commits are in, when there is one.

fn publish_after_held( &self, kind: FrameKind, body: impl FnOnce(u64) -> Value, ) -> Option<Frame>

Publishes a frame that is not a commit, after every commit handed over before it. See the module docs.

pub fn close(&self)

Ends every consumer’s stream once it has sent what is already published.

A subscribeRepos connection outlives the server that upgraded it until this process exits. A clean shutdown calls this after Self::release, so each connection sends what it holds, closes and gives its slot back. See Self::detached.

pub async fn detached(&self, timeout: Duration) -> usize

Waits, for at most timeout, until no consumer is attached. Answers how many still are.

pub fn subscribe( &self, cursor: Option<i64>, exclude: Exclusions, ) -> Result<ReposStream, Vec<u8>>

Opens one connection’s stream, resuming from cursor if it can.

Err is the frame to send before closing, which is how the lexicon’s declared errors go out: an error is a frame on the socket, not an HTTP status, because by the time a cursor has been read the upgrade has already happened.

Trait Implementations

impl Clone for Repos

fn clone(&self) -> Repos

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Repos

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for Repos

fn default() -> Self

Returns the “default value” for a type.

impl RepoEventSink for Repos

A commit is held when the producer is over a log; every other frame publishes after the commits held before it. See the module docs.

fn commit(&self, event: &RepoCommit)

Delivers one commit, with the blocks it created.

fn identity(&self, event: &RepoIdentity)

Says an account’s DID document is worth re-resolving.

fn account(&self, event: &RepoAccount) -> bool

Says whether an account’s repository can still be fetched here.

fn sync(&self, event: &RepoSync) -> bool

Says a repository was reset to a new commit. Answers whether the frame was numbered.