Module subscribe 

com.atproto.sync.subscribeRepos: this server’s repositories, as atproto wants to hear about them.

A WebSocket of framed DAG-CBOR carrying signed commits and the blocks that prove them, which is what a relay crawls.

The frame

One WebSocket binary message is two concatenated DAG-CBOR values: a header and a body. The header is {"op": 1, "t": "#commit"} for a message, and {"op": -1} for an error, whose body is {"error": .., "message": ..}. Nothing separates the two values; a reader decodes the header and continues from where that decode stopped.

Sequence numbers and cursors

Every message frame carries a sequence number, and a consumer hands back the last one it applied as ?cursor=. The cursor is a bare integer with nowhere to name the run of the server that minted it, so the number itself survives a restart — see didbot_pds::sequence for how, and for why a restart skips numbers rather than repeating them.

The replay buffer holds at most DEFAULT_REPLAY_FRAMES frames and DEFAULT_REPLAY_BYTES of them, evicting the oldest first, unless a deployment sets its own ReplayBounds. It lives in memory, so a restart empties it even though the numbering survives. A cursor the buffer cannot reach is answered with #info OutdatedCursor and then everything still held: the consumer has a hole it must fill by reading the repositories, and being told so is the point. A cursor naming a number this server has not assigned yet is a different thing — it cannot be filled in from anywhere — so it is the FutureCursor error the lexicon declares, and the connection closes.

A boot numbers frames before anything can connect: the positions that changed while the process was down, and a restore’s resets. A producer built with Repos::booting keeps every one of them until Repos::listening, and its bounds’ worth more on top, so a relay that connects after the boot is handed all of them.

One copy of each frame

The replay buffer is the only place a frame is held. A consumer keeps its position in the buffer and reads each frame from there when it is ready to send it, the replay and the live stream alike.

A frame an open stream has not sent yet is kept past the bounds, up to twice them, so a consumer resuming from the oldest frame held has as much room to fall behind as one following live. Past twice the bounds the oldest frames go anyway, and a consumer that had not sent one is ended with ConsumerTooSlow: its next frame is gone. So the firehose’s memory is at most twice the bounds, plus the one frame each connection is writing to its socket.

Only what is durable

A record write is in the kernel when it returns and on the medium within DEFAULT_SYNC_INTERVAL. A #commit published in between announces a commit a power cut can take back, and a relay that applied it holds a head this server no longer has. So a producer over a write-ahead log holds each commit, beside the position the log had reached when it was handed over, and publishes it once the log has synced through that position. A commit’s frame is therefore up to one sync interval later than its write.

Every other frame syncs the log and publishes the commits held before it first, so the stream keeps the order the writes made: an erasure’s #account after the repository’s last #commit, and a restore’s #sync after the commit it carries is on the medium.

Structs

Exclusions

Which message types one connection wants.

Frame

One numbered frame, already encoded.

Held 🔒

Commits handed over before the log they are written to has synced past them. See the module docs.

Inner 🔒

ReplayBounds

How much the replay buffer holds. It evicts its oldest frame when a new one would take it past either bound, unless an open stream has not sent that frame yet. Those it keeps up to twice the bounds. See the module docs.

Repos

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

ReposStream

One consumer’s stream: the replay, then whatever arrives live, both read from the replay buffer.

Ring 🔒

The replay buffer’s frames, and how much it holds.

Start 🔒

Where one consumer starts reading the replay buffer, decided in Repos::start_for.

Enums

FrameKind

Which of the lexicon’s message types a frame is.

Read 🔒

What one look at the buffer found for a stream.

Resume 🔒

What a connection asked for.

Constants

DEFAULT_REPLAY_BYTES

How many encoded bytes the replay buffer holds, if a deployment does not choose.

DEFAULT_REPLAY_FRAMES

How many frames the replay buffer holds, if a deployment does not choose.

FUTURE_CURSOR

The lexicon’s error for a cursor ahead of the stream.

OP_ERROR 🔒

The op an error frame’s header carries.

OP_MESSAGE 🔒

The op a message frame’s header carries.

OUTDATED_CURSOR

The one name this server ever sends on an #info frame.

Functions

account_body 🔒

The #account body.

clamp 🔒

The lexicon’s seq, which is a signed integer where this counter is not.

commit_body 🔒

The #commit body, as the lexicon defines it.

encode_error

Encodes one error frame, which closes the connection after it.

encode_message

Encodes one message frame: the header, then the body.

identity_body 🔒

The #identity body.

info_body 🔒

The #info body.

link 🔒

A CID-link, or null for the fields the lexicon marks nullable.

map 🔒

A DAG-CBOR map from pairs, so the bodies below read as their lexicons.

now_rfc3339 🔒

The current time, RFC 3339, seconds precision.

publish_as_synced 🔒

Publishes held commits as the log syncs past them, on a thread of its own.

sync_body 🔒

The #sync body.