[READ-ONLY] Mirror of https://github.com/flo-bit/contrail. atproto backend in a bottle
flo-bit.dev/contrail/
identity fix, heal missing handles
+170
-17
+25
.changeset/identity-handle-resolution.md
+25
.changeset/identity-handle-resolution.md
···
1
+
---
2
+
"@atmo-dev/contrail-base": patch
3
+
---
4
+
5
+
fix(identity): stop stranding/clobbering handles during resolution (#42)
6
+
7
+
Backfill left a meaningful fraction of identities with a PDS but no handle.
8
+
Two root causes:
9
+
10
+
- `resolvePDSCached` short-circuited on any row with a non-null PDS and returned
11
+
without ever resolving the handle. A partial resolution (slingshot can return
12
+
a PDS without a handle under load) was therefore persisted and never healed.
13
+
It now treats a row as a complete cache hit only when both PDS *and* handle are
14
+
present; a PDS-only row falls through to re-resolve and fill the handle, while
15
+
still serving the known PDS (including if the re-resolution fails).
16
+
- `saveIdentity` overwrote `handle`/`pds` unconditionally, so
17
+
`refreshStaleIdentities` (which passes a null handle through when slingshot
18
+
omits one) could clobber a previously-resolved handle with null. The upsert now
19
+
COALESCEs both columns: a fresh non-null value still applies (handle changes
20
+
work), but a null never nulls a good value.
21
+
22
+
Backfill also resolves PDS endpoints up front instead of in a detached
23
+
background promise, so identity resolution no longer competes with record
24
+
backfill for slingshot — reducing the partial responses that triggered the
25
+
above in the first place.
+14
-5
packages/contrail-base/src/client.ts
+14
-5
packages/contrail-base/src/client.ts
···
196
196
db?: Database,
197
197
config?: ContrailConfig,
198
198
): Promise<string | undefined> {
199
+
let knownPds: string | undefined;
199
200
if (db) {
200
201
const cached = await db
201
-
.prepare("SELECT pds FROM identities WHERE did = ? AND pds IS NOT NULL")
202
+
.prepare("SELECT pds, handle FROM identities WHERE did = ? AND pds IS NOT NULL")
202
203
.bind(did)
203
-
.first<{ pds: string }>();
204
+
.first<{ pds: string; handle: string | null }>();
204
205
if (cached?.pds) {
205
206
pdsCacheSet(did, cached.pds);
206
-
return cached.pds;
207
+
// A row with both a PDS and a handle is a complete cache hit. A row with
208
+
// a PDS but no handle is a *partial* resolution — slingshot can return a
209
+
// PDS without a handle under load — so fall through to re-resolve and
210
+
// fill the handle instead of stranding the row forever (this DB
211
+
// short-circuit previously meant the handle was never backfilled). We
212
+
// keep serving the known PDS meanwhile, including if the re-resolve fails.
213
+
if (cached.handle) return cached.pds;
214
+
knownPds = cached.pds;
207
215
}
208
216
}
209
217
210
218
const resolved = await resolvePDS(did, config);
211
-
if (!resolved?.pds) return undefined;
219
+
if (!resolved?.pds) return knownPds;
212
220
213
221
pdsCacheSet(did, resolved.pds);
214
222
215
-
// Persist to DB for future runs
223
+
// Persist to DB for future runs. COALESCE keeps an existing handle when this
224
+
// resolution didn't return one, and never nulls a good handle.
216
225
if (db) {
217
226
await db
218
227
.prepare(
+5
-1
packages/contrail-base/src/identity.ts
+5
-1
packages/contrail-base/src/identity.ts
···
15
15
async function saveIdentity(db: Database, identity: Identity): Promise<void> {
16
16
await db
17
17
.prepare(
18
-
"INSERT INTO identities (did, handle, pds, resolved_at) VALUES (?, ?, ?, ?) ON CONFLICT(did) DO UPDATE SET handle = excluded.handle, pds = excluded.pds, resolved_at = excluded.resolved_at"
18
+
// COALESCE so a null handle/pds from a partial resolution never clobbers
19
+
// a previously-resolved value (e.g. refreshStaleIdentities passes through
20
+
// a null handle when slingshot omits it). A fresh non-null value still
21
+
// overwrites — handle changes apply normally.
22
+
"INSERT INTO identities (did, handle, pds, resolved_at) VALUES (?, ?, ?, ?) ON CONFLICT(did) DO UPDATE SET handle = COALESCE(excluded.handle, identities.handle), pds = COALESCE(excluded.pds, identities.pds), resolved_at = excluded.resolved_at"
19
23
)
20
24
.bind(identity.did, identity.handle, identity.pds, identity.resolved_at)
21
25
.run();
+114
packages/contrail/tests/identity-handle.test.ts
+114
packages/contrail/tests/identity-handle.test.ts
···
1
+
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
2
+
import { getPDS, __resetPdsCachesForTests } from "../src/core/client";
3
+
import { refreshStaleIdentities } from "../src/core/identity";
4
+
import { createTestDbWithSchema } from "./helpers";
5
+
import type { Database } from "../src/core/types";
6
+
import type { Did } from "@atcute/lexicons";
7
+
8
+
// Root-cause coverage for PR #42: identities that end up with a PDS but no
9
+
// handle must not be stranded, and a partial re-resolution must never clobber a
10
+
// handle that was already known.
11
+
12
+
const DID = "did:plc:aaaaaaaaaaaaaaaaaaaaaaaa" as Did;
13
+
const PDS = "https://pds.example.host.bsky.network";
14
+
15
+
let db: Database;
16
+
let fetchSpy: ReturnType<typeof vi.spyOn>;
17
+
18
+
async function seedIdentity(
19
+
did: string,
20
+
handle: string | null,
21
+
pds: string | null,
22
+
resolvedAt: number
23
+
): Promise<void> {
24
+
await db
25
+
.prepare(
26
+
"INSERT INTO identities (did, handle, pds, resolved_at) VALUES (?, ?, ?, ?)"
27
+
)
28
+
.bind(did, handle, pds, resolvedAt)
29
+
.run();
30
+
}
31
+
32
+
async function readIdentity(
33
+
did: string
34
+
): Promise<{ handle: string | null; pds: string | null; resolved_at: number } | null> {
35
+
return db
36
+
.prepare("SELECT handle, pds, resolved_at FROM identities WHERE did = ?")
37
+
.bind(did)
38
+
.first();
39
+
}
40
+
41
+
function slingshotReturns(body: { did?: string; handle?: string; pds?: string }) {
42
+
fetchSpy.mockResolvedValue(new Response(JSON.stringify(body), { status: 200 }));
43
+
}
44
+
45
+
beforeEach(async () => {
46
+
db = await createTestDbWithSchema();
47
+
__resetPdsCachesForTests();
48
+
fetchSpy = vi.spyOn(global, "fetch");
49
+
});
50
+
51
+
afterEach(() => {
52
+
fetchSpy.mockRestore();
53
+
__resetPdsCachesForTests();
54
+
});
55
+
56
+
describe("getPDS heals a PDS-but-no-handle row", () => {
57
+
it("re-resolves and backfills the missing handle", async () => {
58
+
// A partial resolution persisted earlier: PDS known, handle null.
59
+
await seedIdentity(DID, null, PDS, Date.now());
60
+
// Slingshot now returns the handle.
61
+
slingshotReturns({ did: DID, handle: "alice.test", pds: PDS });
62
+
63
+
const pds = await getPDS(DID, db);
64
+
65
+
expect(pds).toBe(PDS);
66
+
expect(fetchSpy).toHaveBeenCalled(); // it did re-resolve, not strand
67
+
expect((await readIdentity(DID))?.handle).toBe("alice.test");
68
+
});
69
+
70
+
it("still serves the known PDS if the heal resolution fails", async () => {
71
+
await seedIdentity(DID, null, PDS, Date.now());
72
+
fetchSpy.mockRejectedValue(new Error("slingshot down"));
73
+
74
+
const pds = await getPDS(DID, db);
75
+
76
+
expect(pds).toBe(PDS); // known PDS still returned
77
+
expect((await readIdentity(DID))?.handle).toBeNull();
78
+
});
79
+
80
+
it("does NOT re-resolve a complete row (PDS + handle)", async () => {
81
+
await seedIdentity(DID, "bob.test", PDS, Date.now());
82
+
83
+
const pds = await getPDS(DID, db);
84
+
85
+
expect(pds).toBe(PDS);
86
+
expect(fetchSpy).not.toHaveBeenCalled(); // short-circuit preserved
87
+
expect((await readIdentity(DID))?.handle).toBe("bob.test");
88
+
});
89
+
});
90
+
91
+
describe("refreshStaleIdentities does not clobber a known handle", () => {
92
+
it("keeps the existing handle when slingshot omits one", async () => {
93
+
// Stale row with a good handle.
94
+
await seedIdentity(DID, "carol.test", PDS, 0);
95
+
// Partial response: PDS but no handle.
96
+
slingshotReturns({ did: DID, pds: PDS });
97
+
98
+
await refreshStaleIdentities(db, [DID]);
99
+
100
+
const row = await readIdentity(DID);
101
+
expect(row?.handle).toBe("carol.test"); // preserved, not nulled
102
+
expect(row?.pds).toBe(PDS);
103
+
expect(row?.resolved_at).toBeGreaterThan(0); // refresh did run
104
+
});
105
+
106
+
it("still applies a changed handle (non-null overwrites)", async () => {
107
+
await seedIdentity(DID, "carol.test", PDS, 0);
108
+
slingshotReturns({ did: DID, handle: "carol.new", pds: PDS });
109
+
110
+
await refreshStaleIdentities(db, [DID]);
111
+
112
+
expect((await readIdentity(DID))?.handle).toBe("carol.new");
113
+
});
114
+
});
+12
-11
packages/contrail-appview/src/core/backfill.ts
+12
-11
packages/contrail-appview/src/core/backfill.ts
···
351
351
352
352
const dids = [...byDid.keys()];
353
353
354
-
// Resolve PDS endpoints in background (populates in-memory cache)
355
-
const resolvePromise = (async () => {
356
-
for (let i = 0; i < dids.length; i += 200) {
357
-
await Promise.allSettled(
358
-
dids.slice(i, i + 200).map((did) =>
359
-
getPDS(did as Did, db, config).catch(() => {})
360
-
)
361
-
);
362
-
}
363
-
})();
354
+
// Resolve PDS endpoints up front (populates the in-memory cache) rather
355
+
// than concurrently with the backfill passes below. Overlapping the two put
356
+
// identity resolution and record backfill in contention for slingshot at
357
+
// once, and the partial responses that produced (a PDS without a handle)
358
+
// got persisted and stranded. Resolving first keeps that load separate.
359
+
for (let i = 0; i < dids.length; i += 200) {
360
+
await Promise.allSettled(
361
+
dids.slice(i, i + 200).map((did) =>
362
+
getPDS(did as Did, db, config).catch(() => {})
363
+
)
364
+
);
365
+
}
364
366
365
367
let roundBackfilled = 0;
366
368
let usersComplete = 0;
···
472
474
}
473
475
}
474
476
475
-
await resolvePromise;
476
477
totalBackfilled += roundBackfilled;
477
478
478
479
// If nothing was backfilled this round, we're stuck