@@ -52,6 +52,7 @@ import {
5252 MEMBER_SYNC_SOFT_BUDGET_SECONDS ,
5353} from '@/lib/knowledge/connectors/sync-limits'
5454import {
55+ assertSyncLeaseHeldInTx ,
5556 createMemberSyncLease ,
5657 holdsMemberSyncLockToken ,
5758 MEMBER_LOCKABLE_CONNECTOR_STATUSES ,
@@ -660,15 +661,22 @@ async function reconcileMembership(
660661 * member be aborted at the deadline without touching the others.
661662 */
662663async function claimNextMember ( run : MemberSyncRun ) : Promise < MemberRow | null > {
663- const [ claimed ] = await db
664- . update ( knowledgeConnectorMember )
665- . set ( { lastStartedAt : new Date ( ) , updatedAt : new Date ( ) } )
666- . where (
667- and (
668- eq ( knowledgeConnectorMember . connectorId , run . connectorId ) ,
669- eq (
670- knowledgeConnectorMember . id ,
671- sql `(
664+ /**
665+ * Proved under the lease: a run reclaimed while it slept must not stamp
666+ * `lastStartedAt`, which would hide the member from its replacement's
667+ * selection and defer that member's access updates to a later run.
668+ */
669+ const [ claimed ] = await db . transaction ( async ( tx ) => {
670+ await assertSyncLeaseHeldInTx ( tx , run . connectorId , run . lease )
671+ return tx
672+ . update ( knowledgeConnectorMember )
673+ . set ( { lastStartedAt : new Date ( ) , updatedAt : new Date ( ) } )
674+ . where (
675+ and (
676+ eq ( knowledgeConnectorMember . connectorId , run . connectorId ) ,
677+ eq (
678+ knowledgeConnectorMember . id ,
679+ sql `(
672680 SELECT ${ knowledgeConnectorMember . id } FROM ${ knowledgeConnectorMember }
673681 WHERE ${ knowledgeConnectorMember . connectorId } = ${ run . connectorId }
674682 AND ${ knowledgeConnectorMember . status } = 'active'
@@ -678,10 +686,11 @@ async function claimNextMember(run: MemberSyncRun): Promise<MemberRow | null> {
678686 LIMIT 1
679687 FOR UPDATE SKIP LOCKED
680688 )`
689+ )
681690 )
682691 )
683- )
684- . returning ( )
692+ . returning ( )
693+ } )
685694 return claimed ?? null
686695}
687696
0 commit comments