Skip to content

Commit 08d3b4d

Browse files
committed
improvement(knowledge): finish Confluence passes past unreadable attachments, rank unauthenticated keyword search in one statement
- Confluence: a 403/404 on one parent's attachment listing is skipped with a warning instead of recorded as a listing failure, so the pass can finish and advance lastSyncAt. The page still marks the listing reconciliation-unsafe, so attachments it did not list are kept, never deleted. Missing scopes and server failures stay listing failures. - Connector counts: the member engine keeps the recorded document count on an unfinished slice once a pass has finished, the same gate the content engine uses. The count query and gate now live in sync-primitives. - Keyword search: callers with no person behind them rank through the same materialized candidate statement signed-in readers use, instead of ORDER BY ts_rank_cd over a per-chunk join. Same ACL predicate and global ordering, ties now broken by chunk id; still unbudgeted.
1 parent 3d0e791 commit 08d3b4d

7 files changed

Lines changed: 192 additions & 134 deletions

File tree

‎apps/sim/connectors/confluence/attachments.test.ts‎

Lines changed: 58 additions & 26 deletions
Original file line numberDiff line numberDiff line change
@@ -215,41 +215,73 @@ describe('Confluence attachment listing', () => {
215215
expect(last.hasMore).toBe(false)
216216
})
217217

218-
it.each([403, 401])(
219-
'preserves parent pages and cumulative partial progress for attachment access failure %s',
218+
it('preserves parent pages and cumulative partial progress when the attachment scope is missing', async () => {
219+
fetchMock.mockImplementation(async () =>
220+
Response.json({ code: 401, message: 'Unauthorized; scope does not match' }, { status: 401 })
221+
)
222+
const listParents = vi.fn(
223+
async (cursor: string | undefined): Promise<ExternalDocumentList> => ({
224+
documents: [parent(cursor ? 'p2' : 'p1')],
225+
hasMore: !cursor,
226+
nextCursor: cursor ? undefined : 'next',
227+
})
228+
)
229+
const first = await listConfluenceAttachments({ ...INPUT, listParents })
230+
const context: Record<string, unknown> = {}
231+
const last = await listConfluenceAttachments({
232+
...INPUT,
233+
listParents,
234+
cursor: first.nextCursor,
235+
syncContext: context,
236+
})
237+
expect(first.documents.map((doc) => doc.externalId)).toEqual(['p1'])
238+
expect(last.documents.map((doc) => doc.externalId)).toEqual(['p2'])
239+
expect(last.listingFailures).toEqual({
240+
count: 2,
241+
samples: ['p1', 'p2'].map((scope) => ({
242+
scope,
243+
operation: 'confluence.attachments.list',
244+
status: 401,
245+
reasons: ['attachment_scope_mismatch'],
246+
})),
247+
})
248+
expect(last.reconciliationSafe).toBe(false)
249+
expect(context.reconciliationUnsafe).toBe(true)
250+
})
251+
252+
/**
253+
* One page whose attachments the credential cannot open must not leave every pass incomplete,
254+
* which would keep the connector from ever finishing a sync. Its attachments were not listed,
255+
* so the page still withholds deletion of anything it did not list.
256+
*/
257+
it.each([403, 404])(
258+
'skips a parent whose attachments answer %s without failing the listing or deleting what it did not list',
220259
async (status) => {
221-
fetchMock.mockImplementation(async () =>
222-
status === 401
223-
? Response.json({ code: 401, message: 'Unauthorized; scope does not match' }, { status })
224-
: new Response(null, { status })
225-
)
260+
fetchMock.mockImplementation(async (input) => {
261+
const parentId = new URL(String(input)).pathname.split('/').at(-2) ?? ''
262+
if (parentId === 'p1') return new Response(null, { status })
263+
return Response.json({ results: [file({ id: `${parentId}-file`, pageId: parentId })] })
264+
})
226265
const listParents = vi.fn(
227-
async (cursor: string | undefined): Promise<ExternalDocumentList> => ({
228-
documents: [parent(cursor ? 'p2' : 'p1')],
229-
hasMore: !cursor,
230-
nextCursor: cursor ? undefined : 'next',
266+
async (): Promise<ExternalDocumentList> => ({
267+
documents: [parent('p1'), parent('p2')],
268+
hasMore: false,
231269
})
232270
)
233-
const first = await listConfluenceAttachments({ ...INPUT, listParents })
234271
const context: Record<string, unknown> = {}
235-
const last = await listConfluenceAttachments({
272+
const result = await listConfluenceAttachments({
236273
...INPUT,
237274
listParents,
238-
cursor: first.nextCursor,
239275
syncContext: context,
240276
})
241-
expect(first.documents.map((doc) => doc.externalId)).toEqual(['p1'])
242-
expect(last.documents.map((doc) => doc.externalId)).toEqual(['p2'])
243-
expect(last.listingFailures).toEqual({
244-
count: 2,
245-
samples: ['p1', 'p2'].map((scope) => ({
246-
scope,
247-
operation: 'confluence.attachments.list',
248-
status,
249-
reasons: [status === 401 ? 'attachment_scope_mismatch' : 'attachment_access_denied'],
250-
})),
251-
})
252-
expect(last.reconciliationSafe).toBe(false)
277+
expect(result.documents.map((doc) => doc.externalId)).toEqual([
278+
'p1',
279+
'p2',
280+
'attachment:page:p2:p2-file',
281+
])
282+
expect(result.hasMore).toBe(false)
283+
expect(result.listingFailures).toBeUndefined()
284+
expect(result.reconciliationSafe).toBe(false)
253285
expect(context.reconciliationUnsafe).toBe(true)
254286
}
255287
)

‎apps/sim/connectors/confluence/attachments.ts‎

Lines changed: 27 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -378,6 +378,7 @@ export async function listConfluenceAttachments(
378378
): Promise<ExternalDocumentList> {
379379
const state = decodeCursor(input.cursor, input.syncContext?.totalDocsFetched)
380380
const documents: ExternalDocument[] = []
381+
let skippedUnreadableParent = false
381382
if (state.parents.length === 0 && !state.parentsDone) {
382383
const parentContext = { ...input.syncContext, totalDocsFetched: state.parentsListed }
383384
const page = await input.listParents(state.parentCursor, parentContext)
@@ -413,15 +414,30 @@ export async function listConfluenceAttachments(
413414
})
414415
}
415416
if (state.serverFailureStreak >= MAX_CONSECUTIVE_SERVER_FAILURES) throw listing.error
416-
state.failures ??= { count: 0, samples: [] }
417-
state.failures.count += 1
418-
if (state.failures.samples.length < MAX_LISTING_FAILURE_SAMPLES) {
419-
state.failures.samples.push({
420-
scope: parent.id,
421-
operation: 'confluence.attachments.list',
417+
if (listing.reason === 'attachment_access_denied') {
418+
/**
419+
* The parent was listed a moment ago, so a 403 or 404 on its attachments is that one
420+
* item's restriction, not a scope the source failed to list: skipping it leaves the pass
421+
* complete. Its unlisted attachments are not evidence of deletion, so the page still
422+
* marks the listing unsafe and reconciliation keeps them.
423+
*/
424+
logger.warn('Skipping a Confluence parent whose attachments are not readable', {
425+
parentId: parent.id,
426+
parentType: parent.type,
422427
status: listing.status,
423-
reasons: [listing.reason],
424428
})
429+
skippedUnreadableParent = true
430+
} else {
431+
state.failures ??= { count: 0, samples: [] }
432+
state.failures.count += 1
433+
if (state.failures.samples.length < MAX_LISTING_FAILURE_SAMPLES) {
434+
state.failures.samples.push({
435+
scope: parent.id,
436+
operation: 'confluence.attachments.list',
437+
status: listing.status,
438+
reasons: [listing.reason],
439+
})
440+
}
425441
}
426442
state.parents.shift()
427443
state.attachmentCursor = undefined
@@ -453,7 +469,8 @@ export async function listConfluenceAttachments(
453469
state.attachmentCursor = nextCursor
454470
if (!nextCursor) state.parents.shift()
455471
}
456-
if (state.failures && input.syncContext) input.syncContext.reconciliationUnsafe = true
472+
const reconciliationUnsafe = Boolean(state.failures) || skippedUnreadableParent
473+
if (reconciliationUnsafe && input.syncContext) input.syncContext.reconciliationUnsafe = true
457474
const hasMore = state.parents.length > 0 || !state.parentsDone
458475
const nextCursor = hasMore ? CURSOR_PREFIX + JSON.stringify(state) : undefined
459476
if (nextCursor && Buffer.byteLength(nextCursor, 'utf8') > MAX_CURSOR_BYTES) {
@@ -463,7 +480,8 @@ export async function listConfluenceAttachments(
463480
documents,
464481
hasMore,
465482
nextCursor,
466-
...(state.failures ? { listingFailures: state.failures, reconciliationSafe: false } : {}),
483+
...(state.failures ? { listingFailures: state.failures } : {}),
484+
...(reconciliationUnsafe ? { reconciliationSafe: false } : {}),
467485
}
468486
}
469487

‎apps/sim/lib/knowledge/connectors/member-sync-engine-content-credential.test.ts‎

Lines changed: 43 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -185,6 +185,8 @@ function arrange(
185185
existingDocument?: boolean
186186
openMemberFeed?: boolean
187187
contentFresh?: boolean
188+
/** A content pass that finished before this run, past the content interval. */
189+
finishedPass?: { lastSyncAt: Date; lastSyncDocCount: number }
188190
forceContentRefresh?: boolean
189191
noDueMembers?: boolean
190192
syncIntervalMinutes?: number
@@ -211,7 +213,8 @@ function arrange(
211213
memberSyncStatus: 'idle',
212214
memberSyncConsecutiveFailures: 0,
213215
syncIntervalMinutes: options.syncIntervalMinutes ?? 1440,
214-
lastSyncAt: options.contentFresh ? new Date() : null,
216+
lastSyncAt: options.finishedPass?.lastSyncAt ?? (options.contentFresh ? new Date() : null),
217+
lastSyncDocCount: options.finishedPass?.lastSyncDocCount ?? null,
215218
directoryCheckpoint: options.directoryCheckpoint ?? null,
216219
sourceConfig: { folderId: 'shared-folder', adminEmail: 'admin@example.com' },
217220
accessRewritePending: false,
@@ -280,7 +283,8 @@ function arrange(
280283
},
281284
])
282285
}
283-
queueTableRows(schemaMock.document, [{ count: 1 }])
286+
if (!options.finishedPass || !options.contentIncomplete)
287+
queueTableRows(schemaMock.document, [{ count: 1 }])
284288
}
285289
if (options.members && !options.noDueMembers)
286290
queueTableRows(schemaMock.document, [
@@ -553,6 +557,43 @@ describe('member engine with a dedicated content credential', () => {
553557
expect(bounded).toEqual([])
554558
})
555559

560+
/**
561+
* A finished pass recorded the count, and an unfinished slice of the next pass keeps it rather
562+
* than scanning the whole connector again; the count is replaced when that pass finishes.
563+
*/
564+
it('keeps the recorded document count on an unfinished slice once a pass has finished', async () => {
565+
const run = arrange({
566+
contentIncomplete: true,
567+
finishedPass: { lastSyncAt: new Date(Date.now() - 2 * 86_400_000), lastSyncDocCount: 7 },
568+
})
569+
const now = Date.now()
570+
const clock = vi.spyOn(Date, 'now')
571+
mocks.list.mockResolvedValue({
572+
documents: Array.from({ length: 26 }, (_, i) => ({
573+
...serviceDocument,
574+
externalId: `content-${i}`,
575+
})),
576+
hasMore: false,
577+
})
578+
mocks.get.mockImplementation(async (_token, _config, externalId: string) => {
579+
clock.mockReturnValue(now + 35 * 60_000)
580+
return { ...serviceDocument, externalId, content: 'Service content', contentDeferred: false }
581+
})
582+
try {
583+
const result = await run()
584+
expect(result.error).toBeUndefined()
585+
expect(result.docsAdded).toBeGreaterThan(0)
586+
expect(dbChainMockFns.values).toHaveBeenCalledWith(
587+
expect.objectContaining({ status: 'partial', listedCount: null })
588+
)
589+
expect(
590+
dbChainMockFns.set.mock.calls.some(([value]) => value && 'lastSyncDocCount' in value)
591+
).toBe(false)
592+
} finally {
593+
clock.mockRestore()
594+
}
595+
})
596+
556597
it('reserves time for member permissions when a slow dedicated content page has more batches', async () => {
557598
const run = arrange({ members: true, contentIncomplete: true })
558599
const now = Date.now()

‎apps/sim/lib/knowledge/connectors/member-sync-engine.ts‎

Lines changed: 6 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -121,7 +121,9 @@ import {
121121
ConnectorSyncCapacityError,
122122
classifyListing,
123123
classifySuspectListing,
124+
countLiveConnectorDocuments,
124125
createSyncRunState,
126+
keepsRecordedDocCount,
125127
loadPageCorpus,
126128
processDocOps,
127129
RETRY_WINDOW_DAYS,
@@ -1711,17 +1713,9 @@ async function syncDedicatedMemberContent(input: {
17111713
run.result.listingIncomplete = true
17121714
const contentNotice = pass.holdNotice
17131715
/** Counted before the lease transaction: a scan of the whole connector never holds its row. */
1714-
const [{ count }] = await db
1715-
.select({ count: sql<number>`count(*)::int` })
1716-
.from(document)
1717-
.where(
1718-
and(
1719-
eq(document.connectorId, run.connectorId),
1720-
eq(document.userExcluded, false),
1721-
isNull(document.archivedAt),
1722-
isNull(document.deletedAt)
1723-
)
1724-
)
1716+
const count = keepsRecordedDocCount(pass.complete, connector)
1717+
? null
1718+
: await countLiveConnectorDocuments(run.connectorId)
17251719
await withMemberLease(run, async (tx) => {
17261720
const now = new Date()
17271721
await tx.insert(knowledgeConnectorSyncLog).values({
@@ -1746,7 +1740,7 @@ async function syncDedicatedMemberContent(input: {
17461740
? { lastSyncAt: new Date(pass.checkpoint.startedAt) }
17471741
: {}),
17481742
...(pass.complete ? { listingCheckpoint: null } : {}),
1749-
lastSyncDocCount: count,
1743+
...(count === null ? {} : { lastSyncDocCount: count }),
17501744
lastSyncError: contentNotice,
17511745
updatedAt: now,
17521746
})

‎apps/sim/lib/knowledge/connectors/sync-engine.ts‎

Lines changed: 13 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,8 @@ import {
8989
ConnectorDeletedException,
9090
ConnectorSyncCapacityError,
9191
checkSyncTargetPresence,
92+
countLiveConnectorDocuments,
93+
keepsRecordedDocCount,
9294
RETRY_WINDOW_DAYS,
9395
shouldRunIncrementalSync,
9496
sweepStuckDocuments,
@@ -368,22 +370,6 @@ export function isContentPassIncomplete(
368370
)
369371
}
370372

371-
/** Live documents the connector owns: what the connector list shows as its document count. */
372-
async function countLiveConnectorDocuments(connectorId: string): Promise<number> {
373-
const [row] = await db
374-
.select({ count: sql<number>`count(*)::int` })
375-
.from(document)
376-
.where(
377-
and(
378-
eq(document.connectorId, connectorId),
379-
eq(document.userExcluded, false),
380-
isNull(document.archivedAt),
381-
isNull(document.deletedAt)
382-
)
383-
)
384-
return row?.count ?? 0
385-
}
386-
387373
/**
388374
* Atomically publishes the completed log and connector terminal state.
389375
*
@@ -435,29 +421,19 @@ export async function completeSuccessfulSync(
435421
* owned count the next pass's mass-deletion guard starts from, so a tally of this run's
436422
* adds and deletes would drift; when it cannot be read the previous count stands, which the
437423
* guard already treats as a floor.
438-
*
439-
* Once a pass has finished, a slice that leaves the next pass unfinished keeps the count that
440-
* finished pass recorded, the same pass `lastSyncAt` still names, so a large connector is not
441-
* counted again on every slice of a pass whose count is replaced when it finishes. Unfinished
442-
* slices never advance `lastSyncAt`, so until a pass finishes every slice is counted and a
443-
* first pass shows its progress.
444424
*/
445-
const keepRecordedCount =
446-
contentPass !== undefined &&
447-
!contentPass.complete &&
448-
lockedConnector?.lastSyncAt != null &&
449-
lockedConnector.lastSyncDocCount !== null
450-
const actualDocCount = keepRecordedCount
451-
? null
452-
: await countLiveConnectorDocuments(connectorId).catch((error: unknown) => {
453-
const diagnostic = getConnectorFailureDiagnostic(error)
454-
if (diagnostic?.category !== 'database') throw error
455-
logger.warn('Could not count connector documents; keeping the previous count', {
456-
connectorId,
457-
diagnostic,
425+
const actualDocCount =
426+
contentPass !== undefined && keepsRecordedDocCount(contentPass.complete, lockedConnector)
427+
? null
428+
: await countLiveConnectorDocuments(connectorId).catch((error: unknown) => {
429+
const diagnostic = getConnectorFailureDiagnostic(error)
430+
if (diagnostic?.category !== 'database') throw error
431+
logger.warn('Could not count connector documents; keeping the previous count', {
432+
connectorId,
433+
diagnostic,
434+
})
435+
return null
458436
})
459-
return null
460-
})
461437
try {
462438
return await db.transaction(async (tx) => {
463439
const [lockedKnowledgeBase] = await tx

‎apps/sim/lib/knowledge/connectors/sync-primitives.ts‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -653,6 +653,38 @@ export function resolvePreviousOwnedCount(
653653
return Math.max(lastSyncDocCount ?? 0, ownedDocCount)
654654
}
655655

656+
/** Live documents the connector owns: what the connector list shows as its document count. */
657+
export async function countLiveConnectorDocuments(connectorId: string): Promise<number> {
658+
const [row] = await db
659+
.select({ count: sql<number>`count(*)::int` })
660+
.from(document)
661+
.where(
662+
and(
663+
eq(document.connectorId, connectorId),
664+
eq(document.userExcluded, false),
665+
isNull(document.archivedAt),
666+
isNull(document.deletedAt)
667+
)
668+
)
669+
return row?.count ?? 0
670+
}
671+
672+
/**
673+
* Whether a slice keeps the document count the connector holds instead of counting again.
674+
*
675+
* Once a pass has finished, a slice that leaves the next pass unfinished keeps the count that
676+
* finished pass recorded, the same pass `lastSyncAt` still names, so a large connector is not
677+
* counted again on every slice of a pass whose count is replaced when it finishes. Unfinished
678+
* slices never advance `lastSyncAt`, so until a pass finishes every slice is counted and a first
679+
* pass shows its progress.
680+
*/
681+
export function keepsRecordedDocCount(
682+
passComplete: boolean,
683+
connector: { lastSyncAt: Date | null; lastSyncDocCount: number | null } | null
684+
): boolean {
685+
return !passComplete && connector?.lastSyncAt != null && connector.lastSyncDocCount != null
686+
}
687+
656688
/**
657689
* Fraction of a connector's owned documents that a single reconciliation pass
658690
* may remove before the pass is held.

0 commit comments

Comments
 (0)