Skip to content

Commit 916a355

Browse files
authored
fix(knowledge): bound the members-mode tombstone pass by what a run changed (#8172)
* fix(knowledge): bound the members-mode tombstone pass by what a run changed The tombstone phase of the member document lifecycle ran one anti-join over every live document of the connector on every run, and LIMIT could not bound it because almost every document is observed. On a large connector that statement outgrew the statement timeout, rolled back the member's page, and walked the connector toward disabled. Tombstoning is now driven by the documents whose observations the run itself removed (unseen after a complete listing, withdrawn by a change feed, or held by a member being removed), then by a resumable backstop that walks the reconciliation index one page per statement and checks observations only in the UPDATE over that page's live ids. The backstop saves its position on the connector and does a bounded number of pages per run. Resurrection and purge are unchanged. * fix(knowledge): walk the members-mode reconcile by external id and tombstone explicit removals without a completed listing The absence reconcile ordered its resumable walk by source_seen_at, which every member listing rewrites for what it observed. On a connector larger than one run's page budget the walk chased re-stamped documents and a pass never ended, so a document that lost its observers behind the cursor was never revisited. The walk now keys on external id through doc_connector_external_id_idx, which a document never changes, so a pass completes within ceil(documents / budget) runs. Documents a run explicitly unobserved (a complete listing, a change-feed withdrawal, or a member removal) are now tombstoned even when no member with a completed listing remains, as the stale-member sweep already does. Removing the only listed member previously left every document only it observed live and unobserved indefinitely. The reconcile and the purge stay gated on a completed listing. * fix(knowledge): keep member removal ids across interrupted runs and hide the reconcile cursor - A member removal interrupted by the run deadline resumed from its saved document cursor and only handed the lifecycle the documents above it. The resumed walk now re-reads the removed member's observations below the cursor first. They still exist until the member row is deleted. - The absence reconcile writes its cursor once per run instead of once per page, and documents how many runs a pass takes. - Connector presenters drop member_tombstone_cursor alongside the stored API key, so the internal cursor never reaches API responses. - CI runs member-document-lifecycle.integration.ts with the other knowledge PostgreSQL suites. * chore(db): drop the tombstone cursor migration to regenerate it after staging * chore(db): regenerate the tombstone cursor migration on top of staging * chore(db): format the tombstone cursor migration metadata * fix(knowledge): tombstone a removed member's sole documents with the removal itself A membership removal handed the documents its member alone observed to the lifecycle through run memory. If the run stopped after the member row was deleted, the ids were gone, the observations had cascaded away, and with no member left with a completed listing the absence reconcile never ran. Those documents stayed live indefinitely. Each removal page now tombstones the documents no other member observes in the same transaction that advances the removal's checkpoint, so nothing depends on the run surviving to its lifecycle. This applies only where observations decide existence; a corpus owned by a dedicated content credential is left alone, as before. The in-memory re-collection added earlier is removed. The absence reconcile's external_id IS NOT NULL filter is documented. Every writer that sets connector_id copies a required external id, so the filter excludes nothing. * fix(knowledge): resurrect a withdrawn removal's tombstones with the member's restored ACLs A multi-run member removal tombstones, page by page, the documents only that member observes. If directory re-listing withdrew the removal mid-walk, the restarted walk restored the member's token to those ACLs, but the documents stayed deleted until the lifecycle's resurrection phase ran. That phase runs only after the member loop finishes before the deadline. A membership walk that is not a removal now applies the lifecycle's own resurrect predicate to each page, in the same transaction. The predicate is shared through one helper and covers: connector-owned, not excluded, not archived, content present, observed. Only where observations decide existence. * test(knowledge): insert the content-less fixture row with its own typed insert
1 parent 884c602 commit 916a355

16 files changed

Lines changed: 29358 additions & 110 deletions

‎.github/workflows/test-build.yml‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -274,6 +274,7 @@ jobs:
274274
lib/knowledge/__integration__/user-document-visibility.integration.ts
275275
lib/knowledge/__integration__/listing-continuation.integration.ts
276276
lib/knowledge/__integration__/member-scope-renewal.integration.ts
277+
lib/knowledge/__integration__/member-document-lifecycle.integration.ts
277278
lib/knowledge/__integration__/slack-empty-threads.integration.ts
278279
lib/knowledge/__integration__/kb-block-search.integration.ts
279280
lib/knowledge/__integration__/unfilled-projection-source.integration.ts

‎apps/sim/lib/knowledge/__integration__/member-document-lifecycle.integration.ts‎

Lines changed: 264 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,13 +3,14 @@ import { db } from '@sim/db'
33
import {
44
document,
55
knowledgeConnector,
6+
knowledgeConnectorMember,
67
knowledgeDocumentObservation,
78
organization,
89
user,
910
workspace,
1011
} from '@sim/db/schema'
1112
import { generateId } from '@sim/utils/id'
12-
import { eq, inArray, sql } from 'drizzle-orm'
13+
import { and, eq, inArray, isNotNull, sql } from 'drizzle-orm'
1314
import { afterAll, afterEach, beforeEach, describe, expect, it } from 'vitest'
1415
import {
1516
type createKnowledgeAclFixtureIds,
@@ -19,9 +20,13 @@ import {
1920
import {
2021
applyMemberDocumentLifecycle,
2122
recordMemberObservations,
23+
removeMemberObservationsForDocuments,
2224
} from '@/lib/knowledge/connectors/member-observations'
25+
import { resumeMembershipRewrites } from '@/lib/knowledge/connectors/member-sync-engine'
26+
import { MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN } from '@/lib/knowledge/connectors/sync-limits'
2327
import {
2428
assertSyncLeaseHeldInTx,
29+
createMemberSyncLease,
2530
SyncLockLostException,
2631
stillHoldsMemberSyncLock,
2732
} from '@/lib/knowledge/connectors/sync-lock'
@@ -62,12 +67,19 @@ describe('member document lifecycle in PostgreSQL', () => {
6267
const observe = (documentIds: string[]) =>
6368
recordMemberObservations(db, members.members[0].id, documentIds, members.runId)
6469

65-
const run = (options: { beforeWrite?: () => Promise<void>; allowRemoval?: boolean } = {}) =>
70+
const run = (
71+
options: {
72+
beforeWrite?: () => Promise<void>
73+
allowRemoval?: boolean
74+
unobservedDocumentIds?: string[]
75+
} = {}
76+
) =>
6677
applyMemberDocumentLifecycle({
6778
connectorId: members.connectorId,
6879
knowledgeBaseId: ids.knowledgeBaseId,
6980
runId: members.runId,
7081
allowRemoval: options.allowRemoval ?? true,
82+
unobservedDocumentIds: options.unobservedDocumentIds ?? [],
7183
deadlineAt: Date.now() + 60_000,
7284
lease: { beatIfDue: async () => {} },
7385
withLease: async (fn) => {
@@ -108,6 +120,256 @@ describe('member document lifecycle in PostgreSQL', () => {
108120
expect(byId.get(noContent.id)).toEqual(deletedAt)
109121
})
110122

123+
const insertRows = async (rows: ReturnType<typeof row>[]) => {
124+
for (let offset = 0; offset < rows.length; offset += 500)
125+
await db.insert(document).values(rows.slice(offset, offset + 500))
126+
}
127+
const tombstonedIds = async () =>
128+
new Set(
129+
(
130+
await db
131+
.select({ id: document.id })
132+
.from(document)
133+
.where(and(eq(document.connectorId, members.connectorId), isNotNull(document.deletedAt)))
134+
).map(({ id }) => id)
135+
)
136+
const savedCursor = async () =>
137+
(
138+
await db
139+
.select({ cursor: knowledgeConnector.memberTombstoneCursor })
140+
.from(knowledgeConnector)
141+
.where(eq(knowledgeConnector.id, members.connectorId))
142+
)[0].cursor
143+
144+
it('tombstones what this run unobserved right away and leaves the rest of a large connector to later runs', async () => {
145+
const pageBudget = MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN * 500
146+
const unobserved = Array.from({ length: pageBudget + 20 }, (_, index) =>
147+
row(`unobserved-${index}`)
148+
)
149+
/** Both sort after every unobserved document, beyond what this run's reconcile reaches. */
150+
const [lostByThisRun, stillObservedByBob] = [
151+
row('zz-lost-by-this-run'),
152+
row('zz-still-observed-by-bob'),
153+
]
154+
await insertRows([...unobserved, lostByThisRun, stillObservedByBob])
155+
await observe([lostByThisRun.id, stillObservedByBob.id])
156+
await recordMemberObservations(
157+
db,
158+
members.members[1].id,
159+
[stillObservedByBob.id],
160+
members.runId
161+
)
162+
const removed = await removeMemberObservationsForDocuments(db, members.members[0].id, [
163+
lostByThisRun.id,
164+
stillObservedByBob.id,
165+
])
166+
expect(removed.sort()).toEqual([lostByThisRun.id, stillObservedByBob.id].sort())
167+
168+
expect(await run({ unobservedDocumentIds: removed })).toEqual({
169+
tombstoned: pageBudget + 1,
170+
resurrected: 0,
171+
purged: 0,
172+
finished: true,
173+
})
174+
const afterFirst = await tombstonedIds()
175+
expect(afterFirst.has(lostByThisRun.id)).toBe(true)
176+
expect(afterFirst.has(stillObservedByBob.id)).toBe(false)
177+
expect(unobserved.filter(({ id }) => !afterFirst.has(id))).toHaveLength(20)
178+
expect(await savedCursor()).toEqual({ externalId: expect.any(String) })
179+
180+
expect(await run()).toEqual({ tombstoned: 20, resurrected: 0, purged: 0, finished: true })
181+
const afterSecond = await tombstonedIds()
182+
expect(unobserved.every(({ id }) => afterSecond.has(id))).toBe(true)
183+
expect(afterSecond.has(stillObservedByBob.id)).toBe(false)
184+
expect(await savedCursor()).toBeNull()
185+
})
186+
187+
it('tombstones what removing the only listed member unobserved, though the run stops right after the removal', async () => {
188+
const [removedMember, otherMember] = members.members
189+
await db
190+
.update(knowledgeConnectorMember)
191+
.set({
192+
lastCompleteListingAt: new Date(),
193+
listingCheckpoint: { kind: 'membership', cursor: null, removeMember: true },
194+
})
195+
.where(eq(knowledgeConnectorMember.id, removedMember.id))
196+
const onlyRemoved = Array.from({ length: 1_200 }, (_, index) => row(`only-removed-${index}`))
197+
const sharedWithOther = row('shared-with-other-member')
198+
const neverObserved = row('never-observed')
199+
await insertRows([...onlyRemoved, sharedWithOther, neverObserved])
200+
await observe([...onlyRemoved.map(({ id }) => id), sharedWithOther.id])
201+
await recordMemberObservations(db, otherMember.id, [sharedWithOther.id], members.runId)
202+
const removal = (stopAfterFirstPage: boolean) => {
203+
const lease = createMemberSyncLease(members.connectorId, members.runId)
204+
const input: Parameters<typeof resumeMembershipRewrites>[0] = {
205+
connectorId: members.connectorId,
206+
runId: members.runId,
207+
deadlineAt: Date.now() + 60_000,
208+
tombstonesUnobserved: true,
209+
lease: {
210+
...lease,
211+
beatIfDue: async () => {
212+
await lease.beatIfDue()
213+
if (stopAfterFirstPage) input.deadlineAt = Date.now() - 1
214+
},
215+
},
216+
}
217+
return resumeMembershipRewrites(input)
218+
}
219+
220+
/** The first run walks one page before its deadline; the second finishes and deletes the member. */
221+
expect(await removal(true)).toBe(false)
222+
const [paused] = await db
223+
.select({ checkpoint: knowledgeConnectorMember.listingCheckpoint })
224+
.from(knowledgeConnectorMember)
225+
.where(eq(knowledgeConnectorMember.id, removedMember.id))
226+
expect(paused.checkpoint).toMatchObject({ removeMember: true, cursor: expect.any(String) })
227+
expect(await removal(false)).toBe(true)
228+
const [completed] = await db
229+
.select({ count: sql<number>`count(*)::int` })
230+
.from(knowledgeConnectorMember)
231+
.where(
232+
and(
233+
eq(knowledgeConnectorMember.connectorId, members.connectorId),
234+
isNotNull(knowledgeConnectorMember.lastCompleteListingAt)
235+
)
236+
)
237+
expect(completed.count).toBe(0)
238+
/** No lifecycle ran in either run: the tombstones landed with the removal itself. */
239+
const afterRemoval = await tombstonedIds()
240+
expect(onlyRemoved.every(({ id }) => afterRemoval.has(id))).toBe(true)
241+
expect(afterRemoval.has(sharedWithOther.id)).toBe(false)
242+
expect(afterRemoval.has(neverObserved.id)).toBe(false)
243+
244+
expect(await run({ allowRemoval: false })).toEqual({
245+
tombstoned: 0,
246+
resurrected: 0,
247+
purged: 0,
248+
finished: true,
249+
})
250+
expect(await tombstonedIds()).toEqual(afterRemoval)
251+
})
252+
253+
it('brings back what a withdrawn removal tombstoned within the same run, and nothing else', async () => {
254+
const [member] = members.members
255+
await db
256+
.update(knowledgeConnectorMember)
257+
.set({ listingCheckpoint: { kind: 'membership', cursor: null, removeMember: true } })
258+
.where(eq(knowledgeConnectorMember.id, member.id))
259+
const walked = Array.from({ length: 600 }, (_, index) => row(`walked-${index}`))
260+
const excluded = { ...row('excluded'), userExcluded: true, deletedAt }
261+
const archived = { ...row('archived'), archivedAt: new Date(), deletedAt }
262+
const noContent = { ...row('no-content'), contentHash: null, deletedAt }
263+
await insertRows([...walked, excluded, archived])
264+
await db.insert(document).values(noContent)
265+
await observe([...walked, excluded, archived, noContent].map(({ id }) => id))
266+
const walk = (stopAfterFirstPage: boolean) => {
267+
const lease = createMemberSyncLease(members.connectorId, members.runId)
268+
const input: Parameters<typeof resumeMembershipRewrites>[0] = {
269+
connectorId: members.connectorId,
270+
runId: members.runId,
271+
deadlineAt: Date.now() + 60_000,
272+
tombstonesUnobserved: true,
273+
lease: {
274+
...lease,
275+
beatIfDue: async () => {
276+
await lease.beatIfDue()
277+
if (stopAfterFirstPage) input.deadlineAt = Date.now() - 1
278+
},
279+
},
280+
}
281+
return resumeMembershipRewrites(input)
282+
}
283+
284+
expect(await walk(true)).toBe(false)
285+
const tombstonedByRemoval = await tombstonedIds()
286+
const walkedIds = new Set(walked.map(({ id }) => id))
287+
const removedPage = [...tombstonedByRemoval].filter((id) => walkedIds.has(id))
288+
expect(removedPage.length).toBeGreaterThan(0)
289+
expect(removedPage.length).toBeLessThan(walked.length)
290+
291+
/** Directory re-listing withdraws the removal: the member is active again and its walk restarts. */
292+
await db
293+
.update(knowledgeConnectorMember)
294+
.set({
295+
status: 'active',
296+
listingCheckpoint: { kind: 'membership', cursor: null, removeMember: false },
297+
})
298+
.where(eq(knowledgeConnectorMember.id, member.id))
299+
expect(await walk(false)).toBe(true)
300+
301+
const after = await tombstonedIds()
302+
expect(walked.every(({ id }) => !after.has(id))).toBe(true)
303+
for (const kept of [excluded, archived, noContent]) expect(after.has(kept.id)).toBe(true)
304+
})
305+
306+
it('leaves a service-owned corpus alone when a member is removed', async () => {
307+
const [removedMember] = members.members
308+
await db
309+
.update(knowledgeConnectorMember)
310+
.set({ listingCheckpoint: { kind: 'membership', cursor: null, removeMember: true } })
311+
.where(eq(knowledgeConnectorMember.id, removedMember.id))
312+
const onlyRemoved = row('service-owned')
313+
await insertRows([onlyRemoved])
314+
await observe([onlyRemoved.id])
315+
expect(
316+
await resumeMembershipRewrites({
317+
connectorId: members.connectorId,
318+
runId: members.runId,
319+
deadlineAt: Date.now() + 60_000,
320+
lease: createMemberSyncLease(members.connectorId, members.runId),
321+
tombstonesUnobserved: false,
322+
})
323+
).toBe(true)
324+
expect((await tombstonedIds()).size).toBe(0)
325+
})
326+
327+
it('finishes a pass within its page budget while listings re-stamp every observed document', async () => {
328+
const pageBudget = MEMBER_TOMBSTONE_RECONCILE_PAGES_PER_RUN * 500
329+
const total = pageBudget + 500
330+
const runsPerPass = Math.ceil(total / pageBudget)
331+
const firstInEveryOrder = {
332+
...row('walk-0000000'),
333+
id: '00000000-0000-4000-8000-000000000000',
334+
}
335+
const rest = Array.from({ length: total - 1 }, (_, index) =>
336+
row(`walk-${String(index + 1).padStart(7, '0')}`)
337+
)
338+
await insertRows([firstInEveryOrder, ...rest])
339+
const all = [firstInEveryOrder, ...rest].map(({ id }) => id)
340+
for (let offset = 0; offset < all.length; offset += 5000)
341+
await observe(all.slice(offset, offset + 5000))
342+
/** A listing stamps what it saw with its start; the document nobody observes keeps its old stamp. */
343+
const relist = () =>
344+
db
345+
.update(document)
346+
.set({ sourceSeenAt: new Date() })
347+
.where(
348+
and(
349+
eq(document.connectorId, members.connectorId),
350+
sql`EXISTS (SELECT 1 FROM knowledge_document_observation o WHERE o.document_id = ${document.id})`
351+
)
352+
)
353+
354+
expect(await run()).toMatchObject({ tombstoned: 0 })
355+
expect(await savedCursor()).not.toBeNull()
356+
await db
357+
.delete(knowledgeDocumentObservation)
358+
.where(eq(knowledgeDocumentObservation.documentId, firstInEveryOrder.id))
359+
for (let pass = 1; pass < runsPerPass; pass++) {
360+
await relist()
361+
await run()
362+
}
363+
expect(await savedCursor()).toBeNull()
364+
expect((await tombstonedIds()).has(firstInEveryOrder.id)).toBe(false)
365+
366+
for (let next = 0; next < runsPerPass; next++) {
367+
await relist()
368+
await run()
369+
}
370+
expect(await tombstonedIds()).toEqual(new Set([firstInEveryOrder.id]))
371+
})
372+
111373
it('continues past a full selected batch even if its observations change before UPDATE', async () => {
112374
const rows = Array.from({ length: 501 }, (_, index) => row(String(index)))
113375
await db.insert(document).values(rows)

‎apps/sim/lib/knowledge/application/connectors.ts‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,7 @@ import {
9898
performSyncKnowledgeConnector,
9999
performUpdateKnowledgeConnector,
100100
type SourceConfigRejection,
101+
withoutSecret,
101102
} from '@/lib/knowledge/orchestration/connectors'
102103
import type {
103104
KnowledgeOperationSource,
@@ -548,11 +549,14 @@ export const listKnowledgeConnectors = defineAuthorizedKnowledgeUseCase({
548549
})
549550
: new Map<string, ViewerConnectorMembership>()
550551
return {
551-
connectors: page.map(({ encryptedApiKey: _encryptedApiKey, ...rest }) => ({
552-
...rest,
553-
permissionConfig: permissionSummaries.get(rest.id),
554-
viewerMembership: memberships.get(rest.id) ?? null,
555-
})),
552+
connectors: page.map((row) => {
553+
const rest = withoutSecret(row)
554+
return {
555+
...rest,
556+
permissionConfig: permissionSummaries.get(rest.id),
557+
viewerMembership: memberships.get(rest.id) ?? null,
558+
}
559+
}),
556560
hasMore,
557561
offset,
558562
limit: input.limit ?? page.length,
@@ -709,7 +713,7 @@ export const readKnowledgeConnector = defineAuthorizedKnowledgeUseCase({
709713
? summarizeConnectorMembers(context.connectorId, connector.syncIntervalMinutes)
710714
: { active: 0, suspended: 0, stale: 0 },
711715
])
712-
const { encryptedApiKey: _encryptedApiKey, ...connectorData } = connector
716+
const connectorData = withoutSecret(connector)
713717
const viewerUserId = principal.kind === 'session' ? principal.userId : null
714718
const memberships =
715719
viewerUserId && (context.workspaceId || context.organizationId)

0 commit comments

Comments
 (0)