[READ-ONLY] Mirror of https://github.com/flo-bit/contrail. atproto backend in a bottle
flo-bit.dev/contrail/
fix(ingest): avoid re-ingesting 10s every cron cycle for single-instance jetstream
@atcute/jetstream rolls the cursor back 10s on first connect when given an
array url, to absorb clock skew across a randomly-selected pool of
instances. It is meant to fire once per session. But contrail's cron
ingestion rebuilds the JetstreamSubscription every cycle, so for a
single-instance config that "first-connect" rollback fires every tick and
redundantly re-delivers the last 10s of events each cycle.
Add a jetstreamUrlOption(jetstreams) helper in contrail-base that maps the
configured list onto @atcute's url shape by topology:
- one instance -> a string (one fixed instance, no skew, no rollback)
- a real pool (2+) -> an array (cross-instance rollback preserved)
Applied at both construction sites (the cron ingestEvents path and the
runPersistent daemon). Setting two jetstreams does not sidestep this: an
array keeps the per-cycle rollback, and with genuine cross-instance skew
that rollback is then actually needed. A string for the single-instance
case is the only shape that removes the redundant re-ingest.
Also fix a misleading per-cycle reconnect log that claimed every reconnect
picks a URL at random and rolls back 10s, which is only true for a
multi-instance pool. The warning is now gated on urls.length > 1 and
reports the actual rolled_back value; single-instance reconnects log at
info level.
Tests: a helper unit test (single->string, pool->array, empty->empty), an
ingestEvents wiring test, and a characterization test that drives the real
JetstreamSubscription (mocking only the injected socket) to pin the
array-rolls-back / string-does-not behavior the fix relies on, so a future
@atcute change fails loudly here.
authored by
Tom Scanlan
and committed by
GitHub
(Jun 15, 2026, 7:01 PM EDT)
98947870
833a6590
+248
-5
+10
.changeset/single-jetstream-no-cursor-rollback.md
+10
.changeset/single-jetstream-no-cursor-rollback.md
···
1
+
---
2
+
"@atmo-dev/contrail-base": patch
3
+
"@atmo-dev/contrail-appview": patch
4
+
---
5
+
6
+
Stop re-ingesting the last 10s on every cron cycle for single-instance jetstream configs.
7
+
8
+
`@atcute/jetstream` rolls the cursor back 10s on the first connect when given an array `url`, to absorb clock skew across a pool of interchangeable instances. Contrail's cron ingestion rebuilds the subscription every cycle, so for a single-instance config that once-per-session rollback fired every cycle and redundantly re-delivered the last 10s of events. A new `jetstreamUrlOption` helper hands a one-element config to `@atcute` as a string (one fixed instance, no skew, no rollback) while leaving real multi-instance pools as an array so their cross-instance rollback is preserved. Applied at both subscription construction sites (cron `ingestEvents` and the persistent daemon).
9
+
10
+
The per-cycle reconnect log is now accurate for single-instance configs: a reconnect to one fixed instance no longer claims to "pick a URL at random and roll the cursor back 10s" (it can't). The warning now fires only for multi-instance pools and reports the actual `rolled_back` value; single-instance reconnects log at info level confirming no rollback.
+28
packages/contrail-base/src/types.ts
+28
packages/contrail-base/src/types.ts
···
195
195
"wss://jetstream1.us-east.bsky.network",
196
196
];
197
197
198
+
/**
199
+
* Shape a configured jetstream list for `@atcute/jetstream`'s `url` option.
200
+
*
201
+
* @atcute distinguishes a string url (one fixed instance) from an array url (a
202
+
* pool it picks from at random each connect). For an array it seeds
203
+
* `#lastUsedUrl=''` and rolls the cursor back 10s on the first connect, to absorb
204
+
* clock skew between whichever pooled instances a resumed cursor may have
205
+
* crossed. A string takes no rollback: a single instance emits a monotonic
206
+
* cursor, so resuming at the saved value on that same instance can't skip its own
207
+
* events — there is no second instance to be skewed against.
208
+
*
209
+
* Contrail's cron ingestion rebuilds the subscription every cycle, so for a
210
+
* single-instance config that "first-connect" rollback fires *every* cycle and
211
+
* redundantly re-ingests the last 10s. Collapsing a one-element pool to a string
212
+
* matches @atcute's own single-instance semantics and drops that dead margin; a
213
+
* real pool (2+) stays an array so the cross-instance rollback is preserved.
214
+
*/
215
+
export function jetstreamUrlOption(jetstreams: string[]): string | string[] {
216
+
return jetstreams.length === 1 ? jetstreams[0] : jetstreams;
217
+
}
218
+
198
219
export const DEFAULT_RELAYS = [
199
220
"https://relay1.us-east.bsky.network"
200
221
];
···
212
233
collections: Record<string, CollectionConfig>;
213
234
profiles?: (string | ProfileConfig)[];
214
235
relays?: string[];
236
+
/** Jetstream endpoints to ingest from (defaults to {@link DEFAULT_JETSTREAMS}).
237
+
* Prefer a single endpoint: one instance has no clock skew, so `@atcute` takes
238
+
* no cursor rollback (see {@link jetstreamUrlOption}) — important for the cron
239
+
* model, which rebuilds the subscription every cycle. Use 2+ only for failover
240
+
* across interchangeable endpoints, and ideally only with a persistent
241
+
* connection (`runPersistent`), where the per-switch 10s skew rollback fires
242
+
* about once rather than every cycle. */
215
243
jetstreams?: string[];
216
244
feeds?: Record<string, FeedConfig>;
217
245
logger?: Logger;
+94
packages/contrail/tests/jetstream-rollback-behavior.test.ts
+94
packages/contrail/tests/jetstream-rollback-behavior.test.ts
···
1
+
import { describe, it, expect } from "vitest";
2
+
import { JetstreamSubscription } from "@atcute/jetstream";
3
+
import { jetstreamUrlOption } from "../src/core/types";
4
+
5
+
// This test does NOT mock @atcute. It drives the real JetstreamSubscription and
6
+
// captures the actual cursor it would connect with, to PIN the assumption our
7
+
// fix relies on: an array url rolls the cursor back 10s on first connect, a
8
+
// string url does not. partysocket (the real socket layer @atcute uses) accepts
9
+
// an injected `WebSocket` constructor via the `ws` option, so we capture the
10
+
// connect URL there without touching @atcute or the network.
11
+
const TEN_SECONDS_US = 10_000_000;
12
+
13
+
/** Construct a real subscription, let it build its connect URL, and return the
14
+
* `cursor` query param @atcute put on it. */
15
+
async function firstConnectCursor(
16
+
url: string | string[],
17
+
cursor: number,
18
+
): Promise<number> {
19
+
const connectUrls: string[] = [];
20
+
21
+
// Minimal stand-in for a browser WebSocket. partysocket constructs this with
22
+
// the resolved subscribe URL; we record it and otherwise no-op.
23
+
class CapturingSocket {
24
+
static readonly CONNECTING = 0;
25
+
static readonly OPEN = 1;
26
+
static readonly CLOSING = 2;
27
+
static readonly CLOSED = 3;
28
+
readyState = CapturingSocket.CONNECTING;
29
+
binaryType = "blob";
30
+
constructor(connectUrl: string | URL) {
31
+
connectUrls.push(String(connectUrl));
32
+
}
33
+
addEventListener() {}
34
+
removeEventListener() {}
35
+
send() {}
36
+
close() {}
37
+
}
38
+
39
+
const sub = new JetstreamSubscription({
40
+
url,
41
+
cursor,
42
+
wantedCollections: [],
43
+
ws: { WebSocket: CapturingSocket as unknown as typeof WebSocket },
44
+
});
45
+
46
+
// Subscribing triggers the lazy #create() that builds the connect URL.
47
+
const iterator = sub[Symbol.asyncIterator]();
48
+
void iterator.next();
49
+
50
+
// partysocket resolves the URL provider and constructs the socket on a later
51
+
// tick; give it a few macrotasks to do so.
52
+
for (let i = 0; i < 10 && connectUrls.length === 0; i++) {
53
+
await new Promise((r) => setTimeout(r, 5));
54
+
}
55
+
await iterator.return?.();
56
+
57
+
if (connectUrls.length === 0) {
58
+
throw new Error("subscription never attempted a connection");
59
+
}
60
+
const got = new URL(connectUrls[0]).searchParams.get("cursor");
61
+
return Number(got);
62
+
}
63
+
64
+
describe("@atcute jetstream cursor rollback (behavior our fix depends on)", () => {
65
+
it("rolls the cursor back 10s on first connect for an array url", async () => {
66
+
const cursor = 2_000_000_000;
67
+
const got = await firstConnectCursor(["wss://jetstream.example"], cursor);
68
+
expect(got).toBe(cursor - TEN_SECONDS_US);
69
+
});
70
+
71
+
it("does NOT roll the cursor back for a string url", async () => {
72
+
const cursor = 2_000_000_000;
73
+
const got = await firstConnectCursor("wss://jetstream.example", cursor);
74
+
expect(got).toBe(cursor);
75
+
});
76
+
77
+
it("jetstreamUrlOption gives a single-instance config the no-rollback (string) shape", async () => {
78
+
const cursor = 2_000_000_000;
79
+
const got = await firstConnectCursor(
80
+
jetstreamUrlOption(["wss://jetstream.example"]),
81
+
cursor,
82
+
);
83
+
expect(got).toBe(cursor);
84
+
});
85
+
86
+
it("jetstreamUrlOption keeps a real pool on the rollback (array) shape", async () => {
87
+
const cursor = 2_000_000_000;
88
+
const got = await firstConnectCursor(
89
+
jetstreamUrlOption(["wss://a.example", "wss://b.example"]),
90
+
cursor,
91
+
);
92
+
expect(got).toBe(cursor - TEN_SECONDS_US);
93
+
});
94
+
});
+67
packages/contrail/tests/jetstream-single-url.test.ts
+67
packages/contrail/tests/jetstream-single-url.test.ts
···
1
+
import { describe, it, expect, vi, beforeEach } from "vitest";
2
+
3
+
// Records what `url` shape ingestEvents hands to @atcute's JetstreamSubscription.
4
+
// @atcute treats a string url and an array url differently: for an ARRAY it seeds
5
+
// `#lastUsedUrl = ''`, so the very first connect satisfies its "switched instance"
6
+
// check and rolls the cursor back 10s — even when the array holds a single URL and
7
+
// nothing switched. Passing a string seeds `#lastUsedUrl = undefined`, so no
8
+
// spurious rollback (and thus no re-delivered events / DUPLICATE-in-cycle flood).
9
+
const recorder = vi.hoisted(() => ({ url: undefined as unknown }));
10
+
11
+
vi.mock("@atcute/jetstream", () => {
12
+
class MockJetstreamSubscription {
13
+
cursor: number | null = null;
14
+
constructor(opts: { url?: unknown; cursor?: number }) {
15
+
recorder.url = opts?.url;
16
+
this.cursor = typeof opts?.cursor === "number" ? opts.cursor : null;
17
+
}
18
+
async *[Symbol.asyncIterator]() {
19
+
// Quiet stream: block so ingestEvents exits via its safety timeout.
20
+
await new Promise(() => {});
21
+
}
22
+
}
23
+
return { JetstreamSubscription: MockJetstreamSubscription };
24
+
});
25
+
26
+
import { ingestEvents } from "../src/core/jetstream";
27
+
import { resolveConfig } from "../src/core/types";
28
+
import type { ContrailConfig } from "../src/core/types";
29
+
30
+
const silentLogger = { log() {}, warn() {}, error() {} };
31
+
32
+
function configWithJetstreams(jetstreams: string[]): ContrailConfig {
33
+
return {
34
+
...resolveConfig({
35
+
namespace: "com.example",
36
+
jetstreams,
37
+
collections: {
38
+
event: { collection: "community.lexicon.calendar.event" },
39
+
},
40
+
}),
41
+
logger: silentLogger,
42
+
};
43
+
}
44
+
45
+
describe("ingestEvents — jetstream url shape (om-6zdo)", () => {
46
+
beforeEach(() => {
47
+
recorder.url = undefined;
48
+
});
49
+
50
+
it("passes a single configured jetstream as a string (avoids @atcute's array-only cursor rollback)", async () => {
51
+
await ingestEvents(
52
+
configWithJetstreams(["wss://jetstream.example/subscribe"]),
53
+
999_999,
54
+
50,
55
+
);
56
+
expect(recorder.url).toBe("wss://jetstream.example/subscribe");
57
+
});
58
+
59
+
it("passes multiple configured jetstreams as an array (preserves failover + intentional switch rollback)", async () => {
60
+
await ingestEvents(
61
+
configWithJetstreams(["wss://a.example", "wss://b.example"]),
62
+
999_999,
63
+
50,
64
+
);
65
+
expect(recorder.url).toEqual(["wss://a.example", "wss://b.example"]);
66
+
});
67
+
});
+27
packages/contrail/tests/jetstream-url-option.test.ts
+27
packages/contrail/tests/jetstream-url-option.test.ts
···
1
+
import { describe, it, expect } from "vitest";
2
+
import { jetstreamUrlOption } from "../src/core/types";
3
+
4
+
// @atcute/jetstream treats a string url and an array url differently *by design*:
5
+
// an array is a pool of interchangeable instances, so it seeds `#lastUsedUrl=''`
6
+
// and rolls the cursor back 10s on first connect to absorb clock skew between
7
+
// whichever instances a cursor crossed. A string is one fixed instance — no skew,
8
+
// no rollback. Contrail's cron model rebuilds the subscription every cycle, so for
9
+
// a single-instance config that "first-connect" rollback fires *every* cycle and
10
+
// redundantly re-ingests 10s. Collapsing a 1-instance pool to a string opts out of
11
+
// a safety margin we don't need; a real pool (2+) keeps it.
12
+
describe("jetstreamUrlOption", () => {
13
+
it("collapses a single-instance pool to a string (no per-cycle skew rollback)", () => {
14
+
expect(jetstreamUrlOption(["wss://one"])).toBe("wss://one");
15
+
});
16
+
17
+
it("keeps a real pool as an array (preserves cross-instance skew rollback)", () => {
18
+
expect(jetstreamUrlOption(["wss://a", "wss://b"])).toEqual([
19
+
"wss://a",
20
+
"wss://b",
21
+
]);
22
+
});
23
+
24
+
it("returns an empty list unchanged", () => {
25
+
expect(jetstreamUrlOption([])).toEqual([]);
26
+
});
27
+
});
+18
-4
packages/contrail-appview/src/core/jetstream.ts
+18
-4
packages/contrail-appview/src/core/jetstream.ts
···
3
3
import {
4
4
getCollectionNsids,
5
5
getDependentNsids,
6
+
jetstreamUrlOption,
6
7
shortNameForNsid,
7
8
buildFeedTargetCaps,
8
9
getFeedMutatingNsids,
···
210
211
const identityUpdates = new Map<string, string>();
211
212
212
213
const subscription = new JetstreamSubscription({
213
-
url: urls,
214
+
// A single-instance config is handed over as a string so @atcute skips its
215
+
// array-only first-connect cursor rollback (see jetstreamUrlOption). On the
216
+
// cron model that rollback would otherwise re-ingest 10s every cycle.
217
+
url: jetstreamUrlOption(urls),
214
218
wantedCollections: collections,
215
219
...(cursor !== null ? { cursor } : {}),
216
220
onConnectionOpen() {
···
390
394
}
391
395
392
396
if (connectCount > 1) {
393
-
log.warn(
394
-
`[ingest] RECONNECTED ${connectCount} times during cycle — each reconnect picks a URL at random and rolls cursor back 10s`
395
-
);
397
+
if (urls.length > 1) {
398
+
// Multi-instance pool: each reconnect picks a URL at random, and @atcute
399
+
// rolls the cursor back up to 10s on a fresh instance to absorb clock skew.
400
+
log.warn(
401
+
`[ingest] RECONNECTED ${connectCount} times during cycle across a ${urls.length}-instance pool — each reconnect picks a URL at random and may roll the cursor back up to 10s (rolled_back=${rolledBackUs}us this cycle)`
402
+
);
403
+
} else {
404
+
// Single fixed instance (see jetstreamUrlOption): reconnects resume on the
405
+
// same instance from the saved cursor, so there is no rollback.
406
+
log.log(
407
+
`[ingest] reconnected ${connectCount} times during cycle to the single fixed instance — no cursor rollback (rolled_back=${rolledBackUs}us)`
408
+
);
409
+
}
396
410
}
397
411
398
412
return { events: collected, lastCursor, newlyKnownDids: [...newlyKnownDids], identityUpdates };
+4
-1
packages/contrail-appview/src/core/persistent.ts
+4
-1
packages/contrail-appview/src/core/persistent.ts
···
5
5
getDependentNsids,
6
6
buildFeedTargetCaps,
7
7
getFeedMutatingNsids,
8
+
jetstreamUrlOption,
8
9
resolveConfig,
9
10
shortNameForNsid,
10
11
} from "./types";
···
138
139
const subscription = opts.createSubscription
139
140
? opts.createSubscription(cursor)
140
141
: new (await import("@atcute/jetstream")).JetstreamSubscription({
141
-
url: config.jetstreams ?? [],
142
+
// Single-instance config → string, so @atcute skips its array-only
143
+
// first-connect cursor rollback (see jetstreamUrlOption).
144
+
url: jetstreamUrlOption(config.jetstreams ?? []),
142
145
wantedCollections: collections,
143
146
...(cursor !== null ? { cursor } : {}),
144
147
onConnectionOpen() { log.log("Connected to Jetstream"); },