~ knotless

chadtmiller.com/knotless

Git hosted in your atproto account, on any stock PDS, fast enough for agents.

git clone knotless://chadtmiller.com/knotless

needs the knotless git helper. set it up

Jetstream: give up a silent connection, and resume from the index

A connection gone half-open delivered nothing and never ended, so the
watcher waited on it forever: prod stopped indexing for three hours. Now a
stream silent through a ping is dropped and followed again, and Jetstream's
cursor is kept in the index, so a restart replays what it missed.

CLAUDE.md: land changes as pulls rather than on main.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Chad Miller and Claude Opus 5.5 committed on Oct 22, 2026 parent c2f591c commit a0ee6a9
Showing 5 changed files with 171 additions and 11 deletions
CLAUDE.md+8−0
@@ -40,3 +40,11 @@ work is done.
`cargo fmt` can put a `?` on a line of its own, which coverage then counts
as a line never run: bind the value first, or use a named function for a
row mapper, rather than lowering the gate.
## Landing a change
Open a pull rather than committing to `main`: push the work to a branch
named for it (`web/mermaid`, `serve/jetstream`), then `knotless pull
knotless://chadtmiller.com/knotless main knotless://chadtmiller.com/knotless
<branch> "<title>" -m "<body>"`. It is merged into `main` and deployed from
there.
crates/knotless-serve/src/firehose.rs+34−3
@@ -3,12 +3,42 @@
//! refreshes them, so the next clone needs no round trip to the PDS.
use crate::App;
use futures::StreamExt;
use futures::{SinkExt, StreamExt};
use knotless::records::nsid;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::{MaybeTlsStream, WebSocketStream};
/// How long a stream may stay silent before it is pinged, and then before
/// it is given up on.
pub const IDLE: Duration = Duration::from_secs(60);
/// A connection to a firehose or Jetstream.
pub type Socket = WebSocketStream<MaybeTlsStream<tokio::net::TcpStream>>;
/// Hands `each` the frames of `ws` until it ends, fails, or stays silent
/// through a ping: a connection gone half-open delivers nothing and never
/// ends, so without one a watcher waits on it forever.
pub async fn frames(ws: &mut Socket, idle: Duration, mut each: impl FnMut(Message)) {
let mut pinged = false;
loop {
match tokio::time::timeout(idle, ws.next()).await {
Ok(Some(Ok(msg))) => {
pinged = false;
each(msg);
}
Ok(_) => return,
Err(_) if pinged => return,
// A failed send shows as silence on the next wait.
Err(_) => {
pinged = true;
let _ = ws.send(Message::Ping(Default::default())).await;
}
}
}
}
/// What a firehose frame says happened.
#[derive(Debug, Default, PartialEq)]
@@ -102,14 +132,15 @@ pub async fn watch(app: Arc<App>, base: String) {
loop {
if let Ok((mut ws, _)) = tokio_tungstenite::connect_async(endpoint(&base, cursor)).await {
app.watching.store(true, Ordering::Relaxed);
while let Some(Ok(msg)) = ws.next().await {
frames(&mut ws, IDLE, |msg| {
if let Message::Binary(frame) = msg
&& let Some(event) = parse(&frame)
{
cursor = event.seq.or(cursor);
app.observe(event);
}
}
})
.await;
}
app.watching.store(false, Ordering::Relaxed);
tokio::time::sleep(Duration::from_millis(500)).await;
crates/knotless-serve/src/index.rs+22−1
@@ -66,7 +66,7 @@ const VERSION: i64 = 6;
const CACHE: &str =
"DROP TABLE IF EXISTS repo; DROP TABLE IF EXISTS forge; DROP TABLE IF EXISTS push;
DROP TABLE IF EXISTS moved; DROP TABLE IF EXISTS block; DROP TABLE IF EXISTS label;
DROP TABLE IF EXISTS labeler;";
DROP TABLE IF EXISTS labeler; DROP TABLE IF EXISTS stream;";
impl Index {
pub fn open(path: &Path) -> rusqlite::Result<Index> {
@@ -106,6 +106,7 @@ impl Index {
src TEXT NOT NULL, uri TEXT NOT NULL, val TEXT NOT NULL, exp TEXT,
PRIMARY KEY (src, uri, val));
CREATE TABLE IF NOT EXISTS labeler (src TEXT PRIMARY KEY, seq INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS stream (src TEXT PRIMARY KEY, seq INTEGER NOT NULL);
CREATE TABLE IF NOT EXISTS takedown (
subject TEXT PRIMARY KEY, value TEXT NOT NULL, at TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS emitted (seq INTEGER PRIMARY KEY AUTOINCREMENT, label TEXT NOT NULL);
@@ -364,6 +365,26 @@ impl Index {
.optional()
}
/// Where the index was last fed to from the event stream `src`.
pub fn stream_seq(&self, src: &str) -> rusqlite::Result<Option<i64>> {
let db = self.db.lock().unwrap();
db.query_row("SELECT seq FROM stream WHERE src = ?1", [src], |row| {
row.get(0)
})
.optional()
}
/// Records that the index was fed to `seq` from the event stream `src`.
pub fn set_stream_seq(&self, src: &str, seq: i64) -> rusqlite::Result<()> {
let db = self.db.lock().unwrap();
db.execute(
"INSERT INTO stream (src, seq) VALUES (?1, ?2)
ON CONFLICT (src) DO UPDATE SET seq = excluded.seq",
params![src, seq],
)?;
Ok(())
}
/// Keeps a report, as `com.atproto.moderation.createReport` answers it
/// without its `id`, and returns it with the `id` it was given.
pub fn report(&self, mut report: Value) -> rusqlite::Result<Value> {
crates/knotless-serve/src/jetstream.rs+31−7
@@ -10,12 +10,11 @@
//! the relay it follows, and the index is a cache either way.
use crate::App;
use crate::firehose::{Event, Op, knotless};
use futures::StreamExt;
use crate::firehose::{Event, IDLE, Op, frames, knotless};
use serde_json::Value;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::{Duration, Instant};
use tokio_tungstenite::tungstenite::Message;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::http::HeaderValue;
@@ -26,6 +25,9 @@ pub const COLLECTIONS: &str = "dev.knotless.*";
/// The one framing the endpoint speaks.
const SUBPROTOCOL: &str = "xrpc.v1.json";
/// How often the index is told where the stream was read to.
const SAVE: Duration = Duration::from_secs(5);
/// The payload `$type`s, after this prefix, name the event kind.
const KIND: &str = "network.bsky.jetstream.subscribeEvents#";
@@ -84,8 +86,24 @@ fn text(v: &Value, name: &str) -> Option<String> {
/// Follows Jetstream until the process ends, reconnecting after the last
/// sequence number seen. While disconnected, mirrors fall back to the TTL.
pub async fn watch(app: Arc<App>, base: String) {
let mut cursor = None;
pub fn watch(app: Arc<App>, base: String) -> impl Future<Output = ()> {
follow(app, base, IDLE)
}
/// Follows Jetstream, giving a connection up after it stays silent through
/// a ping `idle` long. With an index, it starts where the index was last
/// fed from `base`, so a restart misses nothing Jetstream still keeps.
pub async fn follow(app: Arc<App>, base: String, idle: Duration) {
let index = app.index.clone();
let mut cursor = index
.as_ref()
.and_then(|i| i.stream_seq(&base).ok().flatten());
let mut saved: Option<Instant> = None;
let save = |cursor: Option<i64>| {
if let (Some(index), Some(seq)) = (&index, cursor) {
let _ = index.set_stream_seq(&base, seq);
}
};
loop {
if let Ok(mut request) = endpoint(&base, cursor).into_client_request() {
request.headers_mut().insert(
@@ -94,15 +112,21 @@ pub async fn watch(app: Arc<App>, base: String) {
);
if let Ok((mut ws, _)) = tokio_tungstenite::connect_async(request).await {
app.watching.store(true, Ordering::Relaxed);
while let Some(Ok(msg)) = ws.next().await {
frames(&mut ws, idle, |msg| {
if let Message::Text(frame) = msg
&& let Some(event) = parse(&frame)
{
// The cursor is inclusive: resume after the last one.
cursor = event.seq.map(|s| s + 1).or(cursor);
app.observe(event);
if saved.is_none_or(|at| at.elapsed() >= SAVE) {
save(cursor);
saved = Some(Instant::now());
}
}
}
})
.await;
save(cursor);
}
}
app.watching.store(false, Ordering::Relaxed);
crates/knotless-serve/tests/jetstream.rs+76−0
@@ -153,6 +153,9 @@ fn the_index_fills_from_jetstream() {
let (third, _) = rt.block_on(asks.recv()).unwrap();
assert!(third.ends_with("&cursor=15"), "{third}");
// The index keeps where it was fed to, for a restart to resume from.
assert_eq!(index.stream_seq(&format!("ws://{addr}")).unwrap(), Some(15));
// A Jetstream it cannot reach, or cannot even name, leaves the TTL in
// charge.
for url in ["ws://127.0.0.1:9", "not a url"] {
@@ -165,3 +168,76 @@ fn the_index_fills_from_jetstream() {
wait_until(|| !lost.watching.load(Ordering::Relaxed));
}
}
// The handshake callback's error type is tungstenite's, not ours.
#[allow(clippy::result_large_err)]
#[test]
fn a_silent_jetstream_is_left() {
let dir = tempfile::tempdir().unwrap();
let rt = tokio::runtime::Runtime::new().unwrap();
let listener = rt
.block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
.unwrap();
let addr = listener.local_addr().unwrap();
let (asked, mut asks) = tokio::sync::mpsc::unbounded_channel();
rt.spawn(async move {
let mut held = Vec::new();
loop {
let (tcp, _) = listener.accept().await.unwrap();
let asked = asked.clone();
let mut ws =
tokio_tungstenite::accept_hdr_async(tcp, |req: &Request, mut res: Response| {
asked.send(req.uri().to_string()).unwrap();
let protocol = req.headers()["sec-websocket-protocol"].clone();
res.headers_mut().insert("sec-websocket-protocol", protocol);
Ok(res)
})
.await
.unwrap();
ws.send(commit(
7,
"did:plc:owner",
"delete",
"dev.knotless.issue",
"3l3qo2vuowo2b",
None,
))
.await
.unwrap();
// Never read again, so its pings go unanswered: a connection
// gone half-open.
held.push(ws);
}
});
// Restarted, it starts where the index was last fed to.
let index = Index::open(&dir.path().join("index.db")).unwrap();
index.set_stream_seq(&format!("ws://{addr}"), 5).unwrap();
index.set_stream_seq(&format!("ws://{addr}"), 6).unwrap();
let app = App::build(
knotless::env::Ctx::new(knotless::env::Env::from_vars(|_| None)),
Duration::ZERO,
None,
Some(index),
false,
);
rt.spawn(knotless_serve::jetstream::follow(
app.clone(),
format!("ws://{addr}"),
Duration::from_millis(100),
));
let first = rt.block_on(asks.recv()).unwrap();
assert!(first.ends_with("&cursor=6"), "{first}");
// Silent through a ping, it is given up on and followed again from
// past what it said.
let second = rt.block_on(asks.recv()).unwrap();
assert!(second.ends_with("&cursor=8"), "{second}");
assert_eq!(
app.index
.as_ref()
.unwrap()
.stream_seq(&format!("ws://{addr}"))
.unwrap(),
Some(8)
);
}