Fix ordered catch-up completion handoff

Author: Bushstar

services/mediators/hyperswarm/src/hyperswarm-mediator.ts

@@ -49,6 +49,7 @@ import {
     shouldDeferLegacySync,
     shouldSchedulePeriodicRepair,
     shouldStartConnectTimeNegentropy,
+    shouldStartPostOrderedCatchupNegentropy,
 } from './negentropy/policy.js';
 import {
     addAggregateSample,
@@ -415,6 +416,7 @@ const addedPeers: Record<string, number> = {};
 const badPeers: Record<string, number> = {};
 const malformedPeers: Record<string, MalformedPeerState> = {};
 const peerSessions = new Map<string, PeerSyncSession>();
+const orderedCatchupPostImportPeers = new Set<string>();
 const syncStats: MediatorSyncStats = {
     modeSelectionsTotal: 0,
     modeSelectionsLegacy: 0,
@@ -2252,18 +2254,16 @@ async function sendOrderedCatchupPage(peerKey: string, msg: OrderedCatchupReqMes
     syncStats.orderedCatchupOpsSent += push.data.length;
 }
 
-async function completeOrderedCatchup(peerKey: string, session: PeerSyncSession, reason: string): Promise<void> {
+function completeOrderedCatchup(peerKey: string, session: PeerSyncSession, reason: string): void {
     const conn = connectionInfo[peerKey];
     if (!conn) {
         return;
     }
 
     conn.orderedCatchupAttempted = true;
     session.reconciliationComplete = true;
-    await waitForImportQueueIdle('ordered_catchup_complete');
     closePeerSession(peerKey, reason);
-    markNegentropyAdapterDirty();
-    await startNegentropySessionForPeer(peerKey, 'ordered_catchup_complete', 'ordered_catchup_complete');
+    queueOrderedCatchupPostImport(peerKey, reason);
 }
 
 async function waitForImportQueueIdle(reason: string): Promise<void> {
@@ -2280,6 +2280,65 @@ async function waitForImportQueueIdle(reason: string): Promise<void> {
     }
 }
 
+function queueOrderedCatchupPostImport(peerKey: string, reason: string): void {
+    if (orderedCatchupPostImportPeers.has(peerKey)) {
+        log.debug({ peer: shortName(peerKey), reason }, 'ordered catch-up post-import continuation already queued');
+        return;
+    }
+
+    orderedCatchupPostImportPeers.add(peerKey);
+    (async () => {
+        await waitForImportQueueIdle('ordered_catchup_complete');
+        await syncGatekeeperIndexToStore('ordered_catchup_complete');
+        markNegentropyAdapterDirty();
+        await maybeStartPostOrderedCatchupNegentropy(peerKey, reason);
+    })()
+        .catch(error => {
+            log.error({ error, peer: shortName(peerKey), reason }, 'ordered catch-up post-import continuation failed');
+        })
+        .finally(() => {
+            orderedCatchupPostImportPeers.delete(peerKey);
+        });
+}
+
+async function maybeStartPostOrderedCatchupNegentropy(peerKey: string, reason: string): Promise<void> {
+    const conn = connectionInfo[peerKey];
+    const shouldStart = shouldStartPostOrderedCatchupNegentropy({
+        syncMode: conn?.syncMode ?? 'unknown',
+        peerConnected: !!conn,
+        peerSupportsNegentropyTransport: conn ? supportsPeerNegentropyTransport(conn) : false,
+        hasActiveSession: peerSessions.has(peerKey),
+        importQueueLength: importQueue.length(),
+        importQueueRunning: importQueue.running(),
+        activeNegentropySessions: getActiveNegentropySessions(),
+        syncCompleted: conn?.negentropySynced ?? false,
+    });
+
+    if (shouldStart) {
+        await maybeStartPeerSync(peerKey, 'periodic');
+    } else {
+        log.debug(
+            {
+                peer: shortName(peerKey),
+                reason,
+                syncMode: conn?.syncMode ?? 'unknown',
+                peerConnected: !!conn,
+                peerSupportsNegentropyTransport: conn ? supportsPeerNegentropyTransport(conn) : false,
+                hasActiveSession: peerSessions.has(peerKey),
+                importQueueLength: importQueue.length(),
+                importQueueRunning: importQueue.running(),
+                activeNegentropySessions: getActiveNegentropySessions(),
+                syncCompleted: conn?.negentropySynced ?? false,
+            },
+            'post ordered catch-up negentropy handoff deferred'
+        );
+    }
+
+    if (getActiveNegentropySessions() === 0) {
+        await maybeSchedulePreferredSyncs(`ordered_catchup_post_import:${reason}`);
+    }
+}
+
 async function handleOrderedCatchupPush(peerKey: string, msg: OrderedCatchupPushMessage): Promise<void> {
     const session = peerSessions.get(peerKey);
     if (!session || session.mode !== 'ordered_catchup' || session.sessionId !== msg.sessionId) {
@@ -2320,7 +2379,7 @@ async function handleOrderedCatchupPush(peerKey: string, msg: OrderedCatchupPush
         return;
     }
 
-    await completeOrderedCatchup(peerKey, session, 'ordered_catchup_complete');
+    completeOrderedCatchup(peerKey, session, 'ordered_catchup_complete');
 }
 
 async function handleOrderedCatchupDone(peerKey: string, msg: OrderedCatchupDoneMessage): Promise<void> {
@@ -2330,7 +2389,7 @@ async function handleOrderedCatchupDone(peerKey: string, msg: OrderedCatchupDone
         return;
     }
 
-    await completeOrderedCatchup(peerKey, session, 'ordered_catchup_done');
+    completeOrderedCatchup(peerKey, session, 'ordered_catchup_done');
 }
 
 function sendNegMsg(peerKey: string, session: PeerSyncSession, frame: string | Uint8Array): boolean {

services/mediators/hyperswarm/src/negentropy/policy.ts

@@ -12,6 +12,17 @@ export interface RepairSchedulingInput {
     syncCompleted: boolean;
 }
 
+export interface PostOrderedCatchupNegentropyInput {
+    syncMode: SyncMode | 'unknown';
+    peerConnected: boolean;
+    peerSupportsNegentropyTransport: boolean;
+    hasActiveSession: boolean;
+    importQueueLength: number;
+    importQueueRunning: number;
+    activeNegentropySessions: number;
+    syncCompleted: boolean;
+}
+
 export interface LegacySyncPriorityInput {
     syncMode: SyncMode | 'unknown';
     legacySyncEnabled: boolean;
@@ -93,6 +104,38 @@ export function shouldSchedulePeriodicRepair(input: RepairSchedulingInput): bool
     return (input.nowMs - input.lastAttemptAtMs) >= input.repairIntervalMs;
 }
 
+export function shouldStartPostOrderedCatchupNegentropy(input: PostOrderedCatchupNegentropyInput): boolean {
+    if (!input.peerConnected) {
+        return false;
+    }
+
+    if (input.syncMode !== 'negentropy') {
+        return false;
+    }
+
+    if (!input.peerSupportsNegentropyTransport) {
+        return false;
+    }
+
+    if (input.hasActiveSession) {
+        return false;
+    }
+
+    if (input.importQueueLength > 0 || input.importQueueRunning > 0) {
+        return false;
+    }
+
+    if (input.activeNegentropySessions > 0) {
+        return false;
+    }
+
+    if (input.syncCompleted) {
+        return false;
+    }
+
+    return true;
+}
+
 export function shouldDeferLegacySync(input: LegacySyncPriorityInput): boolean {
     if (!input.legacySyncEnabled) {
         return false;

tests/hyperswarm/negentropy-policy.test.ts

@@ -4,6 +4,7 @@ import {
     shouldDeferLegacySync,
     shouldSchedulePeriodicRepair,
     shouldStartConnectTimeNegentropy,
+    shouldStartPostOrderedCatchupNegentropy,
 } from '../../services/mediators/hyperswarm/src/negentropy/policy.ts';
 
 describe('negentropy sync policy', () => {
@@ -57,6 +58,29 @@ describe('negentropy sync policy', () => {
         })).toBe(false);
     });
 
+    it('starts post ordered catch-up negentropy only after import work drains', () => {
+        const base = {
+            syncMode: 'negentropy' as const,
+            peerConnected: true,
+            peerSupportsNegentropyTransport: true,
+            hasActiveSession: false,
+            importQueueLength: 0,
+            importQueueRunning: 0,
+            activeNegentropySessions: 0,
+            syncCompleted: false,
+        };
+
+        expect(shouldStartPostOrderedCatchupNegentropy(base)).toBe(true);
+        expect(shouldStartPostOrderedCatchupNegentropy({ ...base, peerConnected: false })).toBe(false);
+        expect(shouldStartPostOrderedCatchupNegentropy({ ...base, syncMode: 'legacy' })).toBe(false);
+        expect(shouldStartPostOrderedCatchupNegentropy({ ...base, peerSupportsNegentropyTransport: false })).toBe(false);
+        expect(shouldStartPostOrderedCatchupNegentropy({ ...base, hasActiveSession: true })).toBe(false);
+        expect(shouldStartPostOrderedCatchupNegentropy({ ...base, importQueueLength: 1 })).toBe(false);
+        expect(shouldStartPostOrderedCatchupNegentropy({ ...base, importQueueRunning: 1 })).toBe(false);
+        expect(shouldStartPostOrderedCatchupNegentropy({ ...base, activeNegentropySessions: 1 })).toBe(false);
+        expect(shouldStartPostOrderedCatchupNegentropy({ ...base, syncCompleted: true })).toBe(false);
+    });
+
     it('defers legacy while negentropy peers are pending, but allows fallback after timeout', () => {
         const base = {
             syncMode: 'legacy' as const,