[READ-ONLY] Mirror of https://github.com/flo-bit/contrail. atproto backend in a bottle
flo-bit.dev/contrail/
fix: resolve config defensively in raw runPersistent export
The exported runPersistent(db, config, options) from @atmo-dev/contrail
silently skipped grouped relation count updates when passed an
unresolved ContrailConfig. Internals (applyEvents' buildBatchCountStatements,
query planning, etc.) read `config._resolved?.relations[...]?.groups` —
if `_resolved` was missing, the grouped-counts loop never ran, so columns
like count_rsvp_going stayed at 0 while total count_rsvp updated normally.
The Contrail class constructor resolves the config, so callers going
through `new Contrail(...).runPersistent()` were fine. Only the raw
export was affected — a real footgun for library consumers.
Fix: mirror what Contrail's constructor does and defensively call
resolveConfig when `_resolved` is absent. resolveConfig is idempotent,
so this is a no-op for already-resolved configs.
Regression test in packages/contrail/tests/persistent.test.ts passes a
raw ContrailConfig with a grouped relation and asserts
count_rsvp_going = 1. Confirmed to fail without this patch.
Reverts the Contrail.runPersistent() workaround in the postgres-devnet
e2e tests — they now call the raw runPersistent export directly, same
as any external library consumer would.
Tom Scanlan
(Apr 24, 2026, 5:53 PM EDT)
4ef9cf70
2cd5ea3b
+107
-16
+3
-7
apps/contrail-e2e/tests/cursor-resume.test.ts
+3
-7
apps/contrail-e2e/tests/cursor-resume.test.ts
···
24
24
import pg from "pg";
25
25
import { CredentialManager, Client } from "@atcute/client";
26
26
import "@atcute/atproto";
27
-
import { Contrail } from "@atmo-dev/contrail";
27
+
import { runPersistent } from "@atmo-dev/contrail";
28
28
import { createPostgresDatabase } from "@atmo-dev/contrail/postgres";
29
29
import { config as baseConfig } from "../config";
30
30
import {
···
73
73
74
74
it("resumes from saved cursor and picks up commits issued while down", async () => {
75
75
const db = createPostgresDatabase(pool);
76
-
// Go through the Contrail wrapper so the config is resolved; raw
77
-
// runPersistent expects a resolved config.
78
-
const contrail = new Contrail({ ...baseConfig, db });
79
-
await contrail.init();
80
76
const runOpts = { batchSize: 50, flushIntervalMs: 500 };
81
77
82
78
// ---- Phase 1: start ingester, publish A, verify indexed ----
83
79
const c1 = new AbortController();
84
-
const ingest1 = contrail.runPersistent({ ...runOpts, signal: c1.signal });
80
+
const ingest1 = runPersistent(db, baseConfig, { ...runOpts, signal: c1.signal });
85
81
86
82
const nameA = `A-${Date.now()}`;
87
83
const aRes = await client.post("com.atproto.repo.createRecord", {
···
164
160
165
161
// ---- Phase 4: restart ingester, assert replay ----
166
162
const c2 = new AbortController();
167
-
const ingest2 = contrail.runPersistent({ ...runOpts, signal: c2.signal });
163
+
const ingest2 = runPersistent(db, baseConfig, { ...runOpts, signal: c2.signal });
168
164
169
165
try {
170
166
const indexedB = await waitFor(
+2
-5
apps/contrail-e2e/tests/ingest-roundtrip.test.ts
+2
-5
apps/contrail-e2e/tests/ingest-roundtrip.test.ts
···
19
19
import pg from "pg";
20
20
import { CredentialManager, Client } from "@atcute/client";
21
21
import "@atcute/atproto";
22
-
import { Contrail } from "@atmo-dev/contrail";
22
+
import { Contrail, runPersistent } from "@atmo-dev/contrail";
23
23
import { createHandler } from "@atmo-dev/contrail/server";
24
24
import { createPostgresDatabase } from "@atmo-dev/contrail/postgres";
25
25
import { config as baseConfig } from "../config";
···
64
64
await contrail.init();
65
65
handle = createHandler(contrail);
66
66
67
-
// In-process ingester via Contrail so the config is resolved (raw
68
-
// runPersistent expects a resolved config — grouped counts silently
69
-
// don't update otherwise).
70
67
ingestController = new AbortController();
71
-
ingestPromise = contrail.runPersistent({
68
+
ingestPromise = runPersistent(db, baseConfig, {
72
69
batchSize: 50,
73
70
flushIntervalMs: 500,
74
71
signal: ingestController.signal,
+9
-2
packages/contrail/src/core/persistent.ts
+9
-2
packages/contrail/src/core/persistent.ts
···
1
1
import type { JetstreamSubscription } from "@atcute/jetstream";
2
-
import type { ContrailConfig, IngestEvent, Database, Logger } from "./types";
3
-
import { getCollectionNsids, getDependentNsids, DEFAULT_FEED_MAX_ITEMS } from "./types";
2
+
import type { ContrailConfig, IngestEvent, Database, Logger, ResolvedContrailConfig } from "./types";
3
+
import { getCollectionNsids, getDependentNsids, DEFAULT_FEED_MAX_ITEMS, resolveConfig } from "./types";
4
4
import { initSchema, getLastCursor, saveCursor, applyEvents, pruneFeedItems } from "./db";
5
5
import { refreshStaleIdentities } from "./identity";
6
6
import { createIngestState } from "./jetstream";
···
29
29
config: ContrailConfig,
30
30
options?: PersistentIngestOptions,
31
31
): Promise<void> {
32
+
// Internals (applyEvents, count updates, query planning) read `_resolved`
33
+
// and silently skip features when it's missing. The Contrail class resolves
34
+
// in its constructor; callers using this raw export must also get a resolved
35
+
// config, so do it defensively here. resolveConfig is idempotent.
36
+
if (!(config as ResolvedContrailConfig)._resolved) {
37
+
config = resolveConfig(config);
38
+
}
32
39
const log = getLogger(config, options);
33
40
const batchSize = options?.batchSize ?? 50;
34
41
const flushIntervalMs = options?.flushIntervalMs ?? 5_000;
+93
-2
packages/contrail/tests/persistent.test.ts
+93
-2
packages/contrail/tests/persistent.test.ts
···
1
1
import { describe, it, expect, vi, beforeEach } from "vitest";
2
-
import type { Database } from "../src/core/types";
3
-
import { createTestDbWithSchema, TEST_CONFIG } from "./helpers";
2
+
import type { ContrailConfig, Database } from "../src/core/types";
3
+
import { createTestDb, createTestDbWithSchema, TEST_CONFIG } from "./helpers";
4
4
import { runPersistent } from "../src/core/persistent";
5
5
import { getLastCursor, queryRecords } from "../src/core/db/records";
6
+
import { initSchema } from "../src/core/db/schema";
6
7
7
8
// Mock identity resolution to avoid network calls in tests
8
9
vi.mock("../src/core/identity", () => ({
···
199
200
200
201
controller.abort();
201
202
await promise;
203
+
});
204
+
205
+
it("resolves config internally when given an unresolved ContrailConfig", async () => {
206
+
// Regression test for the silent grouped-count bug: passing a raw
207
+
// ContrailConfig (no `_resolved`) used to let total counts update while
208
+
// grouped count columns stayed at 0. runPersistent must defensively
209
+
// resolve so consumers of the raw export get the same behavior as those
210
+
// going through the Contrail class wrapper.
211
+
const rawConfig: ContrailConfig = {
212
+
namespace: "com.example",
213
+
collections: {
214
+
event: {
215
+
collection: "community.lexicon.calendar.event",
216
+
relations: {
217
+
rsvps: {
218
+
collection: "rsvp",
219
+
groupBy: "status",
220
+
groups: {
221
+
going: "community.lexicon.calendar.rsvp#going",
222
+
notgoing: "community.lexicon.calendar.rsvp#notgoing",
223
+
},
224
+
},
225
+
},
226
+
},
227
+
rsvp: {
228
+
collection: "community.lexicon.calendar.rsvp",
229
+
references: {
230
+
event: { collection: "event", field: "subject.uri" },
231
+
},
232
+
},
233
+
},
234
+
};
235
+
// Sanity: the raw object must NOT have _resolved — that's the whole point.
236
+
expect((rawConfig as any)._resolved).toBeUndefined();
237
+
238
+
const freshDb = createTestDb();
239
+
await initSchema(freshDb, rawConfig);
240
+
241
+
const eventUri = "at://did:plc:alice/community.lexicon.calendar.event/evt1";
242
+
const events = [
243
+
{
244
+
kind: "commit" as const,
245
+
did: "did:plc:alice",
246
+
time_us: 9000,
247
+
commit: {
248
+
collection: "community.lexicon.calendar.event",
249
+
operation: "create",
250
+
rkey: "evt1",
251
+
cid: "cidE",
252
+
record: { name: "E", startsAt: "2026-04-01T10:00:00Z", mode: "online" },
253
+
},
254
+
},
255
+
{
256
+
kind: "commit" as const,
257
+
did: "did:plc:alice",
258
+
time_us: 9001,
259
+
commit: {
260
+
collection: "community.lexicon.calendar.rsvp",
261
+
operation: "create",
262
+
rkey: "r1",
263
+
cid: "cidR",
264
+
record: {
265
+
subject: { uri: eventUri, cid: "cidE" },
266
+
status: "community.lexicon.calendar.rsvp#going",
267
+
},
268
+
},
269
+
},
270
+
];
271
+
272
+
const controller = new AbortController();
273
+
const promise = runPersistent(freshDb, rawConfig, {
274
+
batchSize: 100,
275
+
flushIntervalMs: 50,
276
+
signal: controller.signal,
277
+
createSubscription: () => mockSubscription(events) as any,
278
+
});
279
+
await new Promise((r) => setTimeout(r, 300));
280
+
controller.abort();
281
+
await promise;
282
+
283
+
// Query the raw count columns directly — using queryRecords would hide
284
+
// the bug since it hydrates via the resolved config anyway.
285
+
const row = await freshDb
286
+
.prepare(`SELECT count_rsvp, count_rsvp_going FROM records_event WHERE uri = ?`)
287
+
.bind(eventUri)
288
+
.first<{ count_rsvp: number; count_rsvp_going: number }>();
289
+
expect(row).not.toBeNull();
290
+
expect(row!.count_rsvp).toBe(1);
291
+
// This is the assertion that would fail before the fix.
292
+
expect(row!.count_rsvp_going).toBe(1);
202
293
});
203
294
204
295
it("skips non-commit events", async () => {