From 1839da46b92a1ca71d1b023c75ba174b22123ed6 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 16 Sep 2026 23:22:41 -0700 Subject: [PATCH 1/2] fix(knowledge): bound permission pagination and member cleanup --- apps/docs/content/docs/search/confluence.mdx | 2 +- .../[connectorId]/source-detail.test.tsx | 16 ++ .../sources/[connectorId]/source-detail.tsx | 25 ++- .../connectors/confluence/permissions.test.ts | 140 ++++++++++++- apps/sim/connectors/confluence/permissions.ts | 147 +++++++++----- .../member-document-lifecycle.integration.ts | 184 ++++++++++++++++++ .../connectors/member-observations.test.ts | 5 + .../connectors/member-observations.ts | 53 +++-- 8 files changed, 497 insertions(+), 75 deletions(-) create mode 100644 apps/sim/lib/knowledge/__integration__/member-document-lifecycle.integration.ts diff --git a/apps/docs/content/docs/search/confluence.mdx b/apps/docs/content/docs/search/confluence.mdx index ad8475e2b56..d07af6163bf 100644 --- a/apps/docs/content/docs/search/confluence.mdx +++ b/apps/docs/content/docs/search/confluence.mdx @@ -142,7 +142,7 @@ In **Sync history**, **Continuing** means a healthy listing needs another batch. | Space picker is empty or fails | Check the domain, account's space access, and `read:space:confluence` scope. Manual space keys are also supported. | | Service-account validation fails | Check token expiry, site, Confluence app access, and the full scope list above, including `read:confluence-user`. | | Content syncs but Search is empty | Connect your personal Confluence identity. Check permission/directory sync errors and group-read scopes. | -| **Some permissions could not be verified** | Open the source's **Sync history**. Check the service account's space, page, and directory access. If access is correct and the warning persists, ask your operator to inspect the connector run's permission errors. Do not broaden sharing to clear the warning. | +| **Permission verification incomplete** | Open the source's **Sync history**. Check the service account's space, page, and directory access. If access is correct and the warning persists, ask your operator to inspect the connector run's permission errors, including incomplete or repeated permission pages. Documents without verified access stay hidden; do not broaden sharing to clear the warning. | | A new page, blog post, or label is missing | Confluence search can take time to update. Once the content appears in Confluence search with the selected label, sync again. | | A restricted page is missing | Both your account and the crawling account need access to the page and its ancestors. | | Embedded content is missing | Index the referenced page separately; remote macro output is excluded. | diff --git a/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.test.tsx b/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.test.tsx index d277c6a025d..51449023283 100644 --- a/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.test.tsx +++ b/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.test.tsx @@ -5,6 +5,7 @@ import { createRoot, type Root } from 'react-dom/client' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { ApiClientError } from '@/lib/api/client/errors' import type { ConnectorData } from '@/lib/api/contracts/knowledge/connectors' +import { SOURCE_PERMISSION_ERROR } from '@/lib/knowledge/connectors/sync-limits' import type { ConnectorActionsOptions } from '@/app/workspace/[workspaceId]/knowledge/[id]/components/connectors-section/use-connector-actions' const mocks = vi.hoisted(() => ({ @@ -257,6 +258,21 @@ describe('organization source detail navigation', () => { ) }) + it.each(['active', 'pending', 'syncing'] as const)( + 'keeps the safe permission warning visible while a source is %s', + async (status) => { + mocks.detail.mockReturnValue({ + data: { ...connector, accessMode: 'admin', status, lastSyncError: SOURCE_PERMISSION_ERROR }, + }) + await render() + expect(container.textContent).toContain('Permission verification incomplete') + expect(container.textContent).toContain(SOURCE_PERMISSION_ERROR) + expect(container.textContent).not.toContain( + 'Review the connection settings and try syncing again.' + ) + } + ) + it.each(['', '?view=settings', '?view=history'])( 'shows integration deactivation independently of source sync state at %s', async (searchParams) => { diff --git a/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.tsx b/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.tsx index df92230fa62..37efe1a9bae 100644 --- a/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.tsx +++ b/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.tsx @@ -12,6 +12,7 @@ import { useSettingsUnsavedGuard } from '@/components/settings/use-settings-unsa import { isApiClientError } from '@/lib/api/client/errors' import type { ConnectorData, ConnectorDetailData } from '@/lib/api/contracts/knowledge/connectors' import type { ResourceScope } from '@/lib/core/resource-scope' +import { SOURCE_PERMISSION_ERROR } from '@/lib/knowledge/connectors/sync-limits' import { organizationRoutes } from '@/lib/navigation/paths' import { describeSearchSource } from '@/lib/sim-search/source-identity' import { SEARCH_DEBOUNCE_MS } from '@/lib/url-state' @@ -186,6 +187,7 @@ function SourceDetailContent({ ? describeSearchSource(meta, connector.sourceConfig) || meta.name : 'Connection' const { effectiveStatus, lastSyncError } = getConnectorSyncState(connector) + const permissionsIncomplete = lastSyncError === SOURCE_PERMISSION_ERROR const status = effectiveStatus === 'paused' ? 'Sync paused' @@ -271,12 +273,23 @@ function SourceDetailContent({ > {integrationFeedback} - {effectiveStatus === 'active' && lastSyncError && ( - - )} + {lastSyncError && + (effectiveStatus === 'active' || + (permissionsIncomplete && + (effectiveStatus === 'pending' || effectiveStatus === 'syncing'))) && ( + + )} { 'rejects a repeated or missing cursor without publishing partial permissions: %s', async (next) => { mockFetch - .mockResolvedValueOnce(jsonResponse({ results: [], _links: { next: '?cursor=next' } })) - .mockResolvedValueOnce(jsonResponse({ results: [], _links: { next } })) + .mockResolvedValueOnce( + jsonResponse({ results: [{ id: 'first' }], _links: { next: '?cursor=next' } }) + ) + .mockResolvedValueOnce(jsonResponse({ results: [{ id: 'second' }], _links: { next } })) await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space-1')).rejects.toThrow( - 'invalid or repeated space permissions continuation' + 'invalid or repeated permission continuation' ) expect(mockFetch).toHaveBeenCalledTimes(2) } ) it('rejects a cursor cycle rather than making a hundred repeated requests', async () => { - for (const cursor of ['first', 'second', 'first']) { + for (const [page, cursor] of ['first', 'second', 'first'].entries()) { mockFetch.mockResolvedValueOnce( - jsonResponse({ results: [], _links: { next: `?cursor=${cursor}` } }) + jsonResponse({ results: [{ id: String(page) }], _links: { next: `?cursor=${cursor}` } }) ) } await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space-1')).rejects.toThrow('repeated') @@ -315,7 +318,7 @@ describe('listSpaceReadPrincipals', () => { it('rejects a malformed collection instead of treating it as a verified empty grant', async () => { mockFetch.mockResolvedValueOnce(jsonResponse({})) await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space-1')).rejects.toThrow( - 'invalid space permissions' + 'invalid permission page' ) }) @@ -325,6 +328,7 @@ describe('listSpaceReadPrincipals', () => { jsonResponse({ results: [ { + id: String(page), principal: { type: 'user', id: 'reader' }, operation: { key: 'read', targetType: 'space' }, }, @@ -333,9 +337,129 @@ describe('listSpaceReadPrincipals', () => { }) ) await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space-1')).rejects.toThrow( - 'exceeded 100 pages (100 entries)' + 'exceeded 1000 pages (1000 entries)' ) - expect(mockFetch).toHaveBeenCalledTimes(100) + expect(mockFetch).toHaveBeenCalledTimes(1000) + }) + + it('reads past 25,000 assignments and only publishes readers after the final page', async () => { + let page = 0 + mockFetch.mockImplementation(async () => { + const current = page++ + return jsonResponse({ + results: + current < 100 + ? Array.from({ length: 250 }, (_, index) => ({ + id: `${current}-${index}`, + principal: { type: 'user', id: 'editor' }, + operation: { key: 'create', targetType: 'page' }, + })) + : [ + { + id: 'last', + principal: { type: 'group', id: 'readers' }, + operation: { key: 'read', targetType: 'space' }, + }, + ], + _links: current < 100 ? { next: `?cursor=${page}` } : {}, + }) + }) + await expect(listSpaceReadPrincipals(CLOUD, 'token', 'large-space')).resolves.toEqual([ + { kind: 'group', id: 'readers' }, + ]) + expect(mockFetch).toHaveBeenCalledTimes(101) + }) + + it('rejects repeated assignments even when cursors and record order change', async () => { + const reader = { + id: 'one', + principal: { type: 'user', id: 'reader' }, + operation: { key: 'read', targetType: 'space' }, + } + mockFetch + .mockResolvedValueOnce( + jsonResponse({ results: [reader, { id: 'two' }], _links: { next: '?cursor=one' } }) + ) + .mockResolvedValueOnce( + jsonResponse({ results: [{ id: 'two' }, reader], _links: { next: '?cursor=two' } }) + ) + await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space')).rejects.toThrow( + 'repeated a permission page' + ) + expect(mockFetch).toHaveBeenCalledTimes(2) + }) + + it('follows an empty permission page that has a continuation', async () => { + mockFetch.mockResolvedValueOnce(jsonResponse({ results: [], _links: { next: '?cursor=next' } })) + mockFetch.mockResolvedValueOnce( + jsonResponse({ + results: [ + { + principal: { type: 'group', id: 'readers' }, + operation: { key: 'read', targetType: 'space' }, + }, + ], + }) + ) + await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space')).resolves.toEqual([ + { kind: 'group', id: 'readers' }, + ]) + expect(mockFetch).toHaveBeenCalledTimes(2) + }) + + it('bounds retained readers without truncating a large grant', async () => { + let page = 0 + mockFetch.mockImplementation(async () => + jsonResponse({ + results: Array.from({ length: 250 }, (_, index) => ({ + principal: { type: 'user', id: `reader-${page}-${index}` }, + operation: { key: 'read', targetType: 'space' }, + })), + _links: { next: `?cursor=${++page}` }, + }) + ) + await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space')).rejects.toThrow( + 'document permission limit' + ) + expect(mockFetch).toHaveBeenCalledTimes(Math.floor(MAX_ACL_TOKENS / 250) + 1) + }) + + it('rejects an oversized permission response before accepting its readers', async () => { + mockFetch.mockResolvedValueOnce(jsonResponse({ results: [{ id: 'x'.repeat(1024 * 1024) }] })) + await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space')).rejects.toThrow( + 'exceeds maximum size' + ) + }) + + it('does not return a partial reader list when a later permission page fails', async () => { + mockFetch + .mockResolvedValueOnce( + jsonResponse({ + results: [ + { + principal: { type: 'user', id: 'reader' }, + operation: { key: 'read', targetType: 'space' }, + }, + ], + _links: { next: '?cursor=next' }, + }) + ) + .mockResolvedValueOnce(jsonResponse({}, 403)) + await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space')).rejects.toThrow('403') + expect(mockFetch).toHaveBeenCalledTimes(2) + }) + + it('refuses an oversized continuation before issuing another request', async () => { + mockFetch.mockResolvedValueOnce( + jsonResponse({ + results: [], + _links: { next: `?cursor=${'x'.repeat(8193)}` }, + }) + ) + await expect(listSpaceReadPrincipals(CLOUD, 'token', 'space')).rejects.toThrow( + 'invalid or repeated permission continuation' + ) + expect(mockFetch).toHaveBeenCalledTimes(1) }) }) diff --git a/apps/sim/connectors/confluence/permissions.ts b/apps/sim/connectors/confluence/permissions.ts index 89f98371684..5e8e8f1470e 100644 --- a/apps/sim/connectors/confluence/permissions.ts +++ b/apps/sim/connectors/confluence/permissions.ts @@ -1,10 +1,13 @@ +import { createHash } from 'node:crypto' import { createLogger } from '@sim/logger' +import { sortObjectKeysDeep } from '@sim/utils/object' import { readResponseJsonWithLimit } from '@/lib/core/utils/stream-limits' import { type ConfluencePrincipal, type ConfluenceRestriction, confluenceSubjectToken, } from '@/lib/knowledge/access/confluence-permissions' +import { MAX_ACL_TOKENS } from '@/lib/knowledge/access/tokens' import { fetchWithRetry } from '@/lib/knowledge/documents/secure-fetch.server' import type { RetryOptions } from '@/lib/knowledge/documents/utils' import { extractCursor } from '@/connectors/confluence/cursor' @@ -21,6 +24,10 @@ const GROUP_PAGE_SIZE = 200 /** Bounds provider pagination, including malformed continuation responses. */ const MAX_PAGES = 100 +const MAX_PERMISSION_PAGES = 1000 +const PERMISSION_RESPONSE_MAX_BYTES = 1024 * 1024 +const MAX_PERMISSION_CURSOR_LENGTH = 8192 +const PERMISSION_COLLECTION_TIMEOUT_MS = 5 * 60 * 1000 const PREFLIGHT_RESPONSE_MAX_BYTES = 256 * 1024 function apiBase(cloudId: string): string { @@ -72,34 +79,77 @@ async function getJson( } /** - * Drains a v2 collection by following `_links.next`, the only termination - * Confluence documents. The requested page size is a ceiling the server may - * lower, so a page shorter than it proves nothing. + * Reduces permission pages without retaining unrelated assignments. Completion + * requires EOF; distinct cursors cannot conceal a repeated page of assignments. */ -async function drainV2(url: string, accessToken: string, what: string): Promise { - const items: T[] = [] +async function visitPermissionPages( + url: string, + accessToken: string, + context: { cloudId: string; spaceId: string; collection: string }, + visit: (entries: T[]) => void +): Promise { const cursors = new Set() + const pages = new Set() + const signal = AbortSignal.timeout(PERMISSION_COLLECTION_TIMEOUT_MS) let cursor: string | undefined - for (let page = 0; page < MAX_PAGES; page += 1) { + let entries = 0 + let fetchedPages = 0 + const fail = (reason: string): never => { + logger.warn('Confluence permission pagination did not complete', { + ...context, + pages: fetchedPages, + entries, + reason, + }) + throw new Error(`Confluence ${context.collection} ${reason}`) + } + for (let page = 0; page < MAX_PERMISSION_PAGES; page += 1) { const query = new URLSearchParams({ limit: String(PAGE_SIZE) }) if (cursor) query.set('cursor', cursor) const body = await getJson<{ results?: T[]; _links?: { next?: string } }>( `${url}?${query.toString()}`, - accessToken + accessToken, + { maxResponseBytes: PERMISSION_RESPONSE_MAX_BYTES, retryOptions: { signal } } ) - if (!Array.isArray(body.results)) { - throw new Error(`Confluence returned invalid ${what}`) + fetchedPages += 1 + if (!Array.isArray(body.results) || body.results.length > PAGE_SIZE) { + return fail('returned an invalid permission page') } - items.push(...body.results) + const results = body.results const next = body._links?.next - if (!next) return items + if (results.length > 0) { + const fingerprint = createHash('sha256') + .update( + results + .map((entry) => JSON.stringify(sortObjectKeysDeep(entry))) + .sort() + .join('\n') + ) + .digest('hex') + if (pages.has(fingerprint)) fail('repeated a permission page') + pages.add(fingerprint) + } + entries += results.length + visit(results) + if (!next) { + if (page >= MAX_PAGES) { + logger.info('Completed a large Confluence permission collection', { + ...context, + pages: page + 1, + entries, + }) + } + return + } cursor = extractCursor(next) - if (!cursor || cursors.has(cursor)) { - throw new Error(`Confluence returned an invalid or repeated ${what} continuation`) + if (!cursor || cursor.length > MAX_PERMISSION_CURSOR_LENGTH) { + return fail('returned an invalid or repeated permission continuation') } - cursors.add(cursor) + const cursorHash = createHash('sha256').update(cursor).digest('hex') + if (cursors.has(cursorHash)) fail('returned an invalid or repeated permission continuation') + cursors.add(cursorHash) } - throw new Error(`Confluence ${what} exceeded ${MAX_PAGES} pages (${items.length} entries)`) + fail(`exceeded ${MAX_PERMISSION_PAGES} pages (${entries} entries)`) } /** @@ -254,13 +304,8 @@ export async function listSpaceReadPrincipals( accessToken: string, spaceId: string ): Promise { - const entries = await drainV2( - `${apiBase(cloudId)}/api/v2/spaces/${encodeURIComponent(spaceId)}/permissions`, - accessToken, - 'space permissions' - ) - - const principals: ConfluencePrincipal[] = [] + const principals = new Map() + let principalBytes = 0 const accessTypes = new Set<'user' | 'admin'>() let grantedToRole = false let unmapped = 0 @@ -268,7 +313,14 @@ export async function listSpaceReadPrincipals( if (!id) { unmapped += 1 } else if (type === 'user' || type === 'group') { - principals.push({ kind: type, id }) + const key = `${type}:${id}` + if (!principals.has(key)) { + principalBytes += Buffer.byteLength(key, 'utf8') + if (principals.size >= MAX_ACL_TOKENS || principalBytes > PERMISSION_RESPONSE_MAX_BYTES) { + throw new Error('Confluence space readers exceeded the document permission limit') + } + principals.set(key, { kind: type, id }) + } } else if (type === 'access-class') { const accessClass = id.toLowerCase().replaceAll('_', '-') if (accessClass === 'all-licensed-users') { @@ -281,28 +333,37 @@ export async function listSpaceReadPrincipals( unmapped += 1 } } - for (const entry of entries) { - if (entry.operation?.key !== 'read' || entry.operation.targetType !== 'space') continue - const id = entry.principal?.id - const type = entry.principal?.type?.toLowerCase().replaceAll('_', '-') - if (type === 'role') { - grantedToRole = true - continue + await visitPermissionPages( + `${apiBase(cloudId)}/api/v2/spaces/${encodeURIComponent(spaceId)}/permissions`, + accessToken, + { cloudId, spaceId, collection: 'space permissions' }, + (entries) => { + for (const entry of entries) { + if (entry.operation?.key !== 'read' || entry.operation.targetType !== 'space') continue + const id = entry.principal?.id + const type = entry.principal?.type?.toLowerCase().replaceAll('_', '-') + if (type === 'role') { + grantedToRole = true + continue + } + addPrincipal(type, id) + } } - addPrincipal(type, id) - } + ) if (grantedToRole) { - const assignments = await drainV2( + await visitPermissionPages( `${apiBase(cloudId)}/api/v2/spaces/${encodeURIComponent(spaceId)}/role-assignments`, accessToken, - 'space role assignments' + { cloudId, spaceId, collection: 'space role assignments' }, + (assignments) => { + for (const assignment of assignments) { + const id = assignment.principal?.principalId + const type = assignment.principal?.principalType?.toLowerCase().replaceAll('_', '-') + addPrincipal(type, id) + } + } ) - for (const assignment of assignments) { - const id = assignment.principal?.principalId - const type = assignment.principal?.principalType?.toLowerCase().replaceAll('_', '-') - addPrincipal(type, id) - } } for (const accessType of accessTypes) { @@ -313,7 +374,7 @@ export async function listSpaceReadPrincipals( ) for (const group of groups) { if (!group.id) throw new Error('Confluence access group is missing its id') - principals.push({ kind: 'group', id: group.id }) + addPrincipal('group', group.id) } } @@ -327,11 +388,7 @@ export async function listSpaceReadPrincipals( } ) } - return [ - ...new Map( - principals.map((principal) => [`${principal.kind}:${principal.id}`, principal]) - ).values(), - ] + return [...principals.values()] } interface RestrictionPage { diff --git a/apps/sim/lib/knowledge/__integration__/member-document-lifecycle.integration.ts b/apps/sim/lib/knowledge/__integration__/member-document-lifecycle.integration.ts new file mode 100644 index 00000000000..42e15503338 --- /dev/null +++ b/apps/sim/lib/knowledge/__integration__/member-document-lifecycle.integration.ts @@ -0,0 +1,184 @@ +/** Real PostgreSQL coverage for bounded lifecycle writes and changing observations. */ +import { db } from '@sim/db' +import { + document, + knowledgeConnector, + knowledgeDocumentObservation, + organization, + user, + workspace, +} from '@sim/db/schema' +import { generateId } from '@sim/utils/id' +import { eq, inArray, sql } from 'drizzle-orm' +import { afterAll, afterEach, beforeEach, describe, expect, it } from 'vitest' +import { + type createKnowledgeAclFixtureIds, + seedKnowledgeAclFixture, + seedKnowledgeMemberFixture, +} from '@/lib/knowledge/__integration__/seed-source-access-fixture' +import { + applyMemberDocumentLifecycle, + recordMemberObservations, +} from '@/lib/knowledge/connectors/member-observations' +import { + assertSyncLeaseHeldInTx, + SyncLockLostException, + stillHoldsMemberSyncLock, +} from '@/lib/knowledge/connectors/sync-lock' + +describe('member document lifecycle in PostgreSQL', () => { + let ids: ReturnType + let members: Awaited> + const seenAt = sql`'2026-01-01 00:00:00.000123'::timestamp` + const deletedAt = new Date() + + beforeEach(async () => { + ids = await seedKnowledgeAclFixture() + members = await seedKnowledgeMemberFixture(ids) + }) + + afterEach(async () => { + await db.delete(workspace).where(eq(workspace.id, ids.workspaceId)) + await db.delete(organization).where(eq(organization.id, ids.organizationId)) + await db.delete(user).where(inArray(user.id, [ids.aliceId, ids.bobId])) + }) + + afterAll(() => db.$client.end()) + + const row = (externalId: string) => ({ + id: generateId(), + knowledgeBaseId: ids.knowledgeBaseId, + connectorId: members.connectorId, + externalId, + filename: externalId, + fileUrl: '', + fileSize: 0, + mimeType: 'text/plain', + processingStatus: 'completed', + contentHash: 'verified-content', + sourceSeenAt: seenAt, + }) + + const observe = (documentIds: string[]) => + recordMemberObservations(db, members.members[0].id, documentIds, members.runId) + + const run = (options: { beforeWrite?: () => Promise; allowRemoval?: boolean } = {}) => + applyMemberDocumentLifecycle({ + connectorId: members.connectorId, + knowledgeBaseId: ids.knowledgeBaseId, + runId: members.runId, + allowRemoval: options.allowRemoval ?? true, + deadlineAt: Date.now() + 60_000, + lease: { beatIfDue: async () => {} }, + withLease: async (fn) => { + await options.beforeWrite?.() + return db.transaction(async (tx) => { + await assertSyncLeaseHeldInTx(tx, members.connectorId, { + stillHeld: () => stillHoldsMemberSyncLock(members.connectorId, members.runId), + }) + return fn(tx) + }) + }, + }) + + it('walks multiple batches across null and microsecond timestamps and preserves exclusions', async () => { + const stale = Array.from({ length: 1100 }, (_, index) => ({ + ...row(String(index)), + sourceSeenAt: index < 550 ? null : seenAt, + })) + const observed = row('observed') + const excluded = { ...row('excluded'), userExcluded: true } + const archived = { ...row('archived'), archivedAt: new Date() } + const restore = { ...row('restore'), deletedAt } + const noContent = { ...row('no-content'), deletedAt, contentHash: null } + const rows = [...stale, observed, excluded, archived, restore, noContent] + for (let offset = 0; offset < rows.length; offset += 500) + await db.insert(document).values(rows.slice(offset, offset + 500)) + await observe([observed.id, restore.id, noContent.id]) + + expect(await run()).toEqual({ tombstoned: 1100, resurrected: 1, purged: 0, finished: true }) + const stored = await db + .select({ id: document.id, deletedAt: document.deletedAt }) + .from(document) + .where(eq(document.connectorId, members.connectorId)) + const byId = new Map(stored.map((entry) => [entry.id, entry.deletedAt])) + expect(stale.every((entry) => byId.get(entry.id) instanceof Date)).toBe(true) + for (const entry of [observed, excluded, archived, restore]) + expect(byId.get(entry.id)).toBeNull() + expect(byId.get(noContent.id)).toEqual(deletedAt) + }) + + it('continues past a full selected batch even if its observations change before UPDATE', async () => { + const rows = Array.from({ length: 501 }, (_, index) => row(String(index))) + await db.insert(document).values(rows) + const selected = await db + .select({ id: document.id }) + .from(document) + .where(eq(document.connectorId, members.connectorId)) + .orderBy(document.id) + .limit(500) + let writes = 0 + const result = await run({ + beforeWrite: async () => { + if (writes++ === 0) await observe(selected.map(({ id }) => id)) + }, + }) + expect(result).toEqual({ tombstoned: 1, resurrected: 0, purged: 0, finished: true }) + const stored = await db + .select({ id: document.id, deletedAt: document.deletedAt }) + .from(document) + .where(eq(document.connectorId, members.connectorId)) + expect(stored.filter((entry) => entry.deletedAt !== null)).toHaveLength(1) + const observedIds = new Set(selected.map(({ id }) => id)) + expect(stored.every((entry) => !observedIds.has(entry.id) || entry.deletedAt === null)).toBe( + true + ) + }) + + it('does not resurrect a document whose last observation disappeared before UPDATE', async () => { + const target = { ...row('lost-observation'), deletedAt } + await db.insert(document).values(target) + await observe([target.id]) + expect( + await run({ + beforeWrite: async () => { + await db + .delete(knowledgeDocumentObservation) + .where(eq(knowledgeDocumentObservation.documentId, target.id)) + }, + }) + ).toEqual({ tombstoned: 0, resurrected: 0, purged: 0, finished: true }) + const [stored] = await db.select().from(document).where(eq(document.id, target.id)) + expect(stored.deletedAt).toEqual(deletedAt) + }) + + it('leaves unobserved documents alone until a member completed a listing', async () => { + const target = row('not-yet-observed') + await db.insert(document).values(target) + expect(await run({ allowRemoval: false })).toEqual({ + tombstoned: 0, + resurrected: 0, + purged: 0, + finished: true, + }) + const [stored] = await db.select().from(document).where(eq(document.id, target.id)) + expect(stored.deletedAt).toBeNull() + }) + + it('refuses the write when another run claimed the connector after selection', async () => { + const target = row('lost-lease') + await db.insert(document).values(target) + await expect( + run({ + beforeWrite: async () => { + await db + .update(knowledgeConnector) + .set({ memberSyncLockToken: generateId() }) + .where(eq(knowledgeConnector.id, members.connectorId)) + }, + }) + ).rejects.toBeInstanceOf(SyncLockLostException) + const [stored] = await db.select().from(document).where(eq(document.id, target.id)) + expect(stored.deletedAt).toBeNull() + }) +}) diff --git a/apps/sim/lib/knowledge/connectors/member-observations.test.ts b/apps/sim/lib/knowledge/connectors/member-observations.test.ts index 79ccd3b5430..3c7d0a41ab0 100644 --- a/apps/sim/lib/knowledge/connectors/member-observations.test.ts +++ b/apps/sim/lib/knowledge/connectors/member-observations.test.ts @@ -162,6 +162,7 @@ describe('applyMemberDocumentLifecycle', () => { it('tombstones in bounded transactions and leaves remaining work for the next run', async () => { const batch = Array.from({ length: 500 }, (_, i) => ({ id: `document-${i}` })) + queueTableRows(schemaMock.document, batch) dbChainMockFns.returning.mockResolvedValueOnce(batch) const input: Parameters[0] = { connectorId: 'c-1', @@ -184,6 +185,8 @@ describe('applyMemberDocumentLifecycle', () => { }) expect(dbChainMockFns.limit).toHaveBeenCalledWith(500) dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'remaining' }]).mockResolvedValueOnce([]) + queueTableRows(schemaMock.document, [{ id: 'remaining' }]) + queueTableRows(schemaMock.document, []) queueTableRows(schemaMock.document, []) expect( await applyMemberDocumentLifecycle({ ...input, deadlineAt: Date.now() + 60_000 }) @@ -197,6 +200,8 @@ describe('applyMemberDocumentLifecycle', () => { it('reports a reclaimed lease during a purge batch as the run being superseded', async () => { dbChainMockFns.returning.mockResolvedValueOnce([]) + queueTableRows(schemaMock.document, []) + queueTableRows(schemaMock.document, []) queueTableRows(schemaMock.document, [{ id: 'd-1' }]) vi.mocked(hardDeleteDocuments).mockRejectedValueOnce( new ConnectorSyncDeletionGuardError('lease reclaimed') diff --git a/apps/sim/lib/knowledge/connectors/member-observations.ts b/apps/sim/lib/knowledge/connectors/member-observations.ts index ab9efdbd53b..cd3123a5e16 100644 --- a/apps/sim/lib/knowledge/connectors/member-observations.ts +++ b/apps/sim/lib/knowledge/connectors/member-observations.ts @@ -7,6 +7,7 @@ import { } from '@sim/db/schema' import { and, + asc, eq, exists, gt, @@ -291,31 +292,53 @@ export async function applyMemberDocumentLifecycle(input: { } for (const phase of ['tombstoned', 'resurrected'] as const) { if (phase === 'tombstoned' && !input.allowRemoval) continue + const seenOrder = sql`COALESCE(${document.sourceSeenAt}, '-infinity'::timestamp)` + let after: { id: string; seenAt: string } | undefined for (;;) { if (Date.now() >= input.deadlineAt) return result await input.lease.beatIfDue() - const changed = await input.withLease(async (tx) => { - const condition = and( - eq(document.connectorId, connectorId), - eq(document.userExcluded, false), - isNull(document.archivedAt), - phase === 'tombstoned' - ? and(isNull(document.deletedAt), hasNoObservation()) - : and(isNotNull(document.deletedAt), isNotNull(document.contentHash), hasObservation()) + const condition = and( + eq(document.connectorId, connectorId), + eq(document.userExcluded, false), + isNull(document.archivedAt), + phase === 'tombstoned' + ? and(isNull(document.deletedAt), hasNoObservation()) + : and(isNotNull(document.deletedAt), isNotNull(document.contentHash), hasObservation()) + ) + /** Materialize the limited IDs before UPDATE so its observation check stays batch-bound. */ + const candidates = await db + .select({ id: document.id, seenAt: sql`${seenOrder}::text` }) + .from(document) + .where( + and( + condition, + after + ? sql`(${seenOrder}, ${document.id}) > (${after.seenAt}::timestamp, ${after.id})` + : undefined + ) ) - const candidates = tx - .select({ id: document.id }) - .from(document) - .where(condition) - .limit(MATERIALIZE_BATCH_SIZE) + .orderBy(seenOrder, asc(document.id)) + .limit(MATERIALIZE_BATCH_SIZE) + if (candidates.length === 0) break + if (Date.now() >= input.deadlineAt) return result + const changed = await input.withLease(async (tx) => { return tx .update(document) .set({ deletedAt: phase === 'tombstoned' ? now : null }) - .where(and(condition, inArray(document.id, candidates))) + .where( + and( + condition, + inArray( + document.id, + candidates.map(({ id }) => id) + ) + ) + ) .returning({ id: document.id }) }) result[phase] += changed.length - if (changed.length < MATERIALIZE_BATCH_SIZE) break + after = candidates.at(-1) + if (candidates.length < MATERIALIZE_BATCH_SIZE) break } } From 709afcdd7acf4efd464de2e448424490d38dac11 Mon Sep 17 00:00:00 2001 From: Waleed Latif Date: Wed, 16 Sep 2026 23:34:27 -0700 Subject: [PATCH 2/2] fix(knowledge): retain permission notices alongside member errors --- .../[connectorId]/source-detail.test.tsx | 19 +++++++++++++++++++ .../sources/[connectorId]/source-detail.tsx | 2 +- 2 files changed, 20 insertions(+), 1 deletion(-) diff --git a/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.test.tsx b/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.test.tsx index 51449023283..4b7cb5598ad 100644 --- a/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.test.tsx +++ b/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.test.tsx @@ -273,6 +273,25 @@ describe('organization source detail navigation', () => { } ) + it.each(['idle', 'pending', 'running'] as const)( + 'preserves a connector permission warning alongside a member error while %s', + async (memberSyncStatus) => { + mocks.detail.mockReturnValue({ + data: { + ...connector, + accessMode: 'members', + memberSyncStatus, + lastSyncError: SOURCE_PERMISSION_ERROR, + lastMemberSyncError: 'Private member error details', + }, + }) + await render() + expect(container.textContent).toContain('Permission verification incomplete') + expect(container.textContent).toContain(SOURCE_PERMISSION_ERROR) + expect(container.textContent).not.toContain('Private member error details') + } + ) + it.each(['', '?view=settings', '?view=history'])( 'shows integration deactivation independently of source sync state at %s', async (searchParams) => { diff --git a/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.tsx b/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.tsx index 37efe1a9bae..e2ad6985e22 100644 --- a/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.tsx +++ b/apps/sim/app/o/[organizationId]/settings/integrations/sources/[connectorId]/source-detail.tsx @@ -187,7 +187,7 @@ function SourceDetailContent({ ? describeSearchSource(meta, connector.sourceConfig) || meta.name : 'Connection' const { effectiveStatus, lastSyncError } = getConnectorSyncState(connector) - const permissionsIncomplete = lastSyncError === SOURCE_PERMISSION_ERROR + const permissionsIncomplete = connector.lastSyncError === SOURCE_PERMISSION_ERROR const status = effectiveStatus === 'paused' ? 'Sync paused'