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¶

Bytes downloaded so far.

Type:: int

class atproto_jetstream.archive.PlanFilters(kinds: ~typing.List[str] = <factory>, dids: ~typing.List[str] = <factory>, collections: ~typing.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]¶

collections: List[str]¶

dids: List[str]¶

kinds: 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.

collection: str¶: Collection NSID. Empty for non-commit kinds.

did: str¶: Repo DID.

indexed_at: int¶: Display timestamp, unix microseconds. 0 means fall back to witnessed_at.

property is_commit: bool¶

Whether the row is a record mutation.

Type:: bool

kind: int¶: Discriminator; see the KIND_* constants.

property operation: str | None¶

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

Type:: str

payload: bytes | None¶: Raw record CBOR. Absent for deletes.

rev: str¶: Repo rev. Empty for non-commit kinds.

rkey: str¶: Record key. Empty for non-commit kinds.

seq: int¶: Jetstream sequence number.

property time_us: int¶

Display timestamp in unix microseconds.

Type:: int

witnessed_at: int¶: When Jetstream first saw the event, unix microseconds.

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.

block_count: int¶: Number of blocks.

block_index_offset: int¶: Byte offset of the block index.

event_count: int¶: Number of events.

footer_offset: int¶: Byte offset of the footer.

version: int¶: Format version.

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¶

Whether planning has reached the pinned sealed tip.

Type:: bool

next_page_input() → Dict[str, Any]¶: Build the planSnapshot input for the next page.

property pages: int¶

Number of plan pages fetched.

Type:: int

property planned_through_seq: int¶

Highest sealed seq accounted for so far.

Type:: int

property sealed_tip_seq: int | None¶

Pinned sealed tip, or None before the first page.

Type:: int

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.

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

property is_whole_segment: bool¶

Whether the whole segment file must be downloaded.

Type:: bool

max_seq: int¶: Highest seq the unit may contain.

min_seq: int¶: Lowest seq the unit may contain.

segment: str¶: Segment filename.

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.