+17

.changeset/fts-no-duplicate-rows-on-backfill.md

Reviewed

··· 1 1 + --- 2 2 + "@atmo-dev/contrail-appview": patch 3 3 + --- 4 4 + 5 5 + Stop accumulating duplicate FTS rows when records are re-applied during backfill. 6 6 + 7 7 + The FTS-sync path only deleted an existing FTS row before inserting when the 8 8 + record was already in `existingMap`. Backfill runs with `skipReplayDetection`, 9 9 + which leaves `existingMap` empty, so every re-applied record looked brand-new 10 10 + and appended another FTS row. The FTS virtual table has no uniqueness 11 11 + constraint, so these accumulated, and the search JOIN fanned each event out into 12 12 + one result row per duplicate. Make the delete-then-insert unconditional so FTS 13 13 + sync is idempotent regardless of replay detection. 14 14 + 15 15 + The unconditional delete also evicts a stale FTS row when an update clears all 16 16 + searchable fields: in that case there is no content to re-insert, but the prior 17 17 + row must still be removed so old terms stop matching.

+15 -11

packages/contrail-appview/src/core/db/records.ts

Reviewed

··· 161 161 function buildFtsStatements( 162 162 db: Database, 163 163 event: IngestEvent, 164 164 - config: ContrailConfig, 165 165 - existingMap: Map<string, ExistingRecordInfo> 164 164 + config: ContrailConfig 166 165 ): Statement[] { 167 166 // PostgreSQL: tsvector generated column is auto-maintained, no manual FTS sync 168 167 if (getDialect(db).ftsStrategy === "generated-column") return []; ··· 184 183 const record = event.record ? JSON.parse(event.record) : null; 185 184 if (!record) return []; 186 185 186 186 + // Always delete first so FTS sync is idempotent. The FTS virtual table has no 187 187 + // uniqueness constraint, so a bare insert appends a duplicate row when one 188 188 + // already exists. existingMap is unreliable here: backfill runs with 189 189 + // skipReplayDetection, leaving it empty, so a re-applied record would look new 190 190 + // and accumulate duplicate rows that fan out the search JOIN. The delete is 191 191 + // unconditional so it also evicts a stale row when an update clears all 192 192 + // searchable fields (content is null); only the re-insert is gated on content. 193 193 + stmts.push(db.prepare(`DELETE FROM ${table} WHERE uri = ?`).bind(event.uri)); 194 194 + 187 195 const content = buildFtsContent(record, fields); 188 188 - if (!content) return []; 189 189 - 190 190 - // Only delete existing FTS row if this is an update (record already existed) 191 191 - if (existingMap.has(event.uri)) { 192 192 - stmts.push(db.prepare(`DELETE FROM ${table} WHERE uri = ?`).bind(event.uri)); 196 196 + if (content) { 197 197 + stmts.push( 198 198 + db.prepare(`INSERT INTO ${table} (uri, content) VALUES (?, ?)`).bind(event.uri, content) 199 199 + ); 193 200 } 194 194 - stmts.push( 195 195 - db.prepare(`INSERT INTO ${table} (uri, content) VALUES (?, ?)`).bind(event.uri, content) 196 196 - ); 197 201 } 198 202 199 203 return stmts; ··· 644 648 if (!isReplay && !options?.skipFeedFanout) { 645 649 batch.push(...buildFeedStatements(db, e, config, existingRecordStrings)); 646 650 } 647 647 - batch.push(...buildFtsStatements(db, e, config, existingMap)); 651 651 + batch.push(...buildFtsStatements(db, e, config)); 648 652 } 649 653 } 650 654

+34

packages/contrail/tests/search.test.ts

Reviewed

··· 159 159 ], SEARCH_CONFIG); 160 160 expect((await queryRecords(db, SEARCH_CONFIG, { collection, search: "Deletable" })).records).toHaveLength(0); 161 161 }); 162 162 + 163 163 + it("does not duplicate FTS rows when the same record is re-applied during backfill", async () => { 164 164 + // Backfill passes call applyEvents with skipReplayDetection: true, which leaves 165 165 + // existingMap empty so every record looks brand-new. Re-backfilling the same 166 166 + // record must not append a second FTS row; otherwise the search JOIN fans out 167 167 + // and returns the event more than once (which crashes keyed lists downstream). 168 168 + const event = makeEvent({ 169 169 + uri: "at://did:plc:a/community.lexicon.calendar.event/1", 170 170 + collection, 171 171 + rkey: "1", 172 172 + record: { name: "Backfilled Meetup", mode: "online", description: "test" }, 173 173 + time_us: 1000, 174 174 + }); 175 175 + await applyEvents(db, [event], SEARCH_CONFIG, { skipReplayDetection: true }); 176 176 + await applyEvents(db, [event], SEARCH_CONFIG, { skipReplayDetection: true }); 177 177 + 178 178 + const result = await queryRecords(db, SEARCH_CONFIG, { collection, search: "Meetup" }); 179 179 + expect(result.records).toHaveLength(1); 180 180 + }); 181 181 + 182 182 + it("evicts the stale FTS row when an update clears all searchable fields", async () => { 183 183 + // The delete must run unconditionally. If an update leaves every searchable 184 184 + // field empty, buildFtsContent returns null and there is nothing to re-insert, 185 185 + // but the prior FTS row must still be removed so old terms stop matching. 186 186 + await applyEvents(db, [ 187 187 + makeEvent({ uri: "at://did:plc:a/community.lexicon.calendar.event/1", collection, rkey: "1", record: { name: "Searchable Title", mode: "online", description: "find me" }, time_us: 1000 }), 188 188 + ], SEARCH_CONFIG); 189 189 + expect((await queryRecords(db, SEARCH_CONFIG, { collection, search: "Searchable" })).records).toHaveLength(1); 190 190 + 191 191 + await applyEvents(db, [ 192 192 + makeEvent({ uri: "at://did:plc:a/community.lexicon.calendar.event/1", collection, rkey: "1", record: { startsAt: "2026-01-01T00:00:00Z" }, operation: "update", time_us: 2000 }), 193 193 + ], SEARCH_CONFIG); 194 194 + expect((await queryRecords(db, SEARCH_CONFIG, { collection, search: "Searchable" })).records).toHaveLength(0); 195 195 + }); 162 196 }); 163 197 164 198 describe.skipIf(!hasFts)("explicit searchable fields", () => {