[READ-ONLY] Mirror of https://github.com/flo-bit/contrail. atproto backend in a bottle
flo-bit.dev/contrail/
test(ingest): cover caught-up break, filtered live-edge, fast-flow, identity
Add regression tests locking the ingestEvents loop behavior reviewed for the
quiet-stream fix:
- caught-up break collects the batch + cursor when a kept event reaches the
live edge (the exit path neither existing test exercised)
- the caught-up check now fires on a filtered live event too (early return),
deferring a following kept event to the next cycle rather than collecting it
- fast-flowing events within the deadline are all collected (the
next()/timeout race drops nothing)
- #identity events surface as handle updates through the ingest path
Tom Scanlan
(Jun 10, 2026, 2:43 PM EDT)
e362ddab
91278c6d
+102
+102
packages/contrail/tests/ingest-hang.test.ts
+102
packages/contrail/tests/ingest-hang.test.ts
···
149
149
jetstream.abort = true;
150
150
}
151
151
});
152
+
153
+
it("breaks via 'caught up to present' and returns the batch + cursor when a kept event reaches the live edge", async () => {
154
+
// `live` is >= ingestEvents' startTimeUs (captured at the call below), so the
155
+
// 5s safety timeout is irrelevant — the exit must be the caught-up break.
156
+
const liveUs = Date.now() * 1000 + 5_000_000;
157
+
jetstream.script = async function* (self) {
158
+
self.cursor = 1_000_000;
159
+
yield commitEvent("did:plc:author", "community.lexicon.calendar.event", 1_000_000, "hist");
160
+
self.cursor = liveUs;
161
+
yield commitEvent("did:plc:author", "community.lexicon.calendar.event", liveUs, "live");
162
+
// Must NOT be reached: the caught-up break fires on `live` before this.
163
+
self.cursor = liveUs + 1_000;
164
+
yield commitEvent("did:plc:author", "community.lexicon.calendar.event", liveUs + 1_000, "after");
165
+
await new Promise(() => {});
166
+
};
167
+
168
+
const result = await withTimeout(
169
+
ingestEvents(discoverableConfig(), 999_999, 5_000),
170
+
2_000,
171
+
"caught-up kept",
172
+
);
173
+
174
+
expect(result.events.map((e) => e.rkey)).toEqual(["hist", "live"]);
175
+
expect(result.lastCursor).toBe(liveUs);
176
+
});
177
+
178
+
it("caught-up break fires even on a filtered live event, deferring a following kept event to the next cycle", async () => {
179
+
// Behavior change vs the pre-fix loop: filtering now uses early `return`, so
180
+
// the caught-up check runs after a filtered event too. A filtered event at
181
+
// the live edge breaks the loop BEFORE the kept event that follows it.
182
+
// Pre-fix (`continue`) would have skipped the check, collected `evt`, and
183
+
// broken on it with cursor liveUs+1000.
184
+
const liveUs = Date.now() * 1000 + 5_000_000;
185
+
const knownDids = new Set<string>(); // empty -> the follow is filtered (unknown DID)
186
+
jetstream.script = async function* (self) {
187
+
self.cursor = liveUs;
188
+
yield commitEvent("did:plc:stranger", "app.bsky.graph.follow", liveUs, "f1");
189
+
self.cursor = liveUs + 1_000;
190
+
yield commitEvent("did:plc:author", "community.lexicon.calendar.event", liveUs + 1_000, "evt");
191
+
await new Promise(() => {});
192
+
};
193
+
194
+
const result = await withTimeout(
195
+
ingestEvents(dependentConfig(), 999_999, 5_000, knownDids),
196
+
2_000,
197
+
"caught-up filtered",
198
+
);
199
+
200
+
expect(result.events).toHaveLength(0); // kept `evt` deferred, not collected this cycle
201
+
expect(result.lastCursor).toBe(liveUs); // broke at the filtered event, before `evt`
202
+
});
203
+
204
+
it("collects every event when they flow fast but within the safety timeout (the next()/timeout race drops nothing)", async () => {
205
+
const N = 25;
206
+
jetstream.script = async function* (self) {
207
+
let t = 1_000_000; // all historical, so caught-up never fires; deadline ends it
208
+
for (let i = 0; i < N; i++) {
209
+
t += 1_000;
210
+
self.cursor = t;
211
+
// A real await before each yield forces next() down the pending-promise
212
+
// path (not the queue fast-path), so this exercises the race directly.
213
+
await new Promise((r) => setTimeout(r, 1));
214
+
yield commitEvent("did:plc:author", "community.lexicon.calendar.event", t, "e" + i);
215
+
}
216
+
await new Promise(() => {}); // then quiet -> safety timeout returns the batch
217
+
};
218
+
219
+
const result = await withTimeout(
220
+
ingestEvents(discoverableConfig(), 999_999, 300),
221
+
2_000,
222
+
"fast flow",
223
+
);
224
+
225
+
expect(result.events).toHaveLength(N);
226
+
});
227
+
228
+
it("captures #identity events as handle updates through the ingest path", async () => {
229
+
jetstream.script = async function* (self) {
230
+
self.cursor = 1_000_000;
231
+
yield {
232
+
kind: "identity" as const,
233
+
time_us: 1_000_000,
234
+
did: "did:plc:author",
235
+
identity: {
236
+
did: "did:plc:author",
237
+
handle: "alice.test",
238
+
seq: 1,
239
+
time: "2026-04-01T10:00:00Z",
240
+
},
241
+
};
242
+
await new Promise(() => {}); // quiet -> safety timeout returns
243
+
};
244
+
245
+
const result = await withTimeout(
246
+
ingestEvents(discoverableConfig(), 999_999, 150),
247
+
2_000,
248
+
"identity",
249
+
);
250
+
251
+
expect(result.identityUpdates.get("did:plc:author")).toBe("alice.test");
252
+
expect(result.events).toHaveLength(0); // identity events are not record commits
253
+
});
152
254
});