99 user ,
1010 workspace ,
1111} from '@sim/db/schema'
12+ import { createDeferred } from '@sim/testing/helpers/deferred'
1213import { generateId } from '@sim/utils/id'
1314import { and , eq , inArray , isNotNull , sql } from 'drizzle-orm'
1415import { afterAll , afterEach , beforeEach , describe , expect , it , vi } from 'vitest'
@@ -30,7 +31,7 @@ import {
3031} from '@/lib/knowledge/__integration__/seed-source-access-fixture'
3132import { deferConnectorSync } from '@/lib/knowledge/connectors/sync-deferral'
3233import { completeSuccessfulSync } from '@/lib/knowledge/connectors/sync-engine'
33- import { createContentSyncLease } from '@/lib/knowledge/connectors/sync-lock'
34+ import { createContentSyncLease , SyncLockLostException } from '@/lib/knowledge/connectors/sync-lock'
3435import { sweepStuckDocuments } from '@/lib/knowledge/connectors/sync-primitives'
3536import { deleteKnowledgeBase } from '@/lib/knowledge/service'
3637import { GitHubRequestDeferredError } from '@/connectors/github/request'
@@ -40,14 +41,6 @@ const WAIT_OPTIONS = { interval: 5, timeout: 5000 }
4041const ACTIONS = [ 'complete' , 'defer' , 'recover' ] as const
4142type Action = ( typeof ACTIONS ) [ number ]
4243
43- function deferred ( ) {
44- let resolve ! : ( ) => void
45- const promise = new Promise < void > ( ( ready ) => {
46- resolve = ready
47- } )
48- return { promise, resolve }
49- }
50-
5144function emptyResult ( ) : SyncResult {
5245 return {
5346 docsAdded : 0 ,
@@ -156,29 +149,37 @@ describe('source lifecycle KB guards', () => {
156149 }
157150
158151 function holdCommit ( action : Action ) {
159- const gate = { pid : undefined as number | undefined , release : deferred ( ) }
152+ const gate = {
153+ pid : undefined as number | undefined ,
154+ release : createDeferred < void > ( ) ,
155+ resumeAfterCommit : createDeferred < void > ( ) ,
156+ }
160157 const transaction = db . transaction . bind ( db )
161- vi . spyOn ( db , 'transaction' ) . mockImplementation ( ( callback , config ) =>
162- transaction ( async ( tx ) => {
158+ vi . spyOn ( db , 'transaction' ) . mockImplementation ( async ( callback , config ) => {
159+ let heldCommit = false
160+ const result = await transaction ( async ( tx ) => {
163161 await tx . execute ( sql `SET LOCAL statement_timeout = '10s'` )
164162 await tx . execute ( sql `SET LOCAL idle_in_transaction_session_timeout = '15s'` )
165163 const result = await callback ( tx )
166164 if ( gate . pid === undefined && ( await wrote ( action , tx ) ) ) {
167165 const [ backend ] = await tx . execute < { pid : number } > ( sql `SELECT pg_backend_pid() AS pid` )
168166 gate . pid = backend . pid
167+ heldCommit = true
169168 await gate . release . promise
170169 }
171170 return result
172171 } , config )
173- )
172+ if ( heldCommit && action === 'recover' ) await gate . resumeAfterCommit . promise
173+ return result
174+ } )
174175 return gate
175176 }
176177
177178 it . each ( ACTIONS ) (
178179 '%s can commit while an unrelated embedding holds the KB key-share lock' ,
179180 async ( action ) => {
180- const release = deferred ( )
181- const acquired = deferred ( )
181+ const release = createDeferred < void > ( )
182+ const acquired = createDeferred < void > ( )
182183 const holding = db . transaction ( async ( tx ) => {
183184 await tx . execute ( sql `SET LOCAL idle_in_transaction_session_timeout = '15s'` )
184185 await tx
@@ -228,15 +229,23 @@ describe('source lifecycle KB guards', () => {
228229 . where ( eq ( knowledgeBase . id , ids . knowledgeBaseId ) )
229230 expect ( active . deletedAt ) . toBeNull ( )
230231 gate . release . resolve ( )
231- expect ( await completed ) . toMatchObject ( [ { status : 'fulfilled' } ] )
232232 expect ( await deleted ) . toMatchObject ( [ { status : 'fulfilled' } ] )
233+ gate . resumeAfterCommit . resolve ( )
234+ if ( action === 'recover' ) {
235+ expect ( await completed ) . toMatchObject ( [
236+ { status : 'rejected' , reason : expect . any ( SyncLockLostException ) } ,
237+ ] )
238+ } else {
239+ expect ( await completed ) . toMatchObject ( [ { status : 'fulfilled' } ] )
240+ }
233241 const [ archived ] = await db
234242 . select ( { id : document . id } )
235243 . from ( document )
236244 . where ( and ( eq ( document . id , retryDocumentId ) , isNotNull ( document . archivedAt ) ) )
237245 expect ( archived ?. id ) . toBe ( retryDocumentId )
238246 } finally {
239247 gate . release . resolve ( )
248+ gate . resumeAfterCommit . resolve ( )
240249 await completed
241250 await deleted
242251 }
0 commit comments