diff --git a/apps/sim/connectors/gmail/gmail.ts b/apps/sim/connectors/gmail/gmail.ts index b25f6243a8c..910fd35b661 100644 --- a/apps/sim/connectors/gmail/gmail.ts +++ b/apps/sim/connectors/gmail/gmail.ts @@ -4,6 +4,8 @@ import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/document import { DEFAULT_MAX_THREADS, gmailConnectorMeta } from '@/connectors/gmail/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { + BoundedLines, + CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, htmlToPlainText, joinTagArray, parseDefaultedUnlimitedSafeInteger, @@ -323,21 +325,17 @@ function formatThread(thread: GmailThread): { } const labelIds = [...labelIdSet] - const lines: string[] = [] - lines.push(`Subject: ${subject}`) - lines.push(`From: ${from}`) + const lines = new BoundedLines() + lines.push(`Subject: ${subject}`, `From: ${from}`) if (to) lines.push(`To: ${to}`) - lines.push(`Messages: ${messages.length}`) - lines.push('') + lines.push(`Messages: ${messages.length}`, '') for (const msg of messages) { const msgFrom = getHeader(msg.payload, 'From') || 'Unknown' const msgDate = getHeader(msg.payload, 'Date') || '' const body = msg.payload ? extractBody(msg.payload) : '' - lines.push(`--- ${msgFrom} (${msgDate}) ---`) - lines.push(body.trim()) - lines.push('') + if (!lines.push(`--- ${msgFrom} (${msgDate}) ---`, body.trim(), '')) break } const firstDate = firstMessage.internalDate @@ -348,7 +346,7 @@ function formatThread(thread: GmailThread): { : undefined return { - content: lines.join('\n').trim(), + content: lines.join().trim(), subject, metadata: { from, @@ -412,6 +410,7 @@ function threadToStub(thread: { title: thread.snippet || 'Untitled Thread', content: '', contentDeferred: true, + estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, mimeType: 'text/plain', sourceUrl: threadUrl(thread.id), contentHash: `gmail:${thread.id}:${thread.historyId ?? ''}`, diff --git a/apps/sim/connectors/google-chat/google-chat.ts b/apps/sim/connectors/google-chat/google-chat.ts index 1dcca49d7b3..1d694ab6c35 100644 --- a/apps/sim/connectors/google-chat/google-chat.ts +++ b/apps/sim/connectors/google-chat/google-chat.ts @@ -9,7 +9,7 @@ import { SPACES_PAGE_SIZE, } from '@/connectors/google-chat/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' -import { parseTagDate } from '@/connectors/utils' +import { BoundedLines, CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, parseTagDate } from '@/connectors/utils' const logger = createLogger('GoogleChatConnector') @@ -218,6 +218,7 @@ function spaceToStub( title: spaceTitle(space), content: '', contentDeferred: true, + estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, mimeType: 'text/plain', sourceUrl: space.spaceUri, contentHash: buildContentHash(space, maxMessages, lookbackDays, syncContext), @@ -323,37 +324,28 @@ function senderLabel(sender: ChatUser | undefined): string { * that belongs in a knowledge base is synced through the Google Drive connector * instead, which already handles size caps, OCR, and format parsing. */ -function formatSpaceContent(space: Space, messages: ChatMessage[]): string { - const parts: string[] = [`Space: ${spaceTitle(space)}`] +function formatSpaceContent( + space: Space, + messages: ChatMessage[] +): { content: string; messageCount: number } { + const parts = new BoundedLines() + parts.push(`Space: ${spaceTitle(space)}`) const description = space.spaceDetails?.description?.trim() if (description) parts.push(`Description: ${description}`) const guidelines = space.spaceDetails?.guidelines?.trim() if (guidelines) parts.push(`Guidelines: ${guidelines}`) - const lines: string[] = [] + let messageCount = 0 for (const message of messages) { const text = message.text?.trim() || message.fallbackText?.trim() if (!text) continue + if (messageCount === 0) parts.push('', '--- Messages ---') const timestamp = message.createTime ?? '' - lines.push(`[${timestamp}] ${senderLabel(message.sender)}: ${text}`) - } - - if (lines.length > 0) { - parts.push('') - parts.push('--- Messages ---') - parts.push(...lines) + if (!parts.push(`[${timestamp}] ${senderLabel(message.sender)}: ${text}`)) break + messageCount += 1 } - return parts.join('\n') -} - -/** Number of messages that actually contributed text to the transcript. */ -function countIndexedMessages(messages: ChatMessage[]): number { - let count = 0 - for (const message of messages) { - if (message.text?.trim() || message.fallbackText?.trim()) count++ - } - return count + return { content: parts.join(), messageCount } } export const googleChatConnector: ConnectorConfig = { @@ -476,12 +468,12 @@ export const googleChatConnector: ConnectorConfig = { * leave a previously indexed transcript in place after the space was cleared * or `lookbackDays` was tightened past every message. */ - const messageCount = countIndexedMessages(messages) + const { content, messageCount } = formatSpaceContent(space, messages) const stub = spaceToStub(space, maxMessages, lookbackDays, syncContext) return { ...stub, - content: formatSpaceContent(space, messages), + content, contentDeferred: false, metadata: { ...stub.metadata, messageCount }, } diff --git a/apps/sim/connectors/outlook/outlook.ts b/apps/sim/connectors/outlook/outlook.ts index 78f183883d6..c3a912363ba 100644 --- a/apps/sim/connectors/outlook/outlook.ts +++ b/apps/sim/connectors/outlook/outlook.ts @@ -4,6 +4,8 @@ import { fetchWithRetry, VALIDATE_RETRY_OPTIONS } from '@/lib/knowledge/document import { DEFAULT_MAX_CONVERSATIONS, outlookConnectorMeta } from '@/connectors/outlook/meta' import type { ConnectorConfig, ExternalDocument, ExternalDocumentList } from '@/connectors/types' import { + BoundedLines, + CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, htmlToPlainText, isListingScopeUnavailableError, listingRequestError, @@ -563,24 +565,20 @@ function formatConversation( const from = formatRecipient(first.from) const to = first.toRecipients?.map(formatRecipient).join(', ') || '' - const lines: string[] = [] - lines.push(`Subject: ${subject}`) - lines.push(`From: ${from}`) + const lines = new BoundedLines() + lines.push(`Subject: ${subject}`, `From: ${from}`) if (to) lines.push(`To: ${to}`) - lines.push(`Messages: ${sorted.length}`) - lines.push('') + lines.push(`Messages: ${sorted.length}`, '') for (const msg of sorted) { const msgFrom = formatRecipient(msg.from) const msgDate = msg.receivedDateTime || '' const body = extractBodyText(msg.body) - lines.push(`--- ${msgFrom} (${msgDate}) ---`) - lines.push(body.trim()) - lines.push('') + if (!lines.push(`--- ${msgFrom} (${msgDate}) ---`, body.trim(), '')) break } - const content = lines.join('\n').trim() + const content = lines.join().trim() if (!content) return null const categories = new Set() @@ -805,6 +803,7 @@ export const outlookConnector: ConnectorConfig = { title: subject, content: '', contentDeferred: true, + estimatedBytes: CONNECTOR_TEXT_DOCUMENT_MAX_BYTES, mimeType: 'text/plain', sourceUrl, contentHash: `outlook:${convId}:${lastDate}`, diff --git a/apps/sim/connectors/types.ts b/apps/sim/connectors/types.ts index f462cb29dc3..a185a701db6 100644 --- a/apps/sim/connectors/types.ts +++ b/apps/sim/connectors/types.ts @@ -63,6 +63,14 @@ export interface ExternalDocument { skippedRetryContentHash?: string /** When true, content is empty and will be fetched via getDocument for new/changed docs only */ contentDeferred?: boolean + /** + * How large the deferred content is expected to be, in bytes, when the + * listing cannot know exactly. Bounds how many deferred documents hydrate + * at once: without it a deferred document is assumed to be as large as the + * whole in-flight budget and hydrates alone, which turns a mailbox crawl + * into one thread at a time. + */ + estimatedBytes?: number /** * When set, the document was intentionally not indexed (e.g. it exceeds the * connector's size limit). The sync engine records it as a `failed` document diff --git a/apps/sim/connectors/utils.test.ts b/apps/sim/connectors/utils.test.ts index 353d715eaa0..b433006d63c 100644 --- a/apps/sim/connectors/utils.test.ts +++ b/apps/sim/connectors/utils.test.ts @@ -63,6 +63,7 @@ import { typeformConnector } from '@/connectors/typeform/typeform' import { appendPendingMicrosoftGraphFolders, assertMicrosoftGraphNextLink, + BoundedLines, ConnectorFileTooLargeError, ConnectorListingScopeUnavailableError, decodeMicrosoftGraphTraversalCursor, @@ -1639,3 +1640,26 @@ describe('isSkippableMicrosoftGraphFolderError', () => { expect(isSkippableMicrosoftGraphFolderError(new Error('500'), perMember, false)).toBe(false) }) }) + +describe('BoundedLines', () => { + it('joins everything when the text fits', () => { + const lines = new BoundedLines(64) + expect(lines.push('Subject: hi', '')).toBe(true) + expect(lines.push('--- a ---', 'body')).toBe(true) + expect(lines.join()).toBe('Subject: hi\n\n--- a ---\nbody') + }) + + it('refuses a record that would cross the ceiling, whole, and says so in the output', () => { + const lines = new BoundedLines(20) + expect(lines.push('first')).toBe(true) + expect(lines.push('--- header ---', 'a long body')).toBe(false) + expect(lines.push('x')).toBe(false) + expect(lines.join()).toBe('first\n[Truncated: the indexed text reached the size limit]') + }) + + it('counts encoded bytes, not characters', () => { + const lines = new BoundedLines(6) + expect(lines.push('éé')).toBe(true) + expect(lines.push('é')).toBe(false) + }) +}) diff --git a/apps/sim/connectors/utils.ts b/apps/sim/connectors/utils.ts index 857bac814a7..e3bc5485113 100644 --- a/apps/sim/connectors/utils.ts +++ b/apps/sim/connectors/utils.ts @@ -795,3 +795,50 @@ export function isSkippableMicrosoftGraphFolderError( ): boolean { return !isRootFolder && isListingScopeUnavailableError(error) && isPerMemberListing(syncContext) } + +/** + * Ceiling for a document a connector assembles from many source records (a + * mail thread, a chat transcript) rather than downloads as one file. Formatters + * enforce it through `BoundedLines`, and listings advertise it through + * `ExternalDocument.estimatedBytes`, so the sync engine plans hydration around + * a bound it can rely on: five such documents fit its 64 MiB in-flight budget + * and hydrate together instead of one at a time. + */ +export const CONNECTOR_TEXT_DOCUMENT_MAX_BYTES = 12 * 1024 * 1024 + +const TRUNCATION_NOTICE = '[Truncated: the indexed text reached the size limit]' + +/** + * Accumulates newline-joined text under a byte ceiling. A record is appended + * whole or not at all, so a truncated document never ends mid-message, and the + * output carries a notice when something was left out. + */ +export class BoundedLines { + private readonly lines: string[] = [] + private bytes = 0 + private truncated = false + + constructor(private readonly maxBytes = CONNECTOR_TEXT_DOCUMENT_MAX_BYTES) {} + + /** + * Appends the lines together when they fit; otherwise marks the document + * truncated, appends nothing, and returns false so the caller stops. + */ + push(...lines: string[]): boolean { + if (this.truncated) return false + let size = 0 + for (const line of lines) size += Buffer.byteLength(line, 'utf8') + 1 + if (this.bytes + size > this.maxBytes) { + this.truncated = true + return false + } + this.lines.push(...lines) + this.bytes += size + return true + } + + /** Joins the accepted lines, ending with the truncation notice when a push was refused. */ + join(): string { + return this.truncated ? [...this.lines, TRUNCATION_NOTICE].join('\n') : this.lines.join('\n') + } +} diff --git a/apps/sim/lib/knowledge/connectors/member-observations.ts b/apps/sim/lib/knowledge/connectors/member-observations.ts index 361db08c282..086c5166b0d 100644 --- a/apps/sim/lib/knowledge/connectors/member-observations.ts +++ b/apps/sim/lib/knowledge/connectors/member-observations.ts @@ -446,7 +446,13 @@ export async function sweepStaleMemberObservations(now: Date): Promise interval". + */ + const cutoff = sql`${sql.param(now, knowledgeConnectorMember.lastStartedAt)}::timestamp - ${staleWindow}` const staleMembers = await db .select({ id: knowledgeConnectorMember.id, diff --git a/apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts index ba573316dab..30e635f1324 100644 --- a/apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/member-sync-engine.test.ts @@ -26,6 +26,7 @@ import { memberFailureBackoffMs, memberNextAttemptAt, nextMemberSyncTime, + persistedDocumentsByObserver, shouldListFully, } from '@/lib/knowledge/connectors/member-sync-engine' import { @@ -272,6 +273,26 @@ describe('member sync engine decisions', () => { expect(second.retainedBytes).toBeGreaterThan(first.retainedBytes) }) + it('grants a persisted batch to every member who listed each document, as it lands', () => { + const union = new Map() + admitMemberListing(union, 'm-1', [doc('a'), doc('b')], 'c-1', 0) + admitMemberListing(union, 'm-2', [doc('a'), doc('c')], 'c-1', 0) + + const byMember = persistedDocumentsByObserver( + [ + { externalId: 'a', documentId: 'd-a' }, + { externalId: 'b', documentId: 'd-b' }, + { externalId: 'zzz', documentId: 'd-z' }, + ], + union + ) + + expect([...byMember.entries()]).toEqual([ + ['m-1', ['d-a', 'd-b']], + ['m-2', ['d-a']], + ]) + }) + it('counts a member once per external id even when their listing repeats it', () => { const union = new Map() const admitted = admitMemberListing(union, 'm-1', [doc('a'), doc('a')], 'c-1', 0) diff --git a/apps/sim/lib/knowledge/connectors/member-sync-engine.ts b/apps/sim/lib/knowledge/connectors/member-sync-engine.ts index afda33d905f..f451ddae0f6 100644 --- a/apps/sim/lib/knowledge/connectors/member-sync-engine.ts +++ b/apps/sim/lib/knowledge/connectors/member-sync-engine.ts @@ -59,7 +59,10 @@ import { SyncLockLostException, stillHoldsMemberSyncLock, } from '@/lib/knowledge/connectors/sync-lock' -import type { KnowledgeBaseOwner } from '@/lib/knowledge/connectors/sync-persistence' +import type { + KnowledgeBaseOwner, + PersistedDocument, +} from '@/lib/knowledge/connectors/sync-persistence' import { addSourcePagePayloadBytes, ConnectorDeletedException, @@ -169,6 +172,27 @@ interface MemberListingOutcome { changeCursor: string | null | undefined } +/** + * The documents a batch wrote, grouped by each member who listed them, so a + * grant can be recorded per observer the moment the row exists. A document + * nobody in the union listed (it cannot happen for a persisted one, but the + * map is the source of truth) grants nothing. + */ +export function persistedDocumentsByObserver( + persisted: readonly PersistedDocument[], + union: ReadonlyMap> +): Map { + const byMember = new Map() + for (const { externalId, documentId } of persisted) { + for (const memberId of union.get(externalId)?.observers ?? []) { + const documentIds = byMember.get(memberId) + if (documentIds) documentIds.push(documentId) + else byMember.set(memberId, [documentId]) + } + } + return byMember +} + interface UnionEntry { document: ExternalDocument /** Member ids whose listings returned the document, in listing order. */ @@ -1550,6 +1574,33 @@ export async function executeMemberSync( }, lease: run.lease, documentAccess: 'members', + /** + * Grants surface as each batch lands: every member who listed a document + * observes it the moment its row exists, and its ACL is materialised in + * the same lease-proved transaction. The listing's own pass below records + * the same observations again, idempotently, and is still what decides + * removals; this only brings the additions forward from the end of the + * run to the moment they are indexed. + */ + onBatchPersisted: async (persisted) => { + const byMember = persistedDocumentsByObserver(persisted, union) + if (byMember.size === 0) return + await withMemberLease(run, async (tx) => { + for (const [memberId, documentIds] of byMember) { + result.observationsAdded += await recordMemberObservations( + tx, + memberId, + documentIds, + runId + ) + } + await materializeDocumentAcls( + connectorId, + persisted.map(({ documentId }) => documentId), + tx + ) + }) + }, }) const documentIdByExternalId = await loadDocumentIdsByExternalId(connectorId) diff --git a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts index 61ac44d2c7f..d57a57ba7fa 100644 --- a/apps/sim/lib/knowledge/connectors/sync-engine.test.ts +++ b/apps/sim/lib/knowledge/connectors/sync-engine.test.ts @@ -789,7 +789,7 @@ describe('persistSkippedDocuments', () => { 'workspace', lease ) - ).resolves.toBe(1) + ).resolves.toHaveLength(1) expect(dbChainMockFns.values).toHaveBeenCalledWith([ expect.objectContaining({ @@ -835,7 +835,7 @@ describe('persistSkippedDocuments', () => { 'workspace', lease ) - ).resolves.toBe(1) + ).resolves.toHaveLength(1) expect(dbChainMockFns.set).toHaveBeenCalledWith( expect.objectContaining({ @@ -1028,6 +1028,32 @@ describe('chunkOpsByByteBudget', () => { expect(chunks.map((c) => c.length)).toEqual([1, 1]) }) + it('hydrates deferred documents together when the listing estimates their size', async () => { + const { chunkOpsByByteBudget } = await import('@/lib/knowledge/connectors/sync-primitives') + const deferred = (estimatedBytes?: number) => ({ + type: 'add' as const, + extDoc: { + externalId: `d-${generateShortId()}`, + title: 'f', + content: '', + contentDeferred: true, + contentHash: 'h', + mimeType: 'text/plain', + ...(estimatedBytes != null ? { estimatedBytes } : {}), + }, + }) + // Without an estimate each unknown download is assumed to fill the budget and runs alone. + expect(chunkOpsByByteBudget([deferred(), deferred(), deferred()], 64 * MB, 5)).toHaveLength(3) + // A mail thread that says it is small shares a batch with its neighbours. + expect( + chunkOpsByByteBudget( + [deferred(256 * 1024), deferred(256 * 1024), deferred(256 * 1024)], + 64 * MB, + 5 + ) + ).toHaveLength(1) + }) + it('treats skip ops as zero bytes so they do not consume the budget', async () => { const { chunkOpsByByteBudget } = await import('@/lib/knowledge/connectors/sync-primitives') const chunks = chunkOpsByByteBudget( diff --git a/apps/sim/lib/knowledge/connectors/sync-persistence.ts b/apps/sim/lib/knowledge/connectors/sync-persistence.ts index 7fb7f537c22..95106d2786e 100644 --- a/apps/sim/lib/knowledge/connectors/sync-persistence.ts +++ b/apps/sim/lib/knowledge/connectors/sync-persistence.ts @@ -247,6 +247,12 @@ function buildSkippedDocumentRow( * * Returns the number of rows recorded. */ +/** A document a sync wrote, by the id the source knows it by and the id the row has. */ +export interface PersistedDocument { + externalId: string + documentId: string +} + export async function persistSkippedDocuments( knowledgeBaseId: string, connectorId: string, @@ -259,9 +265,9 @@ export async function persistSkippedDocuments( sourceConfig: Record | undefined, access: SyncDocumentAccess, lease: SyncWriteLease -): Promise { +): Promise { if (skipOps.length === 0) { - return 0 + return [] } const inserts = skipOps .filter((op) => !op.existingId) @@ -275,6 +281,12 @@ export async function persistSkippedDocuments( access ) ) + const persisted: PersistedDocument[] = [ + ...inserts.map((row) => ({ externalId: row.externalId, documentId: row.id })), + ...skipOps + .filter((op): op is typeof op & { existingId: string } => Boolean(op.existingId)) + .map((op) => ({ externalId: op.extDoc.externalId, documentId: op.existingId })), + ] const replacements = skipOps.filter((op): op is typeof op & { existingId: string } => Boolean(op.existingId) ) @@ -363,7 +375,7 @@ export async function persistSkippedDocuments( } } - return skipOps.length + return persisted } /** diff --git a/apps/sim/lib/knowledge/connectors/sync-primitives.ts b/apps/sim/lib/knowledge/connectors/sync-primitives.ts index f1b3938dbd1..6e7c6b3a052 100644 --- a/apps/sim/lib/knowledge/connectors/sync-primitives.ts +++ b/apps/sim/lib/knowledge/connectors/sync-primitives.ts @@ -16,6 +16,7 @@ import { SyncLockLostException, type SyncRunLease } from '@/lib/knowledge/connec import { addDocument, type KnowledgeBaseOwner, + type PersistedDocument, persistSkippedDocuments, persistSkippedRetryHashes, type SyncDocumentAccess, @@ -503,7 +504,7 @@ function estimateOpSizeBytes(op: DocOp): number { if (op.type === 'skip') return 0 if (op.extDoc.sourceFile?.bytes) return op.extDoc.sourceFile.bytes.byteLength if (op.extDoc.content) return Buffer.byteLength(op.extDoc.content) - const size = op.extDoc.metadata?.fileSize ?? op.extDoc.metadata?.size + const size = op.extDoc.estimatedBytes ?? op.extDoc.metadata?.fileSize ?? op.extDoc.metadata?.size return typeof size === 'number' && Number.isFinite(size) && size > 0 ? size : DEFAULT_OP_SIZE_BYTES @@ -1496,6 +1497,16 @@ export interface ProcessDocOpsInput { state: SyncRunState hydration: DocOpHydration lease: Pick + /** + * Runs after each batch is written and dispatched, with every row that + * landed: hydrated documents and skipped ones alike. A members-mode run + * grants access here, batch by batch, so a long first crawl becomes + * searchable as it goes rather than all at once at the end. Best effort: a + * failure here is logged and the run continues, because the listing's own + * pass at the end of the run writes the same grants authoritatively; only a + * lost lease ends the run. + */ + onBatchPersisted?: (persisted: readonly PersistedDocument[]) => Promise /** Who may read the documents this pass writes. */ documentAccess: SyncDocumentAccess } @@ -1677,6 +1688,7 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise { } } + const skippedPersisted: PersistedDocument[] = [] if (skipOps.length > 0) { try { const recorded = await persistSkippedDocuments( @@ -1688,7 +1700,8 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise { documentAccess, input.lease ) - result.docsSkipped += recorded + result.docsSkipped += recorded.length + skippedPersisted.push(...recorded) } catch (error) { if (error instanceof SyncLockLostException) throw error /** @@ -1744,10 +1757,15 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise { if (leaseLost) throw leaseLost.reason const batchDocs: DocumentData[] = [] + const persisted: PersistedDocument[] = [...skippedPersisted] for (let j = 0; j < settled.length; j++) { const outcome = settled[j] if (outcome.status === 'fulfilled') { batchDocs.push(outcome.value) + persisted.push({ + externalId: batch[j].extDoc.externalId, + documentId: outcome.value.documentId, + }) if (batch[j].type === 'add') result.docsAdded++ else result.docsUpdated++ } else { @@ -1784,6 +1802,19 @@ export async function processDocOps(input: ProcessDocOpsInput): Promise { }) } } + + if (persisted.length > 0 && input.onBatchPersisted) { + try { + await input.onBatchPersisted(persisted) + } catch (error) { + if (error instanceof SyncLockLostException) throw error + logger.warn('Failed to grant access for a persisted batch — the run end will retry', { + connectorId, + count: persisted.length, + error: toError(error).message, + }) + } + } } }