@@ -10,7 +10,22 @@ vi.mock('@/lib/knowledge/documents/utils', () => ({
1010 VALIDATE_RETRY_OPTIONS : { } ,
1111} ) )
1212vi . mock ( '@/components/icons' , ( ) => ( { GmailIcon : ( ) => null } ) )
13+ vi . mock ( '@/lib/knowledge/documents/service' , ( ) => ( {
14+ isTriggerAvailable : ( ) => false ,
15+ processDocumentsWithQueue : vi . fn ( ) ,
16+ } ) )
17+ vi . mock ( '@/lib/knowledge/connectors/sync-persistence' , ( ) => ( {
18+ addDocument : vi . fn ( ) ,
19+ persistSkippedDocuments : vi . fn ( ) ,
20+ persistSkippedRetryHashes : vi . fn ( ) ,
21+ updateDocument : vi . fn ( ) ,
22+ } ) )
1323
24+ import {
25+ classifyExternalDoc ,
26+ mergeHydratedSkippedDocument ,
27+ shouldReplaceExistingWithSkippedDocument ,
28+ } from '@/lib/knowledge/connectors/sync-primitives'
1429import { gmailConnector } from '@/connectors/gmail/gmail'
1530import { DEFAULT_MAX_THREADS , gmailConnectorMeta } from '@/connectors/gmail/meta'
1631import { CONNECTOR_TEXT_DOCUMENT_MAX_BYTES , PER_MEMBER_LISTING_CONTEXT } from '@/connectors/utils'
@@ -262,43 +277,51 @@ function mockExternalBodyThread(
262277}
263278
264279describe ( 'Gmail full-thread response budget' , ( ) => {
265- it ( 'rejects an oversized Content-Length before reading the thread ' , async ( ) => {
280+ it ( 'records a versioned size skip without reading oversized Content-Length bodies ' , async ( ) => {
266281 const pull = vi . fn ( ( controller : ReadableStreamDefaultController < Uint8Array > ) => {
267282 controller . enqueue ( Buffer . from ( JSON . stringify ( threadFixture ( ) ) ) )
268283 controller . close ( )
269284 } )
270285 const cancel = vi . fn ( )
271- mockFetchWithRetry . mockResolvedValue (
272- new Response ( new ReadableStream ( { pull, cancel } , { highWaterMark : 0 } ) , {
273- headers : { 'Content-Length' : String ( 32 * 1024 * 1024 + 1 ) } ,
274- } )
286+ mockFetchWithRetry . mockImplementation ( async ( url : string ) =>
287+ new URL ( url ) . searchParams . get ( 'format' ) === 'minimal'
288+ ? Response . json ( { id : 'thread-1' , historyId : '10' } )
289+ : new Response ( new ReadableStream ( { pull, cancel } , { highWaterMark : 0 } ) , {
290+ headers : { 'Content-Length' : String ( 32 * 1024 * 1024 + 1 ) } ,
291+ } )
275292 )
276293
277- await expect ( gmailConnector . getDocument ( 'token' , { } , 'thread-1' ) ) . rejects . toMatchObject ( {
278- name : 'PayloadSizeLimitError' ,
279- maxBytes : 32 * 1024 * 1024 ,
280- observedBytes : 32 * 1024 * 1024 + 1 ,
294+ await expect ( gmailConnector . getDocument ( 'token' , { } , 'thread-1' ) ) . resolves . toMatchObject ( {
295+ externalId : 'thread-1' ,
296+ contentHash : 'gmail:thread-1:10:body-v2' ,
297+ content : '' ,
298+ contentDeferred : false ,
299+ skippedExistingDisposition : 'replace' ,
300+ skippedReason : 'File exceeds the 32MB size limit and was not indexed' ,
281301 } )
282302 expect ( pull ) . not . toHaveBeenCalled ( )
283- expect ( cancel ) . toHaveBeenCalledOnce ( )
284- expect ( mockFetchWithRetry ) . toHaveBeenCalledTimes ( 1 )
303+ expect ( cancel ) . toHaveBeenCalledTimes ( 2 )
304+ expect ( mockFetchWithRetry ) . toHaveBeenCalledTimes ( 4 )
285305 } )
286306
287- it ( 'cancels chunked oversized JSON before consuming the complete thread ' , async ( ) => {
288- let chunksRead = 0
307+ it ( 'cancels chunked oversized JSON and skips only after verifying a stable revision ' , async ( ) => {
308+ const chunksRead : number [ ] = [ ]
289309 const chunk = Buffer . alloc ( 1024 * 1024 , 'a' )
290310 const cancel = vi . fn ( )
291- mockFetchWithRetry . mockResolvedValue (
292- new Response (
311+ mockFetchWithRetry . mockImplementation ( async ( url : string ) => {
312+ if ( new URL ( url ) . searchParams . get ( 'format' ) === 'minimal' )
313+ return Response . json ( { id : 'thread-1' , historyId : '10' } )
314+ const index = chunksRead . push ( 0 ) - 1
315+ return new Response (
293316 new ReadableStream (
294317 {
295318 pull ( controller ) {
296- chunksRead += 1
297- if ( chunksRead === 1 ) {
319+ chunksRead [ index ] += 1
320+ if ( chunksRead [ index ] === 1 ) {
298321 controller . enqueue (
299322 Buffer . from ( '{"id":"thread-1","historyId":"10","messages":[],"padding":"' )
300323 )
301- } else if ( chunksRead <= 35 ) {
324+ } else if ( chunksRead [ index ] <= 35 ) {
302325 controller . enqueue ( chunk )
303326 } else {
304327 controller . enqueue ( Buffer . from ( '"}' ) )
@@ -310,14 +333,141 @@ describe('Gmail full-thread response budget', () => {
310333 { highWaterMark : 0 }
311334 )
312335 )
336+ } )
337+
338+ await expect ( gmailConnector . getDocument ( 'token' , { } , 'thread-1' ) ) . resolves . toMatchObject ( {
339+ content : '' ,
340+ skippedReason : 'File exceeds the 32MB size limit and was not indexed' ,
341+ contentHash : 'gmail:thread-1:10:body-v2' ,
342+ } )
343+ expect ( cancel ) . toHaveBeenCalledTimes ( 2 )
344+ expect ( chunksRead ) . toHaveLength ( 2 )
345+ expect ( chunksRead . every ( ( count ) => count < 35 ) ) . toBe ( true )
346+ expect ( mockFetchWithRetry ) . toHaveBeenCalledTimes ( 4 )
347+ } )
348+
349+ it ( 'replaces stale content once, resumes on source change, and preserves member isolation' , async ( ) => {
350+ let historyId = '10'
351+ mockFetchWithRetry . mockImplementation ( async ( url : string , init ?: RequestInit ) => {
352+ const parsed = new URL ( url )
353+ expect ( new Headers ( init ?. headers ) . get ( 'Authorization' ) ) . toBe ( 'Bearer alice-token' )
354+ if ( parsed . pathname . endsWith ( '/threads' ) )
355+ return Response . json ( { threads : [ { id : 'thread-1' , historyId } ] } )
356+ if ( parsed . searchParams . get ( 'format' ) === 'minimal' )
357+ return Response . json ( { id : 'thread-1' , historyId } )
358+ return new Response ( null , {
359+ headers : { 'Content-Length' : String ( 32 * 1024 * 1024 + 1 ) } ,
360+ } )
361+ } )
362+ const context = memberContext ( 'alice' )
363+ const [ stub ] = ( await gmailConnector . listDocuments ( 'alice-token' , { } , undefined , context ) )
364+ . documents
365+ const skipped = await gmailConnector . getDocument ( 'alice-token' , { } , stub . externalId , context )
366+ expect ( skipped ?. externalId ) . toBe ( 'member:alice:thread-1' )
367+ expect ( shouldReplaceExistingWithSkippedDocument ( { storageKey : 'old.txt' } , skipped ! ) ) . toBe ( true )
368+ const merged = mergeHydratedSkippedDocument ( stub , skipped ! )
369+ const stored = { id : 'stored' , contentHash : merged . contentHash , storageKey : null }
370+ expect ( classifyExternalDoc ( stub , stored ) ) . toEqual ( { type : 'unchanged' } )
371+ expect ( classifyExternalDoc ( stub , stored , true ) ) . toEqual ( {
372+ type : 'update' ,
373+ existingId : 'stored' ,
374+ } )
375+ historyId = '11'
376+ const [ updated ] = (
377+ await gmailConnector . listDocuments ( 'alice-token' , { } , undefined , memberContext ( 'alice' ) )
378+ ) . documents
379+ expect ( classifyExternalDoc ( updated , stored ) ) . toEqual ( { type : 'update' , existingId : 'stored' } )
380+ mockFetchWithRetry . mockClear ( )
381+ expect (
382+ await gmailConnector . getDocument ( 'alice-token' , { } , stub . externalId , memberContext ( 'bob' ) )
383+ ) . toBeNull ( )
384+ expect ( mockFetchWithRetry ) . not . toHaveBeenCalled ( )
385+ } )
386+
387+ it ( 'does not cache a size skip for a revision that changes during the bounded retry' , async ( ) => {
388+ let metadataReads = 0
389+ mockFetchWithRetry . mockImplementation ( async ( url : string ) =>
390+ new URL ( url ) . searchParams . get ( 'format' ) === 'minimal'
391+ ? Response . json ( { id : 'thread-1' , historyId : String ( 10 + metadataReads ++ ) } )
392+ : new Response ( null , {
393+ headers : { 'Content-Length' : String ( 32 * 1024 * 1024 + 1 ) } ,
394+ } )
395+ )
396+ await expect ( gmailConnector . getDocument ( 'token' , { } , 'thread-1' ) ) . rejects . toThrow (
397+ 'Gmail thread changed while checking its size'
313398 )
399+ expect ( mockFetchWithRetry ) . toHaveBeenCalledTimes ( 4 )
400+ } )
314401
315- await expect ( gmailConnector . getDocument ( 'token' , { } , 'thread-1' ) ) . rejects . toMatchObject ( {
316- name : 'PayloadSizeLimitError' ,
317- maxBytes : 32 * 1024 * 1024 ,
402+ it ( 'hydrates a thread that becomes small enough during the bounded retry' , async ( ) => {
403+ let fullReads = 0
404+ mockFetchWithRetry . mockImplementation ( async ( url : string ) => {
405+ const parsed = new URL ( url )
406+ if ( parsed . pathname . endsWith ( '/labels' ) ) return Response . json ( { labels : [ ] } )
407+ if ( parsed . searchParams . get ( 'format' ) === 'minimal' )
408+ return Response . json ( { id : 'thread-1' , historyId : '11' } )
409+ if ( fullReads ++ === 0 )
410+ return new Response ( null , {
411+ headers : { 'Content-Length' : String ( 32 * 1024 * 1024 + 1 ) } ,
412+ } )
413+ return Response . json ( threadFixture ( '11' , 'A smaller current thread' ) )
318414 } )
319- expect ( cancel ) . toHaveBeenCalledOnce ( )
320- expect ( chunksRead ) . toBeLessThan ( 35 )
415+ const document = await gmailConnector . getDocument ( 'token' , { } , 'thread-1' )
416+ expect ( document ?. content ) . toContain ( 'A smaller current thread' )
417+ expect ( document ?. skippedReason ) . toBeUndefined ( )
418+ expect ( document ?. contentHash ) . toBe ( 'gmail:thread-1:11:body-v2' )
419+ expect ( fullReads ) . toBe ( 2 )
420+ } )
421+
422+ it . each ( [ 401 , 429 , 503 ] ) (
423+ 'preserves metadata HTTP %s failure instead of caching a skip' ,
424+ async ( status ) => {
425+ mockFetchWithRetry . mockImplementation ( async ( url : string ) =>
426+ new URL ( url ) . searchParams . get ( 'format' ) === 'minimal'
427+ ? new Response ( null , { status } )
428+ : new Response ( null , {
429+ headers : { 'Content-Length' : String ( 32 * 1024 * 1024 + 1 ) } ,
430+ } )
431+ )
432+ await expect ( gmailConnector . getDocument ( 'token' , { } , 'thread-1' ) ) . rejects . toMatchObject ( {
433+ status,
434+ } )
435+ expect ( mockFetchWithRetry ) . toHaveBeenCalledTimes ( 2 )
436+ }
437+ )
438+
439+ it . each ( [ { } , { id : 'thread-1' } , { id : 'other-thread' , historyId : '10' } ] ) (
440+ 'rejects unverified metadata instead of synthesizing a skip revision: %j' ,
441+ async ( metadata ) => {
442+ mockFetchWithRetry . mockImplementation ( async ( url : string ) =>
443+ new URL ( url ) . searchParams . get ( 'format' ) === 'minimal'
444+ ? Response . json ( metadata )
445+ : new Response ( null , {
446+ headers : { 'Content-Length' : String ( 32 * 1024 * 1024 + 1 ) } ,
447+ } )
448+ )
449+ await expect ( gmailConnector . getDocument ( 'token' , { } , 'thread-1' ) ) . rejects . toThrow (
450+ 'Gmail returned malformed thread metadata'
451+ )
452+ expect ( mockFetchWithRetry ) . toHaveBeenCalledTimes ( 2 )
453+ }
454+ )
455+
456+ it ( 'returns null when the oversized thread disappears before revision verification' , async ( ) => {
457+ mockFetchWithRetry . mockImplementation ( async ( url : string ) =>
458+ new URL ( url ) . searchParams . get ( 'format' ) === 'minimal'
459+ ? new Response ( null , { status : 404 } )
460+ : new Response ( null , {
461+ headers : { 'Content-Length' : String ( 32 * 1024 * 1024 + 1 ) } ,
462+ } )
463+ )
464+ await expect ( gmailConnector . getDocument ( 'token' , { } , 'thread-1' ) ) . resolves . toBeNull ( )
465+ expect ( mockFetchWithRetry ) . toHaveBeenCalledTimes ( 2 )
466+ } )
467+
468+ it ( 'does not turn a missing response body into a permanent size skip' , async ( ) => {
469+ mockFetchWithRetry . mockResolvedValueOnce ( new Response ( null ) )
470+ await expect ( gmailConnector . getDocument ( 'token' , { } , 'thread-1' ) ) . rejects . toThrow ( )
321471 expect ( mockFetchWithRetry ) . toHaveBeenCalledTimes ( 1 )
322472 } )
323473
@@ -505,6 +655,8 @@ describe('Gmail separately stored message bodies', () => {
505655 content : '' ,
506656 contentDeferred : false ,
507657 skippedReason : 'File exceeds the 12MB size limit and was not indexed' ,
658+ skippedExistingDisposition : 'replace' ,
659+ skippedRetryPolicy : 'source-change' ,
508660 } )
509661 } )
510662
0 commit comments