diff --git a/.github/workflows/test-build.yml b/.github/workflows/test-build.yml index 191e29a9bce..07364732507 100644 --- a/.github/workflows/test-build.yml +++ b/.github/workflows/test-build.yml @@ -233,6 +233,7 @@ jobs: lib/knowledge/__integration__/search-source-progress.integration.ts lib/knowledge/__integration__/search-source-pagination.integration.ts lib/knowledge/__integration__/search-reference-batching.integration.ts + lib/knowledge/__integration__/embedding-insert-batches.integration.ts lib/knowledge/__integration__/kb-block-search.integration.ts lib/core/outbox/service.integration.ts lib/knowledge/__integration__/connector-upload.integration.ts diff --git a/apps/docs/content/docs/search/gmail.mdx b/apps/docs/content/docs/search/gmail.mdx index 0d86d7a59e9..ecb646f4e65 100644 --- a/apps/docs/content/docs/search/gmail.mdx +++ b/apps/docs/content/docs/search/gmail.mdx @@ -133,7 +133,7 @@ Search schedules syncs hourly. The first sync lists every thread in scope and ca **Member accounts:** later syncs use each mailbox's Gmail change history, unless the configuration has a search filter. A full relisting runs about weekly, or sooner if Gmail no longer retains the saved history. -**Service account:** each sync revisits the selected active mailboxes and resumes unfinished listings. A failed mailbox read leaves the crawl incomplete; it does not cause existing indexed mail to be deleted from Search. +**Service account:** each sync revisits the selected active mailboxes and resumes unfinished listings. If Google reports that a user's mailbox is not set up or returns a mailbox `failedPrecondition`, Sim records a warning and continues with the remaining users. The crawl stays incomplete and retries affected users on the next scheduled crawl; existing indexed mail is not deleted because a mailbox could not be read. Credential, delegation, and Directory failures still stop the crawl. Updates, removals, and access refresh in the background. Empty mailboxes and filters with no matches complete normally with zero documents. Threads exceeding indexing size limits are skipped and reconsidered when they change. @@ -149,7 +149,7 @@ An individual thread failure does not mean the whole mailbox failed. Sim retries | Reconnect | Click **Reconnect** and authorize the same account again. | | Unavailable or needs admin attention | Ask your Sim admin to check source status and the deployment's Google OAuth configuration. | | Directory or delegation error | Check both central crawl scopes, the service-account key, and the Directory administrator's user-read privileges. A normal OAuth account cannot replace the central service account. | -| Gmail access fails for a selected user | Verify delegation is authorized and [Gmail is enabled](https://knowledge.workspace.google.com/admin/gmail/control-gmail-access-for-your-organizations-users) for that primary Workspace account. Set **Users** to accounts with Gmail enabled; leaving it blank includes all active users and can stop sync on a service-access error. Aliases and external accounts cannot be selected. | +| Gmail access fails for a selected user | Verify delegation is authorized and [Gmail is enabled](https://knowledge.workspace.google.com/admin/gmail/control-gmail-access-for-your-organizations-users) for that primary Workspace account. Check the affected users in **Sync history**. Set **Users** to accounts with Gmail enabled if some active users should not be crawled. Aliases and external accounts cannot be selected. | | A central source indexes mail but a teammate sees no results | Confirm their verified Sim email is the mailbox's primary email and they belong to the Sim organization. Administrators do not receive other people's mailbox access. | ## Self-hosted operator setup diff --git a/apps/docs/content/docs/search/google-calendar.mdx b/apps/docs/content/docs/search/google-calendar.mdx index a2317d9d941..d6fd93c15e2 100644 --- a/apps/docs/content/docs/search/google-calendar.mdx +++ b/apps/docs/content/docs/search/google-calendar.mdx @@ -131,7 +131,7 @@ Sim indexes event titles, descriptions, times, locations, and the selected atten Cancelled events, attachment contents, meeting recordings, and transcripts are not indexed. Status entries such as working location, out of office, focus time, and birthdays, and automatically generated reservation events from Gmail are not indexed. Events Google returns only as free/busy blocks, without searchable details, are not indexed. Events outside the selected date window are excluded. Private event details that Google withholds are not available in Search; see [Google's calendar sharing rules](https://developers.google.com/workspace/calendar/api/concepts/sharing). -Search schedules syncs hourly. Event edits, cancellations, access changes, inactive or removed users, and events moving outside the date window are reconciled during completed background syncs. Central crawls page through each selected user and resume unfinished work before removing documents no longer listed. Authorization, quota, and provider failures stop the sync rather than treating unread calendars as empty. The first sync may take longer, and results appear as indexing progresses; Search is not a live Calendar read. +Search schedules syncs hourly. Event edits, cancellations, access changes, inactive or removed users, and events moving outside the date window are reconciled during completed background syncs. Central crawls page through each selected user and resume unfinished work before removing documents no longer listed. If an individual user's event listing returns a `403` with no reason or only `forbidden`, Sim records a warning and continues with the remaining users. The crawl stays incomplete and retries affected users on the next scheduled crawl; unread calendars are not treated as empty. Credential, delegation, Directory, and other provider failures still stop the crawl. The first sync may take longer, and results appear as indexing progresses; Search is not a live Calendar read. ## Troubleshooting @@ -146,7 +146,7 @@ When a sync fails, **Sync history** includes the Google API operation, HTTP stat | Reconnect | Click **Reconnect** and complete Google authorization again. Allow pop-ups if the connection tab does not open. | | Unavailable or needs admin attention | Ask your Sim admin to check source status and the deployment's Google OAuth configuration. | | Service-account authorization or Directory error | Confirm both delegated scopes, enabled APIs, and the Directory administrator's user-read privilege. Check whether delegation still awaits approval or propagation. | -| Calendar is disabled for a selected user | An active Workspace user may have Calendar turned off. [Enable Calendar](https://knowledge.workspace.google.com/admin/users/access/turn-calendar-on-or-off-for-users) for them, or set **Users** to accounts with Calendar enabled. Leaving **Users** blank includes all active users and can stop sync on a service-access error. | +| Calendar access fails for a selected user | Check the affected users in **Sync history**. An active Workspace user may have Calendar turned off, but a `403` alone does not prove this. Check their access to the selected calendars and [Calendar service settings](https://knowledge.workspace.google.com/admin/users/access/turn-calendar-on-or-off-for-users), or set **Users** to accounts that should be crawled. | | User not found or inactive | Use an active primary email in the same Workspace customer. Aliases, external or guest accounts, suspended users, and archived users cannot be selected. | | A central source has no results for a teammate | Confirm their primary Workspace email matches their verified Sim email, they belong to the Sim organization, and they are included in **Users**. Check calendar IDs and **Sync history**. | diff --git a/apps/docs/content/docs/search/google-drive.mdx b/apps/docs/content/docs/search/google-drive.mdx index 3c70e3ccb62..2dd2d0237e9 100644 --- a/apps/docs/content/docs/search/google-drive.mdx +++ b/apps/docs/content/docs/search/google-drive.mdx @@ -155,13 +155,14 @@ Search schedules syncs hourly. Central crawls revisit the selected users' files ## Troubleshooting -**Directory permission sync failed** means Sim could not fully verify group membership. Check the Directory administrator’s access to the affected group and any nested groups; this is separate from file-download access. An incomplete membership read does not replace the last verified membership, which remains subject to freshness checks. +Directory permission warnings mean Sim could not fully verify group membership. When other groups refresh successfully and only nested groups outside your Workspace customer deny access or cannot be found, Sim continues syncing content and reports a partial permission sync. An incomplete membership read does not replace or refresh the last verified membership, which remains subject to freshness checks. Directory-wide failures still stop the sync. Check the Directory administrator’s access to the affected group and nested groups; this is separate from file-download access. | Problem | Next step | | --- | --- | | Google rejects authorization (`unauthorized_client`) | In **Manage Domain Wide Delegation**, verify the numeric **Client ID** matches `client_id` in the JSON key uploaded to Sim and all required scopes appear under **View details**. Check pending approval and allow time for recent changes to propagate. Changing the OAuth consent screen alone does not authorize delegation. | | Directory access failed | Check all four delegated scopes and the **Directory administrator email** user's administrator privileges. A normal Google OAuth credential cannot supply this central Search path. | | Missing files in a central crawl | Check **Users**, folder and file-type filters, and whether selected active Workspace users can download the file and read its permissions. Opening a file alone does not prove either. Check Sync history for errors. Files reachable only by excluded or inactive accounts are not crawled; files with unverified permissions stay hidden. | +| No text could be extracted | Images and scanned PDFs use OCR, but files with no extractable text cannot be indexed. Check whether the original contains readable text; a successful download does not guarantee searchable content. | | User not found or inactive | Use a primary email in the same Google Workspace customer. Aliases, external or guest accounts, suspended users, and archived users cannot be selected for crawling. | | A teammate sees no results | Confirm they have joined the Sim organization and their verified Sim email matches the Drive permission or group membership. For member accounts, finish their personal Drive connection too. | | A public or shared-link file is missing | Check **Openly shared files**. Link-only sharing does not grant Search access. A named user or group permission can still make the file searchable. | diff --git a/apps/sim/connectors/gmail/company-crawl.test.ts b/apps/sim/connectors/gmail/company-crawl.test.ts index 4e73f8ef405..f0e180637a2 100644 --- a/apps/sim/connectors/gmail/company-crawl.test.ts +++ b/apps/sim/connectors/gmail/company-crawl.test.ts @@ -409,6 +409,69 @@ describe('company-wide Gmail indexing', () => { ).rejects.toThrow('403') }) + it('continues after Gmail failedPrecondition with the later mailbox owner ACL intact', async () => { + fetchProvider.mockImplementation(async (url: string, init?: RequestInit) => { + if ( + new URL(url).pathname.endsWith('/threads') && + new Headers(init?.headers).get('Authorization')?.includes(ALICE.email) + ) { + return Response.json( + { + error: { + message: 'private provider response', + errors: [{ reason: 'failedPrecondition' }], + }, + }, + { status: 400 } + ) + } + return providerResponse(url, init) + }) + const first = await gmailConnector.listDocuments( + 'directory-token', + CONFIG, + undefined, + centralContext() + ) + expect(first).toMatchObject({ + documents: [], + reconciliationSafe: false, + listingFailures: { + count: 1, + samples: [ + { + scope: ALICE.email, + operation: 'gmail.threads.list', + status: 400, + reasons: ['failedPrecondition'], + }, + ], + }, + }) + expect(JSON.stringify(first)).not.toContain('private provider response') + const ctx = centralContext() + const second = await gmailConnector.listDocuments( + 'directory-token', + CONFIG, + first.nextCursor, + ctx + ) + expect(second).toMatchObject({ + hasMore: false, + reconciliationSafe: false, + listingFailures: first.listingFailures, + }) + expect(second.documents[0].acl).toEqual([`u:${BOB.email}`]) + const hydrated = await gmailConnector.getDocument( + 'directory-token', + CONFIG, + second.documents[0].externalId, + ctx + ) + expect(hydrated?.acl).toEqual([`u:${BOB.email}`]) + expect(hydrated?.content).toContain('Bob private body') + }) + it('invalidates previous hydration authority when the next mailbox fails', async () => { const context = centralContext() const first = await gmailConnector.listDocuments('directory-token', CONFIG, undefined, context) diff --git a/apps/sim/connectors/google-calendar/company-crawl.test.ts b/apps/sim/connectors/google-calendar/company-crawl.test.ts index d1e49c9d82c..63ecd8afcb2 100644 --- a/apps/sim/connectors/google-calendar/company-crawl.test.ts +++ b/apps/sim/connectors/google-calendar/company-crawl.test.ts @@ -340,16 +340,53 @@ describe('Google Calendar company crawl', () => { expect(next.documents[0].acl).toEqual([`u:${BOB.email}`]) }) - it.each([401, 403])( - 'propagates provider HTTP %s without completing a user’s listing', - async (status) => { - fetchMock.mockResolvedValue(response({ error: { code: status } }, status)) + it.each([ + { status: 401, reason: 'authError' }, + { status: 403, reason: 'insufficientPermissions' }, + { status: 403, reason: 'SERVICE_DISABLED' }, + ])( + 'propagates provider authorization failures without completing a listing: $reason', + async ({ status, reason }) => { + fetchMock.mockResolvedValue(response({ error: { errors: [{ reason }] } }, status)) await expect( googleCalendarConnector.listDocuments('directory-token', {}, undefined, context()) ).rejects.toThrow() } ) + it('continues another user after an unclassified list access denial and retains its diagnostic', async () => { + fetchMock.mockResolvedValueOnce(response({ error: { code: 403 } }, 403)) + const first = await googleCalendarConnector.listDocuments( + 'directory-token', + {}, + undefined, + context() + ) + expect(first).toMatchObject({ + documents: [], + hasMore: true, + reconciliationSafe: false, + listingFailures: { + count: 1, + samples: [ + { scope: ALICE.email, operation: 'calendar.events.list', status: 403, reasons: [] }, + ], + }, + }) + const second = await googleCalendarConnector.listDocuments( + 'directory-token', + {}, + first.nextCursor, + context() + ) + expect(second.documents[0].acl).toEqual([`u:${BOB.email}`]) + expect(second).toMatchObject({ + hasMore: false, + reconciliationSafe: false, + listingFailures: first.listingFailures, + }) + }) + it('skips a user who became inactive before their page and never delegates to them', async () => { mockGetUser.mockResolvedValueOnce({ ...ALICE, active: false }) const syncContext = context() diff --git a/apps/sim/connectors/google-drive/directory.test.ts b/apps/sim/connectors/google-drive/directory.test.ts index 660cd191bf3..0e630e68e11 100644 --- a/apps/sim/connectors/google-drive/directory.test.ts +++ b/apps/sim/connectors/google-drive/directory.test.ts @@ -3,6 +3,7 @@ */ import { beforeEach, describe, expect, it, vi } from 'vitest' import { listDomainGroups, openGoogleDirectory } from '@/connectors/google-drive/directory' +import { ConnectorDirectoryGroupAccessError } from '@/connectors/source-error' const mockFetch = vi.fn() @@ -234,6 +235,61 @@ describe('the membership a directory reports', () => { }) }) + it.each([ + { status: 403, reason: 'forbidden' }, + { status: 404, reason: 'notFound' }, + ])( + 'classifies inaccessible external nested groups explicitly: $status $reason', + async ({ status, reason }) => { + directory({ 'eng@corp.com': [USER('alice@corp.com'), NESTED('restricted@external.com')] }) + const healthy = mockFetch.getMockImplementation()! + mockFetch.mockImplementation(async (url: string) => { + if ( + decodeURIComponent(new URL(url).pathname).includes('/restricted@external.com/members') + ) { + return jsonResponse( + { error: { errors: [{ reason }], message: 'private detail' } }, + status + ) + } + return healthy(url) + }) + const failure = await membersOf(GROUP).catch((error: unknown) => error) + expect(failure).toBeInstanceOf(ConnectorDirectoryGroupAccessError) + expect(failure).toMatchObject({ + cause: { status, diagnostic: { operation: 'directory.members.list', reasons: [reason] } }, + }) + expect(String(failure)).not.toContain('private detail') + } + ) + + it.each([ + { email: 'restricted@corp.io', status: 403, reasons: ['forbidden'] }, + { email: 'restricted@external.com', status: 403, reasons: [] }, + { email: 'restricted@external.com', status: 403, reasons: ['forbidden', 'unknownReason'] }, + { + email: 'restricted@external.com', + status: 403, + reasons: ['forbidden', 'insufficientPermissions'], + }, + { email: 'restricted@external.com', status: 401, reasons: ['authError'] }, + ])( + 'does not classify uncertain or customer-owned failures as external access failures: $email $status $reasons', + async ({ email, status, reasons }) => { + directory({ 'eng@corp.com': [NESTED(email)] }) + const healthy = mockFetch.getMockImplementation()! + mockFetch.mockImplementation(async (url: string) => { + if (decodeURIComponent(new URL(url).pathname).includes(`/${email}/members`)) { + return jsonResponse({ error: { errors: reasons.map((reason) => ({ reason })) } }, status) + } + return healthy(url) + }) + const failure = await membersOf(GROUP).catch((error: unknown) => error) + expect(failure).not.toBeInstanceOf(ConnectorDirectoryGroupAccessError) + expect(failure).toMatchObject({ status }) + } + ) + /** A directory that hiccups must not cost a group its membership; transient errors are retried. */ it('retries a transient directory error before giving up', async () => { directory({ 'eng@corp.com': [USER('alice@corp.com')] }) diff --git a/apps/sim/connectors/google-drive/directory.ts b/apps/sim/connectors/google-drive/directory.ts index dd2b9ec1eda..78bb57e75ef 100644 --- a/apps/sim/connectors/google-drive/directory.ts +++ b/apps/sim/connectors/google-drive/directory.ts @@ -16,6 +16,7 @@ import { fetchGoogleDriveWithRetry, GoogleDriveApiError, } from '@/connectors/google-drive/google-drive-errors' +import { ConnectorDirectoryGroupAccessError } from '@/connectors/source-error' import type { ConnectorDirectory, ConnectorDirectoryGroup, @@ -249,7 +250,28 @@ async function listGroupMembers( return } - for (const member of await membersOf(groupId)) { + let members: RawMember[] + try { + members = await membersOf(groupId) + } catch (error) { + const groupDomain = emailDomain(groupId) + if ( + depth > 0 && + groupDomain && + !customerDomains.includes(groupDomain) && + error instanceof GoogleDriveApiError && + error.reasonsComplete && + ((error.status === 403 && error.reasons.length === 1 && error.reasons[0] === 'forbidden') || + (error.status === 404 && error.reasons.length === 1 && error.reasons[0] === 'notFound')) + ) { + throw new ConnectorDirectoryGroupAccessError('An external nested group cannot be read', { + cause: error, + }) + } + throw error + } + + for (const member of members) { if (member.status && member.status.toUpperCase() !== 'ACTIVE') continue const type = member.type?.toUpperCase() diff --git a/apps/sim/connectors/google-drive/google-drive-errors.ts b/apps/sim/connectors/google-drive/google-drive-errors.ts index 86a582a68b2..141a0195906 100644 --- a/apps/sim/connectors/google-drive/google-drive-errors.ts +++ b/apps/sim/connectors/google-drive/google-drive-errors.ts @@ -6,7 +6,7 @@ import { retryWithExponentialBackoff, } from '@/lib/knowledge/documents/utils' import { - readGoogleErrorReasons, + readGoogleErrorDetails, safeGoogleErrorReasons, } from '@/connectors/google-workspace/api-errors' import { @@ -104,8 +104,14 @@ export class GoogleDriveApiError extends ConnectorSourceError { readonly reasons: readonly string[] readonly kind: GoogleDriveErrorKind readonly rateLimited: boolean + readonly reasonsComplete: boolean - constructor(status: number, normalizedReasons: readonly string[], operation = 'drive.request') { + constructor( + status: number, + normalizedReasons: readonly string[], + operation = 'drive.request', + reasonsComplete = true + ) { const diagnosticReasons = safeGoogleErrorReasons(normalizedReasons).slice( 0, GOOGLE_ERROR_REASON_MAX_COUNT @@ -123,6 +129,8 @@ export class GoogleDriveApiError extends ConnectorSourceError { this.kind = kind this.rateLimited = status === 429 || normalizedReasons.some((reason) => RATE_LIMIT_REASONS.has(reason)) + this.reasonsComplete = + reasonsComplete && normalizedReasons.every((reason) => diagnosticReasons.includes(reason)) } } @@ -135,7 +143,8 @@ export async function readGoogleDriveApiError( response: Response, operation = 'drive.request' ): Promise { - return new GoogleDriveApiError(response.status, await readGoogleErrorReasons(response), operation) + const details = await readGoogleErrorDetails(response) + return new GoogleDriveApiError(response.status, details.reasons, operation, details.complete) } /** diff --git a/apps/sim/connectors/google-workspace/api-errors.test.ts b/apps/sim/connectors/google-workspace/api-errors.test.ts index f6e4ace1b8f..04ca5ec08ce 100644 --- a/apps/sim/connectors/google-workspace/api-errors.test.ts +++ b/apps/sim/connectors/google-workspace/api-errors.test.ts @@ -39,6 +39,7 @@ describe('Google API diagnostics', () => { it('omits unknown reason tokens even when they look like machine codes', async () => { const error = await readGoogleApiError(failure(403, RESPONSE_SECRET), OPERATION) expect(error.diagnostic?.reasons).toEqual([]) + expect(error.reasonsComplete).toBe(false) expect(JSON.stringify(error)).not.toContain(RESPONSE_SECRET) }) @@ -70,8 +71,57 @@ describe('Google API diagnostics', () => { const error = await readGoogleApiError(new Response(body, { status: 400 }), OPERATION) expect(error.status).toBe(400) expect(error.diagnostic?.reasons).toEqual([]) + expect(error.reasonsComplete).toBe(false) } ) + + it('distinguishes a valid reasonless envelope from stripped or malformed reasons', async () => { + const absent = await readGoogleApiError( + Response.json({ error: { code: 403, message: RESPONSE_SECRET } }, { status: 403 }), + 'calendar.events.list' + ) + expect(absent.reasonsComplete).toBe(true) + expect(absent.diagnostic?.reasons).toEqual([]) + const malformed = await readGoogleApiError( + Response.json({ error: { errors: [{ reason: 123 }] } }, { status: 403 }), + 'calendar.events.list' + ) + expect(malformed.reasonsComplete).toBe(false) + const mixed = await readGoogleApiError( + Response.json( + { error: { errors: [{ reason: 'forbidden' }, { reason: RESPONSE_SECRET }] } }, + { status: 403 } + ), + 'calendar.events.list' + ) + expect(mixed.diagnostic?.reasons).toEqual(['forbidden']) + expect(mixed.reasonsComplete).toBe(false) + expect(JSON.stringify([absent, malformed, mixed])).not.toContain(RESPONSE_SECRET) + }) + + it.each([{ error: [] }, { error: {} }, { error: { code: 401 } }])( + 'does not classify a malformed or inconsistent reasonless envelope as complete: %j', + async (payload) => { + const error = await readGoogleApiError( + Response.json(payload, { status: 403 }), + 'calendar.events.list' + ) + expect(error.reasonsComplete).toBe(false) + } + ) + + it('retains recognized diagnostics when another part of the envelope is malformed', async () => { + const error = await readGoogleApiError( + Response.json( + { error: { errors: [{ reason: 'forbidden' }], details: RESPONSE_SECRET } }, + { status: 403 } + ), + 'calendar.events.list' + ) + expect(error.diagnostic?.reasons).toEqual(['forbidden']) + expect(error.reasonsComplete).toBe(false) + expect(JSON.stringify(error)).not.toContain(RESPONSE_SECRET) + }) }) describe('Google API retries', () => { diff --git a/apps/sim/connectors/google-workspace/api-errors.ts b/apps/sim/connectors/google-workspace/api-errors.ts index 12653fd37c8..e5c16f75b02 100644 --- a/apps/sim/connectors/google-workspace/api-errors.ts +++ b/apps/sim/connectors/google-workspace/api-errors.ts @@ -57,35 +57,55 @@ export function safeGoogleErrorReasons(reasons: readonly string[]): string[] { return [...new Set(reasons.filter((reason) => SAFE_REASONS.has(reason)))] } -/** Reads the bounded Google envelope without retaining provider messages or request data. */ -export async function readGoogleErrorReasons(response: Response): Promise { +/** Distinguishes a reasonless envelope from one whose reason codes cannot safely be interpreted. */ +export async function readGoogleErrorDetails( + response: Response +): Promise<{ reasons: string[]; complete: boolean }> { + const unreadable = { reasons: [], complete: false } const body = await readBodyWithLimit(response, ERROR_BODY_MAX_BYTES).catch(() => null) - if (!body) return [] + if (!body) return unreadable try { const payload: unknown = JSON.parse(body.toString('utf8')) - if (!payload || typeof payload !== 'object' || !('error' in payload)) return [] + if (!payload || typeof payload !== 'object' || !('error' in payload)) return unreadable const error = payload.error - if (!error || typeof error !== 'object') return [] + if (!error || typeof error !== 'object' || Array.isArray(error)) return unreadable + const envelopeValid = + (!('errors' in error) || Array.isArray(error.errors)) && + (!('details' in error) || Array.isArray(error.details)) const entries = [ ...('errors' in error && Array.isArray(error.errors) ? error.errors : []), ...('details' in error && Array.isArray(error.details) ? error.details : []), ] - return safeGoogleErrorReasons( - entries.flatMap((entry: unknown) => - entry && typeof entry === 'object' && 'reason' in entry && typeof entry.reason === 'string' - ? [entry.reason] - : [] - ) + const reasons = entries.flatMap((entry: unknown) => + entry && typeof entry === 'object' && 'reason' in entry && typeof entry.reason === 'string' + ? [entry.reason] + : [] ) + const safeReasons = safeGoogleErrorReasons(reasons) + return { + reasons: safeReasons, + complete: + envelopeValid && + (entries.length > 0 || ('code' in error && error.code === response.status)) && + reasons.length === entries.length && + reasons.every((reason) => SAFE_REASONS.has(reason)) && + safeReasons.length <= MAX_REASONS, + } } catch { - return [] + return unreadable } } export class GoogleApiError extends ConnectorSourceError { readonly rateLimited: boolean + readonly reasonsComplete: boolean retryAfterMs?: number - constructor(operation: string, status: number, reasons: readonly string[]) { + constructor( + operation: string, + status: number, + reasons: readonly string[], + reasonsComplete = true + ) { const safeReasons = safeGoogleErrorReasons(reasons) const suffix = safeReasons.length ? ` (${safeReasons.join(', ')})` : '' const category = @@ -105,6 +125,10 @@ export class GoogleApiError extends ConnectorSourceError { reasons: safeReasons.slice(0, MAX_REASONS), }) this.name = 'GoogleApiError' + this.reasonsComplete = + reasonsComplete && + reasons.every((reason) => SAFE_REASONS.has(reason)) && + safeReasons.length <= MAX_REASONS this.rateLimited = status === 429 || safeReasons.some((reason) => RATE_LIMIT_REASONS.has(reason)) } @@ -114,7 +138,8 @@ export async function readGoogleApiError( response: Response, operation: string ): Promise { - return new GoogleApiError(operation, response.status, await readGoogleErrorReasons(response)) + const details = await readGoogleErrorDetails(response) + return new GoogleApiError(operation, response.status, details.reasons, details.complete) } /** Preserves Google diagnostics when the shared transport retries a transient HTTP response. */ diff --git a/apps/sim/connectors/google-workspace/company-crawl.test.ts b/apps/sim/connectors/google-workspace/company-crawl.test.ts index 1e34062abfc..85456985800 100644 --- a/apps/sim/connectors/google-workspace/company-crawl.test.ts +++ b/apps/sim/connectors/google-workspace/company-crawl.test.ts @@ -1,5 +1,6 @@ /** @vitest-environment node */ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { GoogleApiError } from '@/connectors/google-workspace/api-errors' import { getGoogleWorkspaceDocument, InvalidGoogleWorkspaceCursor, @@ -54,9 +55,14 @@ function document(ctx?: Record, id = 'shared-provider-id'): Ext } } const listUserDocuments = vi.fn() -function list(syncContext: Record, cursor?: string, sourceConfig = CONFIG) { +function list( + syncContext: Record, + cursor?: string, + sourceConfig = CONFIG, + provider: 'gmail' | 'google_calendar' = 'gmail' +) { return listGoogleWorkspaceDocuments({ - provider: 'gmail', + provider, accessToken: 'directory-token', sourceConfig, syncContext, @@ -373,7 +379,7 @@ describe('Google Workspace per-user central crawl', () => { const url = new URL(mockFetch.mock.calls[0][0]) expect(url.searchParams.get('maxResults')).toBe('100') expect(url.searchParams.get('fields')).toBe( - 'kind,nextPageToken,users(id,primaryEmail,customerId,suspended,archived,isGuestUser)' + 'kind,nextPageToken,users(id,primaryEmail,customerId,suspended,archived,isGuestUser,isMailboxSetup)' ) }) @@ -410,6 +416,208 @@ describe('Google Workspace per-user central crawl', () => { await list(ctx) expect(ctx.listingCapped).toBe(true) }) + + it('continues after an unavailable Gmail mailbox and persists the failure across fresh workers', async () => { + listUserDocuments.mockRejectedValueOnce( + new GoogleApiError('gmail.threads.list', 400, ['failedPrecondition']) + ) + const first = await list(context()) + expect(first).toMatchObject({ + documents: [], + hasMore: true, + reconciliationSafe: false, + listingFailures: { + count: 1, + samples: [ + { + scope: 'alice@corp.com', + operation: 'gmail.threads.list', + status: 400, + reasons: ['failedPrecondition'], + }, + ], + }, + }) + const second = await list(context(), first.nextCursor) + expect(second.documents[0].acl).toEqual(['u:bob@corp.com']) + expect(second).toMatchObject({ + hasMore: false, + reconciliationSafe: false, + listingFailures: first.listingFailures, + }) + expect(listUserDocuments.mock.calls[1][0]).toBe('delegated:bob@corp.com') + }) + + it('replays a skipped user without double-counting and retries it on the next generation', async () => { + const unavailable = new GoogleApiError('gmail.threads.list', 400, ['failedPrecondition']) + listUserDocuments.mockRejectedValueOnce(unavailable).mockRejectedValueOnce(unavailable) + const first = await list(context()) + const replay = await list(context(), first.currentCursor) + expect(replay.nextCursor).toBe(first.nextCursor) + expect(replay.listingFailures?.count).toBe(1) + const nextGeneration = await list(context()) + expect(nextGeneration.documents[0].acl).toEqual(['u:alice@corp.com']) + expect(nextGeneration.listingFailures).toBeUndefined() + }) + + it('preserves failure evidence when advancing to another Directory page', async () => { + mockFetch.mockResolvedValueOnce(json({ users: [USER('alice')], nextPageToken: 'directory-2' })) + listUserDocuments.mockRejectedValueOnce( + new GoogleApiError('gmail.threads.list', 400, ['failedPrecondition']) + ) + const first = await list(context()) + mockFetch.mockResolvedValueOnce(json({ users: [USER('bob')] })) + const second = await list(context(), first.nextCursor) + expect(second.documents[0].acl).toEqual(['u:bob@corp.com']) + expect(second.listingFailures).toEqual(first.listingFailures) + expect(second.reconciliationSafe).toBe(false) + }) + + it.each([true, undefined])( + 'keeps Gmail users eligible when mailbox metadata is %s', + async (isMailboxSetup) => { + directory([USER('alice', undefined, { isMailboxSetup })]) + expect((await list(context())).documents).toHaveLength(1) + expect(listUserDocuments).toHaveBeenCalledOnce() + } + ) + + it('skips an explicitly unprovisioned Gmail mailbox before requesting a token', async () => { + directory([USER('alice', undefined, { isMailboxSetup: false }), USER('bob')]) + const ctx = context() + const first = await list(ctx) + expect(first.listingFailures?.samples[0]).toEqual({ + scope: 'alice@corp.com', + operation: 'directory.users.get', + reasons: ['mailboxNotSetup'], + }) + expect(ctx.getDelegatedAccessToken).not.toHaveBeenCalled() + const second = await list(context(), first.nextCursor) + expect(second.documents[0].acl).toEqual(['u:bob@corp.com']) + }) + + it('does not use Gmail mailbox eligibility for Calendar', async () => { + directory([USER('alice', undefined, { isMailboxSetup: false })]) + expect((await list(context(), undefined, CONFIG, 'google_calendar')).documents).toHaveLength(1) + }) + + it.each([{ reasons: [] }, { reasons: ['forbidden'] }])( + 'isolates Calendar list access failures without claiming a disabled service (%j)', + async ({ reasons }) => { + listUserDocuments.mockRejectedValueOnce( + new GoogleApiError('calendar.events.list', 403, reasons) + ) + const first = await list(context(), undefined, CONFIG, 'google_calendar') + expect(first.listingFailures?.samples[0]).toEqual({ + scope: 'alice@corp.com', + operation: 'calendar.events.list', + status: 403, + reasons, + }) + const second = await list(context(), first.nextCursor, CONFIG, 'google_calendar') + expect(second.documents[0].acl).toEqual(['u:bob@corp.com']) + expect(second.reconciliationSafe).toBe(false) + } + ) + + it.each([ + [403, ['rateLimitExceeded']], + [403, ['userRateLimitExceeded']], + [403, ['quotaExceeded']], + [403, ['insufficientPermissions']], + [403, ['ACCESS_TOKEN_SCOPE_INSUFFICIENT']], + [403, ['SERVICE_DISABLED']], + [403, ['domainPolicy']], + [403, ['unrecognized-provider-code']], + [403, ['forbidden', 'unrecognized-provider-code']], + [401, ['authError']], + [429, []], + [500, ['backendError']], + ] as const)( + 'does not skip global or retryable Calendar errors (%s %j)', + async (status, reasons) => { + const error = new GoogleApiError('calendar.events.list', status, reasons) + listUserDocuments.mockRejectedValueOnce(error) + await expect(list(context(), undefined, CONFIG, 'google_calendar')).rejects.toBe(error) + } + ) + + it.each([ + new GoogleApiError('gmail.threads.list', 400, ['badRequest']), + new GoogleApiError('gmail.threads.list', 400, []), + new GoogleApiError('gmail.labels.list', 400, ['failedPrecondition']), + new GoogleApiError('calendar.calendarList.list', 403, ['forbidden']), + new Error('unknown provider failure'), + ])('does not suppress unclassified failures: %s', async (error) => { + listUserDocuments.mockRejectedValueOnce(error) + await expect(list(context())).rejects.toBe(error) + }) + + it('does not isolate an unreadable Calendar error envelope', async () => { + const error = new GoogleApiError('calendar.events.list', 403, [], false) + listUserDocuments.mockRejectedValueOnce(error) + await expect(list(context(), undefined, CONFIG, 'google_calendar')).rejects.toBe(error) + }) + + it('does not suppress delegation failures that resemble provider list failures', async () => { + const ctx = context() + const error = new GoogleApiError('gmail.threads.list', 400, ['failedPrecondition']) + ctx.getDelegatedAccessToken.mockRejectedValueOnce(error) + await expect(list(ctx)).rejects.toBe(error) + expect(listUserDocuments).not.toHaveBeenCalled() + }) + + it('honors cancellation before recording an otherwise isolatable error', async () => { + const controller = new AbortController() + listUserDocuments.mockImplementationOnce(async () => { + controller.abort() + throw new GoogleApiError('gmail.threads.list', 400, ['failedPrecondition']) + }) + await expect(list({ ...context(), signal: controller.signal })).rejects.toMatchObject({ + name: 'AbortError', + }) + }) + + it('revokes the prior page hydration authority when the next user fails', async () => { + const ctx = context() + const first = await list(ctx) + listUserDocuments.mockRejectedValueOnce( + new GoogleApiError('gmail.threads.list', 400, ['failedPrecondition']) + ) + await list(ctx, first.nextCursor) + const hydrate = vi.fn() + await expect( + getGoogleWorkspaceDocument({ + provider: 'gmail', + sourceConfig: CONFIG, + externalId: first.documents[0].externalId, + syncContext: ctx, + getUserDocument: hydrate, + }) + ).rejects.toThrow('verified delegated listing identity') + expect(hydrate).not.toHaveBeenCalled() + }) + + it('bounds retained failure samples while counting every unavailable user', async () => { + directory( + Array.from({ length: 15 }, (_, index) => + USER(`user-${index}`, undefined, { isMailboxSetup: false }) + ) + ) + let cursor: string | undefined + let final + for (let i = 0; i < 15; i++) { + final = await list(context(), cursor) + cursor = final.nextCursor + } + expect(final).toMatchObject({ + hasMore: false, + listingFailures: { count: 15 }, + reconciliationSafe: false, + }) + expect(final?.listingFailures?.samples).toHaveLength(10) + expect(listUserDocuments).not.toHaveBeenCalled() + }) }) describe('Google Workspace central validation', () => { diff --git a/apps/sim/connectors/google-workspace/company-crawl.ts b/apps/sim/connectors/google-workspace/company-crawl.ts index cb7a3c7b0b1..f0a07992f8e 100644 --- a/apps/sim/connectors/google-workspace/company-crawl.ts +++ b/apps/sim/connectors/google-workspace/company-crawl.ts @@ -1,6 +1,8 @@ +import { createLogger } from '@sim/logger' import { normalizeEmail } from '@sim/utils/string' import { z } from 'zod' import { mapWithConcurrency } from '@/lib/core/utils/concurrency' +import { GoogleApiError } from '@/connectors/google-workspace/api-errors' import { GOOGLE_WORKSPACE_USERS_PAGE_SIZE, type GoogleWorkspaceUser, @@ -8,11 +10,18 @@ import { listGoogleWorkspaceUsers, selectedGoogleWorkspaceUsers, } from '@/connectors/google-workspace/users' +import { listingFailuresSchema, MAX_LISTING_FAILURE_SAMPLES } from '@/connectors/listing-failures' import { ConnectorSourceError } from '@/connectors/source-error' -import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' +import type { + ConnectorConfig, + ExternalDocument, + ExternalDocumentList, + ExternalListingFailures, +} from '@/connectors/types' import { PER_MEMBER_LISTING_CONTEXT, sourceDocumentId } from '@/connectors/utils' type GoogleWorkspaceProvider = 'gmail' | 'google_calendar' +const logger = createLogger('GoogleWorkspaceCrawl') const CURSOR_PREFIX = 'google-workspace:v1:' const MAX_CURSOR_BYTES = 384 * 1024 const MAX_PROVIDER_CURSOR_BYTES = 256 * 1024 @@ -30,6 +39,7 @@ const cursorSchema = z.object({ .max(GOOGLE_WORKSPACE_USERS_PAGE_SIZE), nextUsersPageToken: z.string().min(1).max(8192).optional(), providerCursor: z.string().min(1).max(MAX_PROVIDER_CURSOR_BYTES).optional(), + listingFailures: listingFailuresSchema.optional(), }) type CompanyCursor = z.infer @@ -128,6 +138,27 @@ function ownerDocument(document: ExternalDocument, access: DelegatedUser): Exter return { ...document, acl: [`u:${access.user.email}`] } } +/** Isolates narrow user-list failures; delegation, known scope errors and quota errors still fail. */ +function userListingFailure( + error: unknown, + provider: GoogleWorkspaceProvider +): Omit | null { + if (!(error instanceof GoogleApiError) || !error.diagnostic || !error.reasonsComplete) return null + const reasons = error.diagnostic.reasons + const isolated = + provider === 'gmail' + ? error.diagnostic.operation === 'gmail.threads.list' && + error.status === 400 && + reasons.length > 0 && + reasons.every((reason) => reason === 'failedPrecondition') + : error.diagnostic.operation === 'calendar.events.list' && + error.status === 403 && + reasons.every((reason) => reason === 'forbidden') + return isolated + ? { operation: error.diagnostic.operation, status: error.status, reasons: [...reasons] } + : null +} + /** Validates directory access and returns one revalidated identity for a provider-specific probe. */ export async function validateGoogleWorkspaceConfig( input: GoogleWorkspaceCrawlInput @@ -217,6 +248,7 @@ export async function listGoogleWorkspaceDocuments( .filter((user) => user.active && (!selected.length || selected.includes(user.email))) .map(({ id, email, customerId }) => ({ id, email, customerId })), nextUsersPageToken: page.nextPageToken, + listingFailures: state.listingFailures, } } const currentCursor = writeCursor(state) @@ -225,6 +257,7 @@ export async function listGoogleWorkspaceDocuments( provider, users: state.users.slice(1), nextUsersPageToken: state.nextUsersPageToken, + listingFailures: state.listingFailures, }) const emptyPage = (next: CompanyCursor): ExternalDocumentList => { const hasMore = Boolean(next.users.length || next.nextUsersPageToken) @@ -233,6 +266,10 @@ export async function listGoogleWorkspaceDocuments( currentCursor, hasMore, nextCursor: hasMore ? writeCursor(next) : undefined, + ...(next.listingFailures && { + listingFailures: next.listingFailures, + reconciliationSafe: false, + }), } } if (!pending) return emptyPage(state) @@ -240,6 +277,25 @@ export async function listGoogleWorkspaceDocuments( if (user) assertIdentity(user, pending) if (!user?.active || (selected.length && !selected.includes(user.email))) return emptyPage(advance()) + const failedUser = ( + failure: Omit + ): ExternalDocumentList => { + const sample = { scope: user.email, ...failure } + const previous = state.listingFailures + state.listingFailures = { + count: (previous?.count ?? 0) + 1, + samples: [...(previous?.samples ?? []), sample].slice(0, MAX_LISTING_FAILURE_SAMPLES), + } + syncContext.reconciliationUnsafe = true + logger.warn('Google Workspace user could not be listed; continuing other users', { + provider, + ...sample, + }) + return emptyPage(advance()) + } + if (provider === 'gmail' && user.isMailboxSetup === false) { + return failedUser({ operation: 'directory.users.get', reasons: ['mailboxNotSetup'] }) + } const access: PageAccess = { provider, user, @@ -254,12 +310,20 @@ export async function listGoogleWorkspaceDocuments( externalIds: new Set(), } access.syncContext.signal = signal - const page = await listUserDocuments( - access.accessToken, - sourceConfig, - state.providerCursor, - access.syncContext - ) + let page: ExternalDocumentList + try { + page = await listUserDocuments( + access.accessToken, + sourceConfig, + state.providerCursor, + access.syncContext + ) + } catch (error) { + signal?.throwIfAborted() + const failure = userListingFailure(error, provider) + if (!failure) throw error + return failedUser(failure) + } signal?.throwIfAborted() if (access.syncContext.listingCapped === true) syncContext.listingCapped = true if (page.documents.length > MAX_PAGE_DOCUMENTS) @@ -277,6 +341,10 @@ export async function listGoogleWorkspaceDocuments( currentCursor: writeCursor(replay), hasMore, nextCursor: hasMore ? writeCursor(next) : undefined, + ...(state.listingFailures && { + listingFailures: state.listingFailures, + reconciliationSafe: false, + }), } access.externalIds = new Set(documents.map((document) => document.externalId)) pageAccess.set(syncContext, access) diff --git a/apps/sim/connectors/google-workspace/users.ts b/apps/sim/connectors/google-workspace/users.ts index f5c2baf3204..49f67dbeca8 100644 --- a/apps/sim/connectors/google-workspace/users.ts +++ b/apps/sim/connectors/google-workspace/users.ts @@ -18,6 +18,7 @@ export interface GoogleWorkspaceUser { email: string customerId: string active: boolean + isMailboxSetup?: boolean } export interface GoogleWorkspaceUserPage { @@ -58,7 +59,8 @@ function parseUser(value: unknown): GoogleWorkspaceUser { !emailSchema.safeParse(value.primaryEmail).success || typeof value.suspended !== 'boolean' || (value.archived !== undefined && typeof value.archived !== 'boolean') || - (value.isGuestUser !== undefined && typeof value.isGuestUser !== 'boolean') + (value.isGuestUser !== undefined && typeof value.isGuestUser !== 'boolean') || + (value.isMailboxSetup !== undefined && typeof value.isMailboxSetup !== 'boolean') ) { throw new Error('Google Workspace returned malformed user metadata') } @@ -67,6 +69,7 @@ function parseUser(value: unknown): GoogleWorkspaceUser { email: normalizeEmail(value.primaryEmail), customerId: value.customerId, active: !value.suspended && value.archived !== true && value.isGuestUser !== true, + ...(typeof value.isMailboxSetup === 'boolean' && { isMailboxSetup: value.isMailboxSetup }), } } @@ -80,7 +83,7 @@ async function readDirectoryJson(response: Response): Promise { } } -const USER_FIELDS = 'id,primaryEmail,customerId,suspended,archived,isGuestUser' +const USER_FIELDS = 'id,primaryEmail,customerId,suspended,archived,isGuestUser,isMailboxSetup' /** One provider page only; the connector checkpoint advances through the directory. */ export async function listGoogleWorkspaceUsers( diff --git a/apps/sim/connectors/listing-failures.ts b/apps/sim/connectors/listing-failures.ts new file mode 100644 index 00000000000..d64e85bfd82 --- /dev/null +++ b/apps/sim/connectors/listing-failures.ts @@ -0,0 +1,18 @@ +import { z } from 'zod' + +export const MAX_LISTING_FAILURE_SAMPLES = 10 + +/** Persist only bounded scope identifiers and provider-owned codes, never raw errors or content. */ +export const listingFailuresSchema = z.object({ + count: z.number().int().positive().max(Number.MAX_SAFE_INTEGER), + samples: z + .array( + z.object({ + scope: z.string().min(1).max(254), + operation: z.string().min(1).max(96), + status: z.number().int().min(100).max(599).optional(), + reasons: z.array(z.string().min(1).max(64)).max(16), + }) + ) + .max(MAX_LISTING_FAILURE_SAMPLES), +}) diff --git a/apps/sim/connectors/source-error.ts b/apps/sim/connectors/source-error.ts index 09e0c854e7f..37e28189674 100644 --- a/apps/sim/connectors/source-error.ts +++ b/apps/sim/connectors/source-error.ts @@ -21,3 +21,6 @@ export class ConnectorSourceError extends Error { /** Keeps directory failures distinct from document-content failures through cause wrapping. */ export class ConnectorDirectoryError extends Error {} + +/** A provider-confirmed inaccessible external group, rather than a directory-wide failure. */ +export class ConnectorDirectoryGroupAccessError extends ConnectorDirectoryError {} diff --git a/apps/sim/connectors/types.ts b/apps/sim/connectors/types.ts index 42636ada4f2..9586bad3c79 100644 --- a/apps/sim/connectors/types.ts +++ b/apps/sim/connectors/types.ts @@ -182,9 +182,18 @@ export interface ExternalDocument { metadata?: Record } -/** - * Paginated result from listing documents in an external source. - */ +/** Bounded provider evidence for scopes that could not be fully listed. */ +export interface ExternalListingFailures { + count: number + samples: { + scope: string + operation: string + status?: number + reasons: string[] + }[] +} + +/** Paginated result from listing documents in an external source. */ export interface ExternalDocumentList { documents: ExternalDocument[] nextCursor?: string @@ -197,6 +206,8 @@ export interface ExternalDocumentList { * provider pagination must set this to false. */ reconciliationSafe?: boolean + /** Cumulative, bounded failure evidence for this listing generation; replay must not add it twice. */ + listingFailures?: ExternalListingFailures } /** diff --git a/apps/sim/lib/knowledge/__integration__/embedding-insert-batches.integration.ts b/apps/sim/lib/knowledge/__integration__/embedding-insert-batches.integration.ts new file mode 100644 index 00000000000..0761840834d --- /dev/null +++ b/apps/sim/lib/knowledge/__integration__/embedding-insert-batches.integration.ts @@ -0,0 +1,191 @@ +/** Real transaction rollback across embedding, search projections, provenance, and completion state. */ +import { mkdtempSync } from 'node:fs' +import { rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import path from 'node:path' +import { db } from '@sim/db' +import { + document, + embedding, + embeddingSecretProvenance, + knowledgeBase, + organization, + user, + workspace, +} from '@sim/db/schema' +import { generateId } from '@sim/utils/id' +import { eq, inArray } from 'drizzle-orm' +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest' + +const fixtures = vi.hoisted(() => ({ root: '', process: vi.fn(), embeddings: vi.fn() })) +vi.mock('@/lib/uploads/core/setup.server', () => ({ + get UPLOAD_DIR_SERVER() { + return fixtures.root + }, +})) +vi.mock('@/lib/knowledge/documents/document-processor', () => ({ + processDocument: fixtures.process, +})) +vi.mock('@/lib/knowledge/embeddings', () => ({ generateEmbeddings: fixtures.embeddings })) + +import { resolveBillingAttribution } from '@/lib/billing/core/billing-attribution' +import * as embeddingClient from '@/lib/embeddings/client' +import { + createKnowledgeAclFixtureIds, + seedKnowledgeAclFixture, +} from '@/lib/knowledge/__integration__/seed-source-access-fixture' +import { createContentSyncLease } from '@/lib/knowledge/connectors/sync-lock' +import { addDocument } from '@/lib/knowledge/connectors/sync-persistence' +import { processDocumentAsync } from '@/lib/knowledge/documents/service' +import { + createKnowledgeDocumentSourceValue, + replaceKnowledgeDocumentSecretProvenanceInTx, +} from '@/lib/knowledge/secret-provenance' + +describe('bounded embedding insert transactions', () => { + const ids = createKnowledgeAclFixtureIds() + const triggerName = `fixture_embedding_batch_${generateId().replaceAll('-', '')}` + + beforeAll(async () => { + fixtures.root = mkdtempSync(path.join(tmpdir(), 'sim-embedding-batches-')) + await seedKnowledgeAclFixture(ids, { connectorType: 'google_drive' }) + vi.spyOn(embeddingClient, 'assertKnowledgeEmbeddingCapacity').mockResolvedValue(undefined) + }) + + afterAll(async () => { + vi.restoreAllMocks() + await db.$client.unsafe(`DROP TRIGGER IF EXISTS ${triggerName} ON embedding`) + await db.$client.unsafe(`DROP FUNCTION IF EXISTS ${triggerName}()`) + await db.delete(knowledgeBase).where(eq(knowledgeBase.id, ids.knowledgeBaseId)) + 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])) + await rm(fixtures.root, { recursive: true, force: true }) + await db.$client.end() + }) + + it('rolls back a second-batch failure and replaces the complete index on retry', async () => { + const file = await addDocument( + ids.knowledgeBaseId, + ids.connectorId, + 'google_drive', + { + externalId: 'atomic-batch-fixture', + title: 'Synthetic batch fixture.txt', + content: 'Synthetic replacement text', + mimeType: 'text/plain', + contentHash: 'synthetic-replacement', + }, + { userId: ids.aliceId, workspaceId: ids.workspaceId }, + undefined, + 'workspace', + createContentSyncLease(ids.connectorId, ids.lockId) + ) + const [source] = await db.select().from(document).where(eq(document.id, file.documentId)) + await db.transaction((tx) => + replaceKnowledgeDocumentSecretProvenanceInTx( + tx, + file.documentId, + createKnowledgeDocumentSourceValue(source), + { status: 'exact', entries: [] } + ) + ) + const previousId = generateId() + await db.insert(embedding).values({ + id: previousId, + knowledgeBaseId: ids.knowledgeBaseId, + documentId: file.documentId, + chunkIndex: 0, + chunkHash: 'previous-hash', + content: 'Previously indexed text', + contentLength: 23, + tokenCount: 5, + startOffset: 0, + endOffset: 23, + embedding: Array(1536).fill(0.1), + secretProvenanceVersion: 1, + }) + await db.insert(embeddingSecretProvenance).values({ + embeddingId: previousId, + contentHash: 'previous-hash', + status: 'exact', + entries: [], + }) + await db.update(document).set({ chunkCount: 1 }).where(eq(document.id, file.documentId)) + + const chunks = Array.from({ length: 205 }, (_, index) => ({ + text: `Synthetic chunk ${index}`, + metadata: { startIndex: index * 20, endIndex: index * 20 + 19 }, + })) + fixtures.process.mockResolvedValue({ + chunks, + metadata: { chunkCount: chunks.length, tokenCount: 615, characterCount: 4100 }, + }) + fixtures.embeddings.mockResolvedValue({ + embeddings: chunks.map(() => Array(1536).fill(0.2)), + billableTokens: 0, + modelName: 'text-embedding-3-small', + pricingId: 'text-embedding-3-small', + }) + const billing = await resolveBillingAttribution({ + actorUserId: ids.aliceId, + workspaceId: ids.workspaceId, + }) + await db.$client.unsafe(`CREATE FUNCTION ${triggerName}() RETURNS trigger LANGUAGE plpgsql AS $$ + BEGIN + IF NEW.document_id = '${file.documentId}' AND NEW.chunk_index >= 100 THEN + RAISE EXCEPTION 'Synthetic second-batch failure' USING ERRCODE = '57014'; + END IF; + RETURN NEW; + END; + $$`) + await db.$client.unsafe(`CREATE TRIGGER ${triggerName} BEFORE INSERT ON embedding + FOR EACH ROW EXECUTE FUNCTION ${triggerName}()`) + + await expect( + processDocumentAsync(ids.knowledgeBaseId, file.documentId, file, {}, billing) + ).rejects.toMatchObject({ cause: { code: '57014' } }) + expect( + await db + .select({ id: embedding.id }) + .from(embedding) + .where(eq(embedding.documentId, file.documentId)) + ).toEqual([{ id: previousId }]) + for (const table of ['embedding_search', 'embedding_keyword_search']) { + expect( + await db.$client.unsafe(`SELECT id FROM ${table} WHERE document_id = $1`, [file.documentId]) + ).toEqual([{ id: previousId }]) + } + expect( + await db + .select() + .from(embeddingSecretProvenance) + .where(eq(embeddingSecretProvenance.embeddingId, previousId)) + ).toMatchObject([{ contentHash: 'previous-hash', status: 'exact' }]) + expect(await db.select().from(document).where(eq(document.id, file.documentId))).toMatchObject([ + { + processingStatus: 'failed', + chunkCount: 1, + processingError: 'Database request failed (SQLSTATE 57014).', + }, + ]) + + await db.$client.unsafe(`DROP TRIGGER ${triggerName} ON embedding`) + await processDocumentAsync(ids.knowledgeBaseId, file.documentId, file, {}, billing) + expect(await db.select().from(document).where(eq(document.id, file.documentId))).toMatchObject([ + { processingStatus: 'completed', chunkCount: 205, processingError: null }, + ]) + for (const table of ['embedding', 'embedding_search', 'embedding_keyword_search']) { + expect( + await db.$client.unsafe( + `SELECT count(*)::int AS count FROM ${table} WHERE document_id = $1`, + [file.documentId] + ) + ).toEqual([{ count: 205 }]) + } + expect( + await db.$client`SELECT count(*)::int AS count FROM embedding_secret_provenance p + INNER JOIN embedding e ON e.id = p.embedding_id WHERE e.document_id = ${file.documentId}` + ).toEqual([{ count: 205 }]) + }) +}) diff --git a/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts b/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts index 5369a251407..85ee6179525 100644 --- a/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts +++ b/apps/sim/lib/knowledge/connectors/external-group-sync.test.ts @@ -5,6 +5,7 @@ import { dbChainMockFns, queueTableRows, resetDbChainMock, schemaMock } from '@s import { beforeEach, describe, expect, it, vi } from 'vitest' import { getConnectorFailureDiagnostic } from '@/lib/knowledge/connectors/connector-error' import { GoogleDriveApiError } from '@/connectors/google-drive/google-drive-errors' +import { ConnectorDirectoryGroupAccessError } from '@/connectors/source-error' import type { ConnectorDirectory } from '@/connectors/types' const { mockResolveTokenUserId, mockResolveToken, mockOpenDirectory, mockAvailability } = @@ -36,6 +37,9 @@ vi.mock('@/connectors/registry.server', () => ({ })) import { + DIRECTORY_ERROR_PREFIX, + DIRECTORY_WARNING_PREFIX, + directorySyncNotice, refreshConnectorDirectory, refreshMirroredDirectory, syncExternalDirectoryGroups, @@ -149,6 +153,30 @@ describe('syncExternalDirectoryGroups', () => { expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastSyncedAt' in value)).toBe(false) }) + it('does not change memberships or directory freshness after losing its lease', async () => { + dbChainMockFns.returning.mockResolvedValueOnce([{ id: 'lease' }]).mockResolvedValueOnce([]) + await expect( + syncExternalDirectoryGroups({ workspaceId: 'ws-1', directory: directory() }) + ).rejects.toThrow('Directory sync lease expired or was replaced') + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastSyncedAt' in value)).toBe(false) + expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastCompleteSyncAt' in value)).toBe( + false + ) + }) + + it('propagates storage failures without pruning or refreshing memberships', async () => { + dbChainMockFns.transaction.mockRejectedValueOnce(new Error('Storage write failed')) + await expect( + syncExternalDirectoryGroups({ workspaceId: 'ws-1', directory: directory() }) + ).rejects.toThrow('Storage write failed') + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastSyncedAt' in value)).toBe(false) + expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastCompleteSyncAt' in value)).toBe( + false + ) + }) + /** * A truncated group listing must not prune: every directory throws rather * than returning a partial page, and the sync fails with it. @@ -301,6 +329,99 @@ describe('refreshConnectorDirectory', () => { expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastSyncedAt' in value)).toBe(false) }) + it('reports inaccessible external groups as partial without refreshing their access', async () => { + queueTableRows(schemaMock.knowledgeConnector, [ + connectorRow({ lastSyncError: 'Content failed' }), + ]) + const providerError = new GoogleDriveApiError(403, ['forbidden'], 'directory.members.list') + mockOpenDirectory.mockResolvedValue( + directory({ + listGroupMembers: vi.fn(async (group) => { + if (group.id === 'all@corp.com') { + throw new ConnectorDirectoryGroupAccessError('External group denied', { + cause: providerError, + }) + } + return { group, memberTokens: ['u:alice@corp.com'], complete: true } + }), + }) + ) + + await expect(refreshConnectorDirectory('connector-1', 'req-1')).resolves.toBe('partial') + const notice = dbChainMockFns.set.mock.calls.find(([value]) => 'lastSyncError' in value)?.[0] + expect(notice.lastSyncError).toContain(`${DIRECTORY_WARNING_PREFIX}1 group memberships`) + expect(notice.lastSyncError).toContain('\nContent failed') + expect(dbChainMockFns.set.mock.calls.filter(([value]) => 'lastSyncedAt' in value)).toHaveLength( + 1 + ) + expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastCompleteSyncAt' in value)).toBe( + false + ) + expect(dbChainMockFns.delete).toHaveBeenCalledTimes(1) + }) + + it('keeps unknown failures blocking even when other group memberships refreshed', async () => { + queueTableRows(schemaMock.knowledgeConnector, [connectorRow()]) + mockOpenDirectory.mockResolvedValue( + directory({ + listGroupMembers: vi.fn(async (group) => { + if (group.id === 'all@corp.com') throw new Error('Connection closed') + return { group, memberTokens: ['u:alice@corp.com'], complete: true } + }), + }) + ) + await expect(refreshConnectorDirectory('connector-1', 'req-1')).rejects.toThrow( + '1 group memberships could not be refreshed' + ) + }) + + it('keeps a directory with no successful memberships blocking', async () => { + queueTableRows(schemaMock.knowledgeConnector, [connectorRow()]) + mockOpenDirectory.mockResolvedValue( + directory({ + listGroupMembers: vi.fn().mockRejectedValue( + new ConnectorDirectoryGroupAccessError('External group denied', { + cause: new GoogleDriveApiError(403, ['forbidden'], 'directory.members.list'), + }) + ), + }) + ) + await expect(refreshConnectorDirectory('connector-1', 'req-1')).rejects.toThrow( + '2 group memberships could not be refreshed' + ) + expect(dbChainMockFns.delete).not.toHaveBeenCalled() + expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastSyncedAt' in value)).toBe(false) + }) + + it('reports a blocking error when an inaccessible external group failed first', async () => { + queueTableRows(schemaMock.knowledgeConnector, [connectorRow()]) + mockOpenDirectory.mockResolvedValue( + directory({ + listGroupMembers: vi.fn(async (group) => { + if (group.id === 'eng@corp.com') { + throw new ConnectorDirectoryGroupAccessError('External group denied', { + cause: new GoogleDriveApiError(403, ['forbidden'], 'directory.members.list'), + }) + } + throw new GoogleDriveApiError(500, ['backendError'], 'directory.members.list') + }), + }) + ) + const failure = await refreshConnectorDirectory('connector-1', 'req-1').catch( + (error: unknown) => error + ) + expect(getConnectorFailureDiagnostic(failure)).toMatchObject({ + status: 500, + operation: 'directory.members.list', + reasons: ['backendError'], + phase: 'directory', + }) + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ lastSyncError: expect.stringContaining('HTTP 500') }) + ) + expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastSyncedAt' in value)).toBe(false) + }) + it('clears a previous directory error after a successful refresh', async () => { queueTableRows(schemaMock.knowledgeConnector, [ connectorRow({ lastSyncError: 'Directory refresh failed: 403' }), @@ -312,6 +433,35 @@ describe('refreshConnectorDirectory', () => { ) }) + it('clears only the recovered directory warning while preserving other diagnostics', async () => { + queueTableRows(schemaMock.knowledgeConnector, [ + connectorRow({ + lastSyncError: `${DIRECTORY_WARNING_PREFIX}1 group memberships could not be verified\nContent failed`, + }), + ]) + mockOpenDirectory.mockResolvedValue(directory({ listGroups: vi.fn().mockResolvedValue([]) })) + await expect(refreshConnectorDirectory('connector-1', 'req-1')).resolves.toBe('refreshed') + expect(dbChainMockFns.set).toHaveBeenCalledWith( + expect.objectContaining({ lastSyncError: 'Content failed' }) + ) + }) + + it.each([DIRECTORY_WARNING_PREFIX, DIRECTORY_ERROR_PREFIX])( + 'preserves the previous directory notice when another refresh owns the lease: %s', + async (prefix) => { + const warning = `${prefix}1 group memberships could not be verified` + queueTableRows(schemaMock.knowledgeConnector, [ + connectorRow({ lastSyncError: `${warning}\nContent failed` }), + ]) + dbChainMockFns.returning.mockResolvedValueOnce([]) + mockOpenDirectory.mockResolvedValue(directory()) + await expect(refreshConnectorDirectory('connector-1', 'req-1')).resolves.toBe('skipped') + expect(dbChainMockFns.set.mock.calls.some(([value]) => 'lastSyncError' in value)).toBe(false) + expect(directorySyncNotice(`${warning}\nContent failed`)).toBe(warning) + expect(directorySyncNotice('Content failed')).toBeNull() + } + ) + it('preserves provider retry metadata through the directory failure cause', async () => { const { getRetryAfterMs, isRateLimitError } = await import('@/lib/knowledge/documents/utils') const providerError = Object.assign(new Error('quota'), { status: 429, retryAfterMs: 60_000 }) diff --git a/apps/sim/lib/knowledge/connectors/external-group-sync.ts b/apps/sim/lib/knowledge/connectors/external-group-sync.ts index 5ea76ba69c0..be56c0abe40 100644 --- a/apps/sim/lib/knowledge/connectors/external-group-sync.ts +++ b/apps/sim/lib/knowledge/connectors/external-group-sync.ts @@ -32,7 +32,10 @@ import { getConnectorFailureDiagnostic } from '@/lib/knowledge/connectors/connec import { RUNNABLE_CONNECTOR_STATUSES } from '@/lib/knowledge/connectors/sync-lock' import { isRateLimitError } from '@/lib/knowledge/documents/utils' import { CONNECTOR_REGISTRY } from '@/connectors/registry.server' -import { ConnectorDirectoryError } from '@/connectors/source-error' +import { + ConnectorDirectoryError, + ConnectorDirectoryGroupAccessError, +} from '@/connectors/source-error' import type { ConnectorConfig, ConnectorDirectory, @@ -45,6 +48,44 @@ const logger = createLogger('ExternalGroupSync') /** Member rows written per statement while replacing a group's membership. */ const MEMBER_WRITE_BATCH_SIZE = 500 export const DIRECTORY_ERROR_PREFIX = 'Directory refresh failed: ' +export const DIRECTORY_WARNING_PREFIX = 'Directory refresh incomplete: ' + +export type DirectoryRefreshResult = + | { status: 'refreshed' | 'skipped' } + | { status: 'partial'; notice: string } + +/** Directory workers replace only their notice, preserving content-sync diagnostics. */ +export function replaceDirectorySyncNotice(previous: string | null, notice: string | null) { + const contentNotices = (previous ?? '') + .split('\n') + .filter( + (line) => + line && + !line.startsWith(DIRECTORY_ERROR_PREFIX) && + !line.startsWith(DIRECTORY_WARNING_PREFIX) + ) + return [notice, ...contentNotices].filter(Boolean).join('\n') || null +} + +export function hasDirectorySyncNotice(notice: string | null | undefined): boolean { + return (notice ?? '') + .split('\n') + .some( + (line) => line.startsWith(DIRECTORY_ERROR_PREFIX) || line.startsWith(DIRECTORY_WARNING_PREFIX) + ) +} + +/** A skipped lease has not established that an earlier directory failure recovered. */ +export function directorySyncNotice(notice: string | null | undefined): string | null { + return ( + (notice ?? '') + .split('\n') + .find( + (line) => + line.startsWith(DIRECTORY_ERROR_PREFIX) || line.startsWith(DIRECTORY_WARNING_PREFIX) + ) ?? null + ) +} interface DirectorySyncResult { /** Groups whose membership was replaced from a complete enumeration. */ @@ -55,6 +96,8 @@ interface DirectorySyncResult { pruned: number /** The directory is already complete and fresh, or another worker holds its lease. */ skipped: boolean + /** Retained groups whose provider identified inaccessible external membership. */ + inaccessible: number error?: Error } @@ -166,7 +209,7 @@ export async function syncExternalDirectoryGroups(input: { const owner = resourceScopeFields(resourceScopeFromOwner(input)) const { providerId, tenantId } = directory const lease = await claimDirectory({ ...owner, providerId, tenantId }, Boolean(input.force)) - if (!lease) return { refreshed: 0, keptStale: 0, pruned: 0, skipped: true } + if (!lease) return { refreshed: 0, keptStale: 0, pruned: 0, skipped: true, inaccessible: 0 } try { const groups = await directory.listGroups() @@ -178,7 +221,9 @@ export async function syncExternalDirectoryGroups(input: { }) let refreshed = 0 let keptStale = 0 + let inaccessible = 0 let firstError: Error | undefined + let blockingError: Error | undefined for (const group of groups) { const groupId = await withDirectoryLease(lease, (tx) => upsertGroup({ ...owner, providerId, tenantId, group }, tx) @@ -189,6 +234,8 @@ export async function syncExternalDirectoryGroups(input: { } catch (error) { if (isRateLimitError(error)) throw error keptStale += 1 + if (error instanceof ConnectorDirectoryGroupAccessError) inaccessible += 1 + else blockingError ??= toError(error) firstError ??= toError(error) const diagnostic = getConnectorFailureDiagnostic(error) logger.warn('Keeping last-known-good membership for a group that failed to enumerate', { @@ -202,7 +249,7 @@ export async function syncExternalDirectoryGroups(input: { } if (!membership.complete) { keptStale += 1 - firstError ??= new Error('A group membership listing was incomplete') + blockingError ??= new Error('A group membership listing was incomplete') continue } await withDirectoryLease(lease, (tx) => @@ -225,12 +272,14 @@ export async function syncExternalDirectoryGroups(input: { }) .where(directoryIdentity(lease)) }) + const error = blockingError ?? firstError return { refreshed, keptStale, pruned, skipped: false, - ...(firstError && { error: firstError }), + inaccessible, + ...(error && { error }), } } finally { await db @@ -336,8 +385,8 @@ async function pruneRemovedGroups(lease: DirectoryLease, keep: readonly string[] * * It is rate-limited on its own clock rather than the connector's, so a * frequently-syncing connector does not re-read the whole directory every run. - * Failures retain the last confirmed membership and propagate to the caller's - * sync status and retry policy. + * Inaccessible external groups retain their previous membership and report a + * partial refresh; other failures propagate to the caller's retry policy. */ export async function refreshMirroredDirectory(input: { workspaceId?: string @@ -347,9 +396,9 @@ export async function refreshMirroredDirectory(input: { syncContext: Record accessToken: string force?: boolean -}): Promise<'refreshed' | 'skipped'> { +}): Promise { const { workspaceId, connectorConfig } = input - if (!connectorConfig.openDirectory) return 'skipped' + if (!connectorConfig.openDirectory) return { status: 'skipped' } try { const directory = await connectorConfig.openDirectory( @@ -362,7 +411,7 @@ export async function refreshMirroredDirectory(input: { workspaceId, connector: connectorConfig.id, }) - return 'skipped' + return { status: 'skipped' } } const result = await syncExternalDirectoryGroups({ ...resourceScopeFields(resourceScopeFromOwner(input)), @@ -370,6 +419,16 @@ export async function refreshMirroredDirectory(input: { force: input.force, }) if (result.keptStale > 0) { + if (result.refreshed > 0 && result.inaccessible === result.keptStale) { + const notice = `${DIRECTORY_WARNING_PREFIX}${result.keptStale} group memberships could not be verified. Previously verified access expires normally; unverified memberships grant no access. Other groups continue to sync.` + logger.warn('Directory refreshed with inaccessible external groups', { + workspaceId, + tenantId: directory.tenantId, + refreshed: result.refreshed, + keptStale: result.keptStale, + }) + return { status: 'partial', notice } + } throw new Error(`${result.keptStale} group memberships could not be refreshed`, { cause: result.error, }) @@ -379,7 +438,7 @@ export async function refreshMirroredDirectory(input: { tenantId: directory.tenantId, ...result, }) - return result.skipped ? 'skipped' : 'refreshed' + return { status: result.skipped ? 'skipped' : 'refreshed' } } catch (error) { const diagnostic = getConnectorFailureDiagnostic(error) logger.error('Directory refresh failed; serving last-known-good group membership', { @@ -394,7 +453,7 @@ export async function refreshMirroredDirectory(input: { } } -type ConnectorDirectoryRefreshOutcome = 'refreshed' | 'skipped' | 'unusable' +type ConnectorDirectoryRefreshOutcome = 'refreshed' | 'skipped' | 'partial' | 'unusable' /** * Refreshes the directory one admin-mode connector mirrors, from its row. @@ -501,19 +560,24 @@ export async function refreshConnectorDirectory( sourceConfig, syncContext, accessToken: token.accessToken, - force: connector.lastSyncError?.startsWith(DIRECTORY_ERROR_PREFIX), + force: hasDirectorySyncNotice(connector.lastSyncError), }) - if ( - outcome === 'refreshed' && - connector.lastSyncError?.startsWith(DIRECTORY_ERROR_PREFIX) + if (outcome.status === 'partial') { + await recordError(replaceDirectorySyncNotice(connector.lastSyncError, outcome.notice)) + } else if ( + outcome.status === 'refreshed' && + hasDirectorySyncNotice(connector.lastSyncError) ) { - await recordError(null) + await recordError(replaceDirectorySyncNotice(connector.lastSyncError, null)) } - return outcome + return outcome.status } catch (error) { const diagnostic = getConnectorFailureDiagnostic(error) await recordError( - diagnostic ? `${DIRECTORY_ERROR_PREFIX}${diagnostic.message}` : getErrorMessage(error) + replaceDirectorySyncNotice( + connector.lastSyncError, + diagnostic ? `${DIRECTORY_ERROR_PREFIX}${diagnostic.message}` : getErrorMessage(error) + ) ) throw error } diff --git a/apps/sim/lib/knowledge/connectors/listing-checkpoint.test.ts b/apps/sim/lib/knowledge/connectors/listing-checkpoint.test.ts index dd8f5df65e4..fede008e1a3 100644 --- a/apps/sim/lib/knowledge/connectors/listing-checkpoint.test.ts +++ b/apps/sim/lib/knowledge/connectors/listing-checkpoint.test.ts @@ -262,8 +262,120 @@ describe('durable connector listing checkpoints', () => { }) it('resumes older checkpoints without inventing permission failures', () => { - const legacy = omit(checkpoint(), ['permissionFailures']) - expect(readListingCheckpoint(legacy, fingerprint)).toMatchObject({ permissionFailures: false }) + const legacy = omit(checkpoint(), ['permissionFailures', 'listingFailures']) + expect(readListingCheckpoint(legacy, fingerprint)).toMatchObject({ + permissionFailures: false, + listingFailures: null, + }) + }) + + it('persists partial-scope failures across workers and holds deletion reconciliation at EOF', async () => { + const failures = { + count: 1, + samples: [ + { + scope: 'user@example.com', + operation: 'gmail.threads.list', + status: 400, + reasons: ['failedPrecondition'], + }, + ], + } + const first = fixture() + first.listDocuments.mockResolvedValueOnce({ + documents: [], + currentCursor: 'user-1', + nextCursor: 'user-2', + hasMore: true, + listingFailures: failures, + }) + await runResumableListing({ ...first.input, maxPages: 1 }) + const resumed = fixture(readListingCheckpoint(first.saved(), fingerprint)!) + resumed.listDocuments.mockResolvedValueOnce({ documents: [doc], hasMore: false }) + expect(await runResumableListing(resumed.input)).toMatchObject({ + complete: true, + unsafe: true, + listedCount: 1, + listingFailures: failures, + }) + }) + + it('does not double-count a replayed cumulative failure snapshot', async () => { + const failures = { + count: 2, + samples: [ + { scope: 'user@example.com', operation: 'calendar.events.list', status: 403, reasons: [] }, + ], + } + const f = fixture({ ...checkpoint(), cursor: 'user-1', listingFailures: failures }) + f.listDocuments.mockResolvedValueOnce({ + documents: [], + currentCursor: 'user-1', + nextCursor: 'user-2', + hasMore: true, + listingFailures: failures, + }) + expect(await runResumableListing({ ...f.input, maxPages: 1 })).toMatchObject({ + listingFailures: failures, + unsafe: true, + }) + expect(f.saved().listingFailures?.count).toBe(2) + }) + + it('clears failed-user evidence only when restarting the entire listing generation', async () => { + const failures = { + count: 1, + samples: [ + { + scope: 'user@example.com', + operation: 'gmail.threads.list', + status: 400, + reasons: ['failedPrecondition'], + }, + ], + } + const f = fixture({ + ...checkpoint(), + cursor: 'expired', + unsafe: true, + listingFailures: failures, + }) + const expired = new Error('cursor expired') + f.listDocuments + .mockRejectedValueOnce(expired) + .mockResolvedValueOnce({ documents: [doc], hasMore: false }) + const result = await runResumableListing({ + ...f.input, + connectorConfig: { + listDocuments: f.listDocuments, + isListingCursorInvalidError: (error) => error === expired, + }, + }) + expect(result).toMatchObject({ complete: true, unsafe: false, listingFailures: null }) + expect(result.generationId).not.toBe('cycle-1') + }) + + it('rejects unbounded or malformed failure evidence before processing a page', async () => { + const f = fixture() + const sample = { + scope: 'user@example.com', + operation: 'gmail.threads.list', + status: 400, + reasons: ['failedPrecondition'], + } + f.listDocuments.mockResolvedValueOnce({ + documents: [doc], + hasMore: false, + listingFailures: { count: 11, samples: Array(11).fill(sample) }, + }) + await expect(runResumableListing(f.input)).rejects.toThrow() + expect(f.processPage).not.toHaveBeenCalled() + expect( + readListingCheckpoint( + { ...checkpoint(), listingFailures: { count: -1, samples: [] } }, + fingerprint + ) + ).toBeNull() }) it('rejects checkpoints from a changed configuration or malformed serialized value', () => { diff --git a/apps/sim/lib/knowledge/connectors/listing-checkpoint.ts b/apps/sim/lib/knowledge/connectors/listing-checkpoint.ts index 7ca43030d82..476a8bfb86a 100644 --- a/apps/sim/lib/knowledge/connectors/listing-checkpoint.ts +++ b/apps/sim/lib/knowledge/connectors/listing-checkpoint.ts @@ -6,6 +6,7 @@ import { addSourcePagePayloadBytes, ConnectorSyncCapacityError, } from '@/lib/knowledge/connectors/sync-primitives' +import { listingFailuresSchema } from '@/connectors/listing-failures' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' const checkpointSchema = z.object({ @@ -22,6 +23,7 @@ const checkpointSchema = z.object({ unsafe: z.boolean(), contentFailures: z.boolean().default(false), permissionFailures: z.boolean().default(false), + listingFailures: listingFailuresSchema.nullable().default(null), changeCursor: z .string() .max(512 * 1024) @@ -68,6 +70,7 @@ export function beginListingCheckpoint(input: { unsafe: false, contentFailures: false, permissionFailures: false, + listingFailures: null, changeCursor: input.changeCursor ?? null, incrementalSince: input.incrementalSince?.toISOString() ?? null, forceRehydrate: input.forceRehydrate ?? false, @@ -135,6 +138,7 @@ export async function runResumableListing(input: { unsafe: false, contentFailures: false, permissionFailures: false, + listingFailures: null, } await input.saveCheckpoint(checkpoint) cursors.clear() @@ -152,11 +156,15 @@ export async function runResumableListing(input: { } checkpoint.unsafe ||= response.reconciliationSafe === false || + response.listingFailures !== undefined || Boolean( input.syncContext.listingCapped || input.syncContext.listingTruncated || input.syncContext.reconciliationUnsafe ) + if (response.listingFailures) { + checkpoint.listingFailures = listingFailuresSchema.parse(response.listingFailures) + } if (response.currentCursor !== undefined && response.currentCursor !== checkpoint.cursor) { if (response.currentCursor.length > 512 * 1024) throw new ConnectorSyncCapacityError('Connector returned an oversized listing cursor') diff --git a/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts b/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts index 798459d4d88..59b87aca767 100644 --- a/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-content-pass.test.ts @@ -385,6 +385,35 @@ function contentWrite(): Record { } describe('content pass checkpoint intent', () => { + it('does not reconcile deletions after a user listing failed, even without the unsafe marker', async () => { + sourceBody = { value: '

Current content

' } + const checkpoint = { + ...beginListingCheckpoint({ + fingerprint: 'a'.repeat(64), + generationId: 'prior', + startedAt: new Date(0), + }), + listingFailures: { + count: 1, + samples: [ + { + scope: 'unavailable@example.com', + operation: 'gmail.threads.list', + status: 400, + reasons: ['failedPrecondition'], + }, + ], + }, + } + const { pass, result } = await runPass({ checkpoint, access: 'admin' }) + expect(pass.complete).toBe(true) + expect(pass.checkpoint.listingFailures).toEqual(checkpoint.listingFailures) + expect(pass.holdNotice).toContain('unlisted documents were kept') + expect(result.docsDeleted).toBe(0) + expect(mocks.hardDelete).not.toHaveBeenCalled() + expect(dbChainMockFns.set.mock.calls.some(([value]) => value.deletedAt != null)).toBe(false) + }) + it('persists unresolved permissions independently of successful content processing', async () => { sourceBody = { value: '

Current content

' } mocks.onPage.mockResolvedValue({ permissionsIncomplete: true }) diff --git a/apps/sim/lib/knowledge/connectors/sync-content-pass.ts b/apps/sim/lib/knowledge/connectors/sync-content-pass.ts index 7ef42e94c5e..6fc86b716fb 100644 --- a/apps/sim/lib/knowledge/connectors/sync-content-pass.ts +++ b/apps/sim/lib/knowledge/connectors/sync-content-pass.ts @@ -224,7 +224,7 @@ async function reconcileCompletedListing( checkpoint: ListingCheckpoint, withLease: (fn: (tx: DbOrTx) => Promise) => Promise ): Promise<{ finished: boolean; notice: string | null }> { - if (checkpoint.unsafe) + if (checkpoint.unsafe || (checkpoint.listingFailures?.count ?? 0) > 0) return { finished: true, notice: diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index ea8dc8c500a..29ad7d9d714 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts @@ -2142,6 +2142,78 @@ describe('completeSuccessfulSync', () => { resetDbChainMock() }) + it.each([false, true])( + 'preserves directory and listing notices without blocking healthy content watermarks: listing failure %s', + async (hasListingFailure) => { + const { completeSuccessfulSync } = await import('@/lib/knowledge/connectors/sync-engine') + queueTableRows(schemaMock.knowledgeBase, [{ id: 'kb-1' }]) + queueTableRows(schemaMock.knowledgeConnector, [{ id: 'c-1' }]) + queueTableRows(schemaMock.document, [{ count: 4 }]) + dbChainMockFns.returning + .mockResolvedValueOnce([]) + .mockResolvedValueOnce([{ id: 'log-1' }]) + .mockResolvedValueOnce([{ id: 'c-1' }]) + const directoryNotice = + 'Directory refresh incomplete: 1 group membership could not be verified.' + const contentNotice = 'Unlisted documents were kept.' + + await expect( + completeSuccessfulSync( + 'c-1', + 'kb-1', + 'log-1', + 60, + { ...RESULT, docsFailed: 0 }, + contentNotice, + { + complete: true, + checkpoint: { + unsafe: hasListingFailure, + startedAt: '2026-09-04T00:00:00Z', + listedCount: 4, + listingFailures: hasListingFailure + ? { + count: 1, + samples: [ + { + scope: 'unavailable@example.com', + operation: 'gmail.threads.list', + status: 400, + reasons: ['failedPrecondition'], + }, + ], + } + : null, + }, + }, + directoryNotice + ) + ).resolves.toBe(true) + + const logUpdate = dbChainMockFns.set.mock.calls.find( + ([value]) => value.status === 'partial' + )?.[0] + const connectorUpdate = dbChainMockFns.set.mock.calls.find( + ([value]) => value.status === 'active' + )?.[0] + expect(logUpdate.errorMessage).toContain(directoryNotice) + expect(logUpdate.errorMessage).toContain(contentNotice) + expect(connectorUpdate.lastSyncError).toBe(logUpdate.errorMessage) + expect(connectorUpdate.listingCheckpoint).toBeNull() + expect(connectorUpdate.consecutiveFailures).toBe(0) + expect(connectorUpdate.nextSyncAt.getTime()).toBeGreaterThan(Date.now() + 50 * 60_000) + if (hasListingFailure) { + expect(connectorUpdate).not.toHaveProperty('lastSyncAt') + expect(logUpdate.errorMessage).toContain( + 'unavailable@example.com (gmail.threads.list, HTTP 400, failedPrecondition)' + ) + expect(logUpdate.errorMessage).toContain('next scheduled sync') + } else { + expect(connectorUpdate.lastSyncAt).toEqual(new Date('2026-09-04T00:00:00Z')) + } + } + ) + it('commits the completed log and connector state in one guarded transaction', async () => { const { completeSuccessfulSync } = await import('@/lib/knowledge/connectors/sync-engine') diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.ts b/apps/sim/lib/knowledge/connectors/sync-engine.ts index dccaf32d775..5632d2e6a43 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.ts @@ -33,7 +33,9 @@ import { } from '@/lib/knowledge/connectors/access-token' import { getConnectorFailureDiagnostic } from '@/lib/knowledge/connectors/connector-error' import { - DIRECTORY_ERROR_PREFIX, + type DirectoryRefreshResult, + directorySyncNotice, + hasDirectorySyncNotice, refreshMirroredDirectory, } from '@/lib/knowledge/connectors/external-group-sync' import { listingFingerprint } from '@/lib/knowledge/connectors/listing-checkpoint' @@ -86,6 +88,7 @@ import type { ConnectorAuthConfig, ConnectorConfig, ExternalDocument, + ExternalListingFailures, SyncResult, } from '@/connectors/types' @@ -298,6 +301,7 @@ export interface ContentPassOutcome { unsafe: boolean contentFailures?: boolean permissionFailures?: boolean + listingFailures?: ExternalListingFailures | null startedAt: string listedCount: number incrementalSince?: string | null @@ -315,7 +319,8 @@ export function isContentPassIncomplete( return ( !contentPass.complete || contentPass.checkpoint.contentFailures === true || - contentPass.checkpoint.permissionFailures === true + contentPass.checkpoint.permissionFailures === true || + (contentPass.checkpoint.listingFailures?.count ?? 0) > 0 ) } @@ -333,14 +338,33 @@ export async function completeSuccessfulSync( syncIntervalMinutes: number, result: SyncResult, reconciliationHoldNotice: string | null, - contentPass?: ContentPassOutcome + contentPass?: ContentPassOutcome, + directoryNotice: string | null = null ): Promise { const processingDispatchFailed = result.processingDispatch.failed > 0 - const completionNotice = + const contentNotice = reconciliationHoldNotice ?? (processingDispatchFailed ? 'Some documents could not be queued for indexing. They will be retried automatically.' : null) + const listingFailures = contentPass?.checkpoint.listingFailures + const failedAccounts = listingFailures?.samples + .map((failure) => { + const details = [ + failure.operation, + failure.status ? `HTTP ${failure.status}` : null, + ...failure.reasons, + ] + .filter(Boolean) + .join(', ') + return `${failure.scope} (${details})` + }) + .join('; ') + const listingNotice = listingFailures?.count + ? `Source listing failed for ${listingFailures.count} ${listingFailures.count === 1 ? 'account' : 'accounts'}.${failedAccounts ? ` ${failedAccounts}.` : ''} Failed accounts will be retried at the next scheduled sync.` + : null + const completionNotice = + [directoryNotice, listingNotice, contentNotice].filter(Boolean).join('\n') || null try { return await db.transaction(async (tx) => { const [lockedKnowledgeBase] = await tx @@ -389,7 +413,9 @@ export async function completeSuccessfulSync( .update(knowledgeConnectorSyncLog) .set({ status: - processingDispatchFailed || (contentPass && isContentPassIncomplete(contentPass)) + directoryNotice || + processingDispatchFailed || + (contentPass && isContentPassIncomplete(contentPass)) ? 'partial' : 'completed', completedAt: now, @@ -1043,7 +1069,9 @@ export async function executeSync( (options?.rehydrate || options?.fullSync) && connectorConfig.rehydrateOnFullSync ) - let directoryRefreshed: Promise = Promise.resolve(undefined) + let directoryRefreshed: Promise = Promise.resolve({ + status: 'skipped', + }) if (mirrored) { /** * A switch into this mode hides every document before it flips, and one @@ -1077,8 +1105,8 @@ export async function executeSync( force: Boolean(options.fullSync) || connector.consecutiveFailures > 0 || - connector.lastSyncError?.startsWith(DIRECTORY_ERROR_PREFIX), - }).then(() => undefined, toError) + hasDirectorySyncNotice(connector.lastSyncError), + }).catch(toError) } const contentPass = await runConnectorContentPass({ @@ -1140,8 +1168,14 @@ export async function executeSync( result.listingIncomplete = isContentPassIncomplete(contentPass) const reconciliationHoldNotice = contentPass.holdNotice - const directoryError = await directoryRefreshed - if (directoryError) throw directoryError + const directoryOutcome = await directoryRefreshed + if (directoryOutcome instanceof Error) throw directoryOutcome + const directoryNotice = + directoryOutcome.status === 'partial' + ? directoryOutcome.notice + : directoryOutcome.status === 'skipped' && mirrored + ? directorySyncNotice(connector.lastSyncError) + : null const postBatchPresence = await checkSyncTargetPresence( connectorId, @@ -1171,7 +1205,8 @@ export async function executeSync( effectiveConnectorSyncIntervalMinutes(connector.accessMode, connector.syncIntervalMinutes), result, reconciliationHoldNotice, - contentPass + contentPass, + directoryNotice ) if (!completionLanded) { diff --git a/apps/sim/lib/knowledge/documents/document-processing-source.test.ts b/apps/sim/lib/knowledge/documents/document-processing-source.test.ts index 73ff94ee27d..dd0941ad827 100644 --- a/apps/sim/lib/knowledge/documents/document-processing-source.test.ts +++ b/apps/sim/lib/knowledge/documents/document-processing-source.test.ts @@ -805,6 +805,110 @@ describe('processDocumentAsync write guards', () => { ).toBe(false) }) + it('writes every chunk in bounded batches before completing the document', async () => { + armProviderSource() + const chunks = Array.from({ length: 205 }, (_, index) => ({ + text: `Chunk ${index}`, + metadata: { startIndex: index * 10, endIndex: index * 10 + 9 }, + })) + mockProcessDocument.mockResolvedValueOnce({ + chunks, + metadata: { chunkCount: chunks.length, tokenCount: 615, characterCount: 2050 }, + }) + mockGenerateEmbeddings.mockResolvedValueOnce({ + embeddings: chunks.map((_, index) => [index / chunks.length]), + billableTokens: 0, + modelName: 'text-embedding-3-small', + pricingId: 'text-embedding-3-small', + }) + + await processDocumentAsync( + 'knowledge-base-1', + 'document-1', + { + filename: 'a.txt', + fileUrl: 'https://example.com/a.txt', + fileSize: 2050, + mimeType: 'text/plain', + }, + {}, + BILLING_ATTRIBUTION + ) + + const batches = dbChainMockFns.values.mock.calls + .map(([value]) => value) + .filter((value) => Array.isArray(value) && value[0]?.documentId === 'document-1') + expect(batches.map((batch) => batch.length)).toEqual([100, 100, 5]) + expect(batches.flat().map((record) => record.chunkIndex)).toEqual( + chunks.map((_, index) => index) + ) + expect(batches.flat().map((record) => record.content)).toEqual( + chunks.map((chunk) => chunk.text) + ) + expect(batches.flat().map((record) => record.embedding)).toEqual( + chunks.map((_, index) => [index / chunks.length]) + ) + expect(dbChainMockFns.transaction).toHaveBeenCalledOnce() + expect(guardForStatusWrite('completed')).toBeDefined() + const completionIndex = dbChainMockFns.set.mock.calls.findIndex( + ([value]) => value.processingStatus === 'completed' + ) + expect(dbChainMockFns.set.mock.invocationCallOrder[completionIndex]).toBeGreaterThan( + Math.max(...dbChainMockFns.values.mock.invocationCallOrder) + ) + }) + + it('aborts the document transaction when a later embedding batch fails', async () => { + armProviderSource() + const chunks = Array.from({ length: 205 }, (_, index) => ({ + text: `Chunk ${index}`, + metadata: { startIndex: index * 10, endIndex: index * 10 + 9 }, + })) + mockProcessDocument.mockResolvedValueOnce({ + chunks, + metadata: { chunkCount: chunks.length, tokenCount: 615, characterCount: 2050 }, + }) + mockGenerateEmbeddings.mockResolvedValueOnce({ + embeddings: chunks.map(() => [0.1]), + billableTokens: 0, + modelName: 'text-embedding-3-small', + pricingId: 'text-embedding-3-small', + }) + const databaseError = new DrizzleQueryError( + 'insert private SQL', + ['private content'], + Object.assign(new Error('canceling statement due to statement timeout'), { code: '57014' }) + ) + dbChainMockFns.values.mockResolvedValueOnce([]).mockRejectedValueOnce(databaseError) + + await expect( + processDocumentAsync( + 'knowledge-base-1', + 'document-1', + { + filename: 'a.txt', + fileUrl: 'https://example.com/a.txt', + fileSize: 2050, + mimeType: 'text/plain', + }, + {}, + BILLING_ATTRIBUTION + ) + ).rejects.toBe(databaseError) + + expect(dbChainMockFns.transaction).toHaveBeenCalledOnce() + await expect(dbChainMockFns.transaction.mock.results[0].value).rejects.toBe(databaseError) + expect(dbChainMockFns.values).toHaveBeenCalledTimes(2) + expect(mockLogError).toHaveBeenCalledWith( + '[document-1] Failed to insert embedding batch', + expect.objectContaining({ batchNumber: 2, batchSize: 100, totalChunks: 205 }) + ) + expect( + dbChainMockFns.set.mock.calls.some(([value]) => value.processingStatus === 'completed') + ).toBe(false) + expect(guardForStatusWrite('failed')).toBeDefined() + }) + it('accepts a legacy queuedAt-only payload only while the row has no token', async () => { dbChainMockFns.limit .mockResolvedValueOnce([PERSISTED_CONTEXT]) diff --git a/apps/sim/lib/knowledge/documents/service.ts b/apps/sim/lib/knowledge/documents/service.ts index 50699895060..1aac135fb6e 100644 --- a/apps/sim/lib/knowledge/documents/service.ts +++ b/apps/sim/lib/knowledge/documents/service.ts @@ -329,7 +329,8 @@ const TIMEOUTS = { } as const const LARGE_DOC_CONFIG = { - MAX_CHUNKS_PER_BATCH: 500, + /** Each chunk also updates search projections and indexes inside the same statement. */ + MAX_CHUNKS_PER_BATCH: 100, MAX_EMBEDDING_BATCH: Math.min( envNumber(env.KB_CONFIG_BATCH_SIZE, 2000, { min: 1, integer: true }), getEmbeddingAggregateItemLimit(MAX_KB_EMBEDDING_DIMENSIONS)