perf(connectors): hydrate deferred mail and chat documents in batches instead of one at a time

Author: waleedlatif1

apps/sim/connectors/gmail/gmail.ts

@@ -398,6 +398,13 @@ async function resolveLabelNames(
     .filter((name) => !name.startsWith('CATEGORY_') && name !== 'UNREAD')
 }
 
+/**
+ * What a hydrated thread retains: message bodies as text, never attachments,
+ * so a generous ceiling for a long thread is still small. Lets a batch of
+ * threads hydrate together instead of one at a time.
+ */
+const ESTIMATED_THREAD_BYTES = 256 * 1024
+
 /**
  * Creates a lightweight document stub from a thread list entry.
  * Uses metadata-based contentHash for change detection without downloading content.
@@ -412,6 +419,7 @@ function threadToStub(thread: {
     title: thread.snippet || 'Untitled Thread',
     content: '',
     contentDeferred: true,
+    estimatedBytes: ESTIMATED_THREAD_BYTES,
     mimeType: 'text/plain',
     sourceUrl: threadUrl(thread.id),
     contentHash: `gmail:${thread.id}:${thread.historyId ?? ''}`,

apps/sim/connectors/google-chat/google-chat.ts

@@ -13,6 +13,9 @@ import { parseTagDate } from '@/connectors/utils'
 
 const logger = createLogger('GoogleChatConnector')
 
+/** A hydrated space retains a bounded window of message text; lets spaces hydrate in batches. */
+const ESTIMATED_SPACE_BYTES = 1024 * 1024
+
 const CHAT_API_BASE = 'https://chat.googleapis.com/v1'
 const MS_PER_DAY = 24 * 60 * 60 * 1000
 
@@ -218,6 +221,7 @@ function spaceToStub(
     title: spaceTitle(space),
     content: '',
     contentDeferred: true,
+    estimatedBytes: ESTIMATED_SPACE_BYTES,
     mimeType: 'text/plain',
     sourceUrl: space.spaceUri,
     contentHash: buildContentHash(space, maxMessages, lookbackDays, syncContext),

apps/sim/connectors/outlook/outlook.ts

@@ -13,6 +13,9 @@ import {
 
 const logger = createLogger('OutlookConnector')
 
+/** A hydrated conversation retains message bodies as text, never attachments; lets threads hydrate in batches. */
+const ESTIMATED_CONVERSATION_BYTES = 256 * 1024
+
 const GRAPH_API_BASE = 'https://graph.microsoft.com/v1.0/me'
 const MESSAGES_PER_PAGE = 50
 /**
@@ -805,6 +808,7 @@ export const outlookConnector: ConnectorConfig = {
         title: subject,
         content: '',
         contentDeferred: true,
+        estimatedBytes: ESTIMATED_CONVERSATION_BYTES,
         mimeType: 'text/plain',
         sourceUrl,
         contentHash: `outlook:${convId}:${lastDate}`,

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

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

@@ -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(

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

@@ -503,7 +503,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