grain.social/grain

Switch repository

Loading…

What changes

grain::jetstream speaks Jetstream v2 (network.bsky.jetstream.subscribeEvents, the xrpc.v1.json framing), as knotless's jetstream.rs does: the endpoint, a parse of each frame into commit, identity, account and info events, and follow, which reconnects with backoff (500 ms doubling to 60 s), pings a silent connection and gives it up, and starts each connection from the last sequence number heard.

A cursor Jetstream no longer keeps is given up, not offered forever. Every third refused attempt in a row asks for the live tip instead; if that connects, the cursor was the trouble, and it is dropped from the index with a warning that what came between was missed; if it is refused too, the instance is down and the cursor is kept. hatk's CURSOR_PROBE_EVERY, which ended the incident where a cursor past retention wedged ingest.

grain-serve's jetstream applies what it hears to the index, as hatk's indexer did:

• it asks for the collections the index keeps, by name rather than social.grain.*;
• a write to a signal collection (backfill's SIGNALS: social.grain.actor.profile, fyi.opensocial.declaration) tracks its author, and an author newly tracked is queued for backfill (Index::resync); anything else is kept only of tracked accounts, so the network's Bluesky posts pass by;
• creates and updates are put by URI and CID at the commit's rev (put_rev, delete_rev), so neither a stream write nor a backfill's read of the repo undoes a later one, and a record redelivered is applied once; a commit naming no rev is written as it comes; a record that does not fit its lexicon is logged and skipped, and so is an event the index fails to apply, which is not heard again once the cursor is saved;
• identity events set a tracked account's handle; account events keep what its host says (deactivated, suspended, takendown), clear it when the account is active again (and queue it for backfill with Index::resync), and on deleted purge its records. An admin takedown is in its own table, so none of this touches it.

The cursor is saved every 5 s and as a connection ends, under jetstream:<url> with the last sequence number seen, as hatk's _cursor row holds it, so the import copies jetstream:wss://jetstream.us-west.bsky.network over unchanged. Jetstream replays from the cursor inclusively; the replayed event is unchanged by its CID.

Settings. GRAIN_INDEX names the index's file; GRAIN_JETSTREAM, beside GRAIN_BACKFILL in Settings, names a Jetstream to follow into it (1 for wss://jetstream.us-west.bsky.network). Unset, as in tests and dev, nothing is followed; Settings::follow, like Settings::backfill, refuses to start without GRAIN_INDEX.

Why

Step two of the data layer: the live tail that keeps the index current, so the feeds that follow have something to read.

How it was checked

crates/grain-serve/tests/jetstream.rs runs a stand-in Jetstream on a local socket: create, update and delete; untracked accounts and unkept collections passing by; a malformed record, bad frames, an error frame and an info frame skipped while the stream goes on; identity and account events (deactivated, ignored statuses, back again, deleted and purged); the cursor resumed on each reconnect and after a restart; a stream write older than the revision a backfill read left alone, and one naming no revision applied; accounts newly tracked or back from deactivation queued, and one that never left not; a cursor past retention given up after three refusals, and not offered again when the live tip ends before it says anything; an instance that is down keeping its cursor; a silent connection given up; an index that fails mid-stream. Unit tests cover the frame parse and the endpoint; tests/main.rs the settings in the binary. Nothing was built or run locally.

Shortcomings

• An identity event without a handle is not re-resolved (hatk asked PLC).
• Queuing an account does not wake a running backfill; it finds the account at its next poll (5 s).
• No reference PDS here: Jetstream is a stand-in. Through the harness, a test could write records to the reference PDS and read its firehose, once there is a Jetstream (or a firehose reader) in front of it.
• Events are applied one at a time on the reading task, each in its own transaction; batching is left until the volume calls for it.

Review

crates/grain/src/jetstream.rs::follow (the cursor probe), then crates/grain-serve/src/jetstream.rs::apply.

🤖 Generated with Claude Code

This adds a Jetstream v2 reader to grain (follow, with backoff, ping-on-idle and the cursor probe) and grain-serve applies its events to the index. I read both source files at the head commit and found nothing that blocks the merge. I did not read the settings or the tests, and I did not read the diff itself (it was too large and includes the stacked commits). Nothing was built or run.

The probe logic in follow holds up. The counter resets on a connect, the probe fires on every third refusal, and a refused probe keeps the cursor. apply is also sound: signal collections track the author, untracked accounts are dropped, and a record that doesn't fit its lexicon is skipped rather than fatal.

Two smaller points, which can be fixed here or in a later pull:

• Feed::cursor_lost leaves self.seq set to the abandoned cursor. If the live-tip connection ends before any event arrives, ended() saves the dead cursor again, and the next restart offers it until the probe gives it up a second time.
• When the index fails, event has already advanced seq, so the failed event is skipped for good once the cursor is saved. That may be the intended trade-off, since the PR says the stream goes on. If so, the doc comment should say that an index failure loses that event.

The pull still does what it says: a Jetstream v2 reader with backoff, ping-on-idle and the cursor probe, and an apply that keeps tracked accounts' records current. Nothing blocks the merge. Nothing was built or run.

My two earlier points were optional, and neither has been taken up:

• Feed::cursor_lost in crates/grain-serve/src/jetstream.rs only logs. It leaves self.seq at the abandoned cursor. If the live-tip connection ends before any event arrives, ended() saves that dead cursor again. The next restart then offers it until the probe gives it up a second time. Setting self.seq = None in cursor_lost would fix this.
• Setting seq before apply means an index failure loses that event once the cursor is saved. That may be intended, but the doc comment doesn't say so.

Both can be fixed here or in a later pull.

The pull still does what it says: a Jetstream v2 reader with backoff, ping-on-idle and the cursor probe, plus an apply that keeps tracked accounts' records current. Nothing blocks the merge. I did not build or run anything.

My two earlier points were optional, and neither has been taken up:

• Feed::cursor_lost in crates/grain-serve/src/jetstream.rs only logs. It leaves self.seq at the abandoned cursor. If the live-tip connection ends before any event arrives, ended() saves that dead cursor again. The next restart then offers it until the probe gives it up a second time. Setting self.seq = None in cursor_lost would fix this.
• self.seq is set before apply runs. So if the index fails, that event is lost once the cursor is saved. That may be intended, but the doc comment doesn't say so.

Both can be fixed here or in a later pull.

Merged at 6101c42. review.knotless.dev had used its three rounds (last approval at 3cb9f56), so I read the rest myself: 2ed6a0c (a dropped cursor is also cleared from the index), 752504a (test timing), and the resolutions in the merges 3b24b9d and 6101c42 (one Settings, backfill's SIGNALS, writes at the commit's rev through put_rev/delete_rev, resync for accounts newly tracked or back). All four checks passed at 6101c42.