Archive replay

class atproto_jetstream.archive.ByteMeter

Bases: object

Counts downloaded bytes. Usage is metered in bytes, so callers need to see the spend.

add(count: int) → None
property total: int

int: Bytes downloaded so far.

class atproto_jetstream.archive.PlanFilters(kinds: List[str] = <factory>, dids: List[str] = <factory>, collections: List[str] = <factory>)

Bases: object

Filters shared by the planner and the row matcher.

as_plan_input(after_seq: int, before_seq: int | None) → Dict[str, Any]
kinds: List[str]
dids: List[str]
collections: List[str]
class atproto_jetstream.archive.RowMatcher(kinds: Sequence[str] | None = None, dids: Sequence[str] | None = None, collections: Sequence[str] | None = None, after_seq: int = 0, before_seq: int | None = None)

Bases: object

Exact row filter over decoded segment rows.

The plan only narrows which blocks are downloaded: it works on whole blocks and coarse collection ids, so a downloaded block carries rows the caller never asked for. This is the authority on what is delivered.

Parameters:

  • kinds – Event kinds to keep. Empty means all.

  • dids – Repo DIDs to keep. Empty means all.

  • collections – Collection NSIDs or <prefix>.* patterns. Empty means all.

  • after_seq – Exclusive lower bound.

  • before_seq – Inclusive upper bound.

advance_to(seq: int) → None

Raise the seq floor after a resume, so a straddling block does not re-emit rows.

filter(events: Iterable[SegmentEvent]) → Iterator[SegmentEvent]

Yield only the rows that match.

matches(event: SegmentEvent) → bool

Whether the row should be delivered.

class atproto_jetstream.archive.SegmentEvent(seq: int, witnessed_at: int, indexed_at: int, kind: int, did: str, collection: str, rkey: str, rev: str, payload: bytes | None)

Bases: object

One row of a segment block.

property is_commit: bool

bool: Whether the row is a record mutation.

property operation: str | None

str: create, update or delete, or None for non-commit kinds.

property time_us: int

int: Display timestamp in unix microseconds.

seq: int

Jetstream sequence number.

witnessed_at: int

When Jetstream first saw the event, unix microseconds.

indexed_at: int

Display timestamp, unix microseconds. 0 means fall back to witnessed_at.

kind: int

Discriminator; see the KIND_* constants.

did: str

Repo DID.

collection: str

Collection NSID. Empty for non-commit kinds.

rkey: str

Record key. Empty for non-commit kinds.

rev: str

Repo rev. Empty for non-commit kinds.

payload: bytes | None

Raw record CBOR. Absent for deletes.

class atproto_jetstream.archive.SegmentHeader(version: int, block_count: int, event_count: int, footer_offset: int, block_index_offset: int)

Bases: object

Fixed header of a sealed .jss segment.

version: int

Format version.

block_count: int

Number of blocks.

event_count: int

Number of events.

footer_offset: int

Byte offset of the footer.

block_index_offset: int

Byte offset of the block index.

class atproto_jetstream.archive.SnapshotPlan(filters: PlanFilters, after_seq: int = 0, before_seq: int | None = None)

Bases: object

Pages planSnapshot and yields work units in seq order.

The sealed tip is pinned from the first page: segments sealed while the sweep runs carry seqs above the pin and are deliberately left to the live tail’s cold replay at cutover.

Parameters:

  • filters – Filters to plan against.

  • after_seq – Exclusive lower bound.

  • before_seq – Inclusive upper bound.

consume_page(page: models.NetworkBskyJetstreamPlanSnapshot.Response) → List[WorkUnit]

Absorb a plan page and return its work units.

property is_complete: bool

bool: Whether planning has reached the pinned sealed tip.

next_page_input() → Dict[str, Any]

Build the planSnapshot input for the next page.

property pages: int

int: Number of plan pages fetched.

property planned_through_seq: int

int: Highest sealed seq accounted for so far.

property sealed_tip_seq: int | None

int: Pinned sealed tip, or None before the first page.

class atproto_jetstream.archive.WorkUnit(segment: str, checksum: str, min_seq: int, max_seq: int, blocks: List[int] | None = None)

Bases: object

One downloadable piece of the archive, in plan order.

blocks: List[int] | None = None

Block indexes to fetch, or None for the whole segment.

property is_whole_segment: bool

bool: Whether the whole segment file must be downloaded.

segment: str

Segment filename.

checksum: str

xxh3 metadata checksum. Doubles as the ETag for a resumed download.

min_seq: int

Lowest seq the unit may contain.

max_seq: int

Highest seq the unit may contain.

atproto_jetstream.archive.compute_record_cid(payload: bytes) → str

Derive a record’s CID from its DAG-CBOR bytes.

Parameters:

payload – Canonical DAG-CBOR of the record.

Returns:

CIDv1 string.

Return type:

str

atproto_jetstream.archive.decode_block(frame: bytes) → List[SegmentEvent]

Decode a stored block frame into rows.

Parameters:

frame – Raw block frame, exactly as getBlock returns it.

Returns:

Rows in sequence order.

Return type:

list of SegmentEvent

Raises:

atproto.exceptions.JetstreamDecodingError – Malformed block.

atproto_jetstream.archive.quota_retry_after(exception: BaseException) → float | None

Seconds to wait if the request was rejected for exceeding the byte quota.

Parameters:

exception – Exception raised by an archive request.

Returns:

Delay to honour, or None if this was not a quota rejection.

Return type:

float

atproto_jetstream.archive.read_sealed_header(data: bytes) → SegmentHeader

Parse the fixed 256-byte header of a sealed segment.

Parameters:

data – At least the first 256 bytes of the segment.

Returns:

Parsed header.

Return type:

SegmentHeader

Raises:

atproto.exceptions.JetstreamDecodingError – Not a sealed segment.

atproto_jetstream.archive.to_message(event: SegmentEvent, *, with_cid: bool = True) → Commit | Identity | Account | Sync

Convert a decoded segment row into the model the live tail delivers.

A caller cannot tell whether an event arrived over the websocket or out of a segment.

Parameters:

  • event – Decoded segment row.

  • with_cid – Derive the record CID. Skipping it saves a hash per record.

Returns:

Corresponding message model.

Return type:

ArchivedEventsMessage

Raises:

atproto.exceptions.JetstreamDecodingError – Unknown row kind or bad payload.

atproto_jetstream.archive.to_messages(events: Iterable[SegmentEvent], *, with_cid: bool = True) → Iterator[Commit | Identity | Account | Sync]

Convert rows, skipping any that cannot be decoded.

type atproto_jetstream.archive.ArchivedEventsMessage

Any message an archive segment can yield.