diff --git a/gitnexus/src/core/group/sync.ts b/gitnexus/src/core/group/sync.ts index 5c535f29f..a329500be 100644 Binary files a/gitnexus/src/core/group/sync.ts and b/gitnexus/src/core/group/sync.ts differ diff --git a/gitnexus/src/core/lbug/pool-adapter.ts b/gitnexus/src/core/lbug/pool-adapter.ts index e5338e3d0..11b1f7690 100644 --- a/gitnexus/src/core/lbug/pool-adapter.ts +++ b/gitnexus/src/core/lbug/pool-adapter.ts @@ -103,6 +103,31 @@ const IDLE_TIMEOUT_MS = 5 * 60 * 1000; // 5 minutes /** Max connections per repo (caps concurrent queries per repo) */ const MAX_CONNS_PER_REPO = 8; +/** + * Repos exempt from AUTOMATIC eviction (LRU + idle timeout) until explicitly + * unpinned. Used by bounded multi-repo operations like `group sync`, which + * initializes one pool per repo and then resolves cross-repo manifest/workspace + * links against ALL of those pools after the init loop. Without pinning, a + * group larger than MAX_POOL_SIZE would LRU-evict the earliest repos before + * resolution runs, leaving the deferred executor closures pointing at dead pool + * entries (issue #2189). + * + * Pins are REFERENCE-COUNTED: the map holds repoId → active lease count. This + * lets overlapping holders (two windows of one sync, or two concurrent + * `group sync` calls sharing a repo) coexist safely — the repo stays exempt + * until the LAST holder releases. A boolean Set could not represent "two + * holders," so the first release would wrongly clear a pin another holder still + * needs (PR #2191 review, Finding 1). + * + * Pins block only automatic eviction (LRU + idle). Explicit teardown + * (closeOne / closeLbug) always closes the entry and force-clears its count — + * teardown is authoritative. A present key always means count ≥ 1. While every + * pooled repo is pinned, evictLRU finds no eligible victim and the pool may + * transiently exceed MAX_POOL_SIZE — the same soft-cap behavior that already + * occurs when every entry is checked out. + */ +const pinnedRepos = new Map(); + // Behavior-neutral RSS tracing for the FTS evict→reload memory repro // (gitnexus/scripts/bench/fts-evict-reload-rss.mjs). Two invariants keep it safe // in the pool init/close hot path: it writes ONLY to stderr (stdout is the MCP @@ -145,6 +170,7 @@ function ensureIdleTimer(): void { idleTimer = setInterval(() => { const now = Date.now(); for (const [repoId, entry] of pool) { + if (pinnedRepos.has(repoId)) continue; if (now - entry.lastUsed > IDLE_TIMEOUT_MS && entry.checkedOut === 0) { closeOne(repoId); } @@ -167,7 +193,64 @@ export const touchRepo = (repoId: string): void => { }; /** - * Evict the least-recently-used repo if pool is at capacity + * Acquire one eviction-exemption lease on a repo (LRU + idle timeout) by + * incrementing its reference count. The repoId must match the key passed to + * initLbug (e.g. group sync leases by handle.id — the same id it inits with). + * Leasing a repoId before it enters the pool is allowed and protects the entry + * once it is created, but the lease does NOT survive a teardown: closeOne + * force-clears the count, so a later re-init of the same repoId starts + * unpinned. Each pinRepo MUST be balanced by exactly one release (the repo + * stays exempt until the last lease is released). See the pinnedRepos docstring + * for the full contract. + * + * Returns a `release` disposer (mirroring addPoolCloseListener) that releases + * THIS lease exactly once — calling it twice is a no-op, so it can never + * over-decrement a sibling holder's count. Prefer the disposer + * (`const release = pinRepo(id); try { … } finally { release(); }`) so the + * pin/release pair is leak-proof; unpinRepo remains available for callers that + * pair explicitly. + */ +export const pinRepo = (repoId: string): (() => void) => { + pinnedRepos.set(repoId, (pinnedRepos.get(repoId) ?? 0) + 1); + let released = false; + return () => { + if (released) return; + released = true; + unpinRepo(repoId); + }; +}; + +/** + * Release one eviction-exemption lease on a repo. The repo becomes eligible for + * automatic eviction again only once its count reaches 0 (the key is deleted). + * Idempotent at the floor: releasing a repo with no active lease is a no-op (no + * negative counts). Does NOT close the repo's pool. + */ +export const unpinRepo = (repoId: string): void => { + const count = pinnedRepos.get(repoId); + if (count === undefined) return; + if (count <= 1) { + pinnedRepos.delete(repoId); + } else { + pinnedRepos.set(repoId, count - 1); + } +}; + +/** + * Maximum number of repos a bounded multi-repo operation (e.g. group sync's + * windowed manifest resolution) should hold resident at once. Equals + * MAX_POOL_SIZE today, but exposed under an intent-named accessor so callers + * size their working set against "max repos a bounded op should hold" rather + * than coupling to the LRU eviction-cap constant, which may be tuned + * independently. + */ +export const getMaxResidentRepos = (): number => MAX_POOL_SIZE; + +/** + * Evict the least-recently-used repo if pool is at capacity. + * Pinned repos are never chosen as the eviction victim — when every eligible + * entry is pinned, no eviction occurs and the pool transiently exceeds + * MAX_POOL_SIZE (see the pinnedRepos docstring). */ function evictLRU(): void { if (pool.size < MAX_POOL_SIZE) return; @@ -175,6 +258,7 @@ function evictLRU(): void { let oldestId: string | null = null; let oldestTime = Infinity; for (const [id, entry] of pool) { + if (pinnedRepos.has(id)) continue; if (entry.checkedOut === 0 && entry.lastUsed < oldestTime) { oldestTime = entry.lastUsed; oldestId = id; @@ -244,6 +328,11 @@ function closeOne(repoId: string): void { pool.delete(repoId); + // Clear any eviction pin — the entry is gone, so the pin is meaningless and + // would otherwise leak across operations in a long-lived process. Teardown + // is authoritative: an explicit close always wins over a pin. + pinnedRepos.delete(repoId); + // Notify listeners AFTER the pool entry is gone so any cache-invalidation // they perform is consistent with `isLbugReady(repoId) === false`. for (const listener of poolCloseListeners) { diff --git a/gitnexus/test/unit/group/sync-windowed-resolution.test.ts b/gitnexus/test/unit/group/sync-windowed-resolution.test.ts new file mode 100644 index 000000000..d7cdb4090 --- /dev/null +++ b/gitnexus/test/unit/group/sync-windowed-resolution.test.ts @@ -0,0 +1,243 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { mkdtempSync, writeFileSync, mkdirSync, rmSync } from 'node:fs'; +import type { GroupConfig, GroupManifestLink } from '../../../src/core/group/types.js'; + +// Two test surfaces for the windowed manifest resolution (issue #2189 / PR #2191 +// review, Finding 3 — bound peak pool residency to MAX_POOL_SIZE regardless of +// group size): +// +// 1. partitionManifestWindows — a pure function; the bounded-residency logic +// lives here (every window references <= maxResident repos, every link in +// exactly one window). Tested directly, no pool. +// 2. A real-pool integration test that drives syncGroup through the actual +// pool (native LadybugDB layer mocked, as in lbug-pool-pinning.test.ts) and +// asserts the count of concurrently-open Databases never exceeds the +// resident cap — the end-to-end residency bound the review flagged as +// missing. + +// ── Surface 1: pure partition function ────────────────────────────────────── + +describe('partitionManifestWindows (issue #2189 windowed resolution)', () => { + const link = (from: string, to: string): GroupManifestLink => ({ + from, + to, + type: 'http', + contract: `GET::/${from}-${to}`, + role: 'consumer', + }); + + it('keeps every window within maxResident repos and places every link exactly once', async () => { + const { partitionManifestWindows } = await import('../../../src/core/group/sync.js'); + const repos = ['r1', 'r2', 'r3', 'r4', 'r5', 'r6', 'r7', 'r8']; + const known = new Set(repos); + // A star: every leaf links to the hub r1, plus a few leaf-leaf links. + const links = [ + link('r1', 'r2'), + link('r1', 'r3'), + link('r1', 'r4'), + link('r1', 'r5'), + link('r1', 'r6'), + link('r7', 'r8'), + link('r2', 'r3'), + ]; + const maxResident = 5; + const windows = partitionManifestWindows(links, known, maxResident); + + // Bounded residency: no window references more than maxResident repos. + for (const w of windows) expect(w.repos.size).toBeLessThanOrEqual(maxResident); + + // True partition: every link appears in exactly one window. + const placed = windows.flatMap((w) => w.links); + expect(placed).toHaveLength(links.length); + const placedKeys = placed.map((l) => `${l.from}->${l.to}`).sort(); + const inputKeys = links.map((l) => `${l.from}->${l.to}`).sort(); + expect(placedKeys).toEqual(inputKeys); + // No link appears twice (the contract-dedup invariant — KTD-4). + expect(new Set(placedKeys).size).toBe(placedKeys.length); + }); + + it('counts only in-group repos toward a window; dangling links consume no budget', async () => { + const { partitionManifestWindows } = await import('../../../src/core/group/sync.js'); + const known = new Set(['r1']); + const links = [ + link('r1', 'external-a'), // 1 in-group repo + link('external-b', 'external-c'), // 0 in-group repos (fully dangling) + ]; + const windows = partitionManifestWindows(links, known, 5); + // Both links are still placed (so they yield synthetic-UID contracts)... + expect(windows.flatMap((w) => w.links)).toHaveLength(2); + // ...but the only repo counted is r1. + const allRepos = new Set(windows.flatMap((w) => [...w.repos])); + expect(allRepos).toEqual(new Set(['r1'])); + }); + + it('returns no windows for an empty link set', async () => { + const { partitionManifestWindows } = await import('../../../src/core/group/sync.js'); + expect(partitionManifestWindows([], new Set(['r1']), 5)).toEqual([]); + }); + + it('splits links across multiple windows when referenced repos exceed maxResident', async () => { + const { partitionManifestWindows } = await import('../../../src/core/group/sync.js'); + const known = new Set(['r1', 'r2', 'r3', 'r4', 'r5', 'r6']); + // 3 disjoint repo-pairs = 6 distinct repos; maxResident 2 forces ≥3 windows. + const links = [link('r1', 'r2'), link('r3', 'r4'), link('r5', 'r6')]; + const windows = partitionManifestWindows(links, known, 2); + expect(windows.length).toBeGreaterThanOrEqual(3); + for (const w of windows) expect(w.repos.size).toBeLessThanOrEqual(2); + expect(windows.flatMap((w) => w.links)).toHaveLength(3); + }); +}); + +// ── Surface 2: real-pool residency bound through syncGroup ─────────────────── + +const { loadFTSExtensionMock, openCounter } = vi.hoisted(() => ({ + loadFTSExtensionMock: vi.fn(), + openCounter: { live: 0, peak: 0 }, +})); + +vi.mock('@ladybugdb/core', () => ({ + default: { + Database: vi.fn(), + Connection: vi.fn(function (this: any) { + this.query = vi.fn().mockResolvedValue({ + getAll: vi.fn().mockResolvedValue([]), + close: vi.fn(), + }); + // executeParameterized's prepare/execute path (manifest resolveSymbol). + this.prepare = vi.fn().mockResolvedValue({ + isSuccess: () => true, + getErrorMessage: vi.fn().mockResolvedValue(''), + }); + this.execute = vi.fn().mockResolvedValue({ + getAll: vi.fn().mockResolvedValue([]), + close: vi.fn(), + }); + this.close = vi.fn().mockResolvedValue(undefined); + }), + }, +})); + +vi.mock('../../../src/core/lbug/lbug-adapter.js', () => ({ + isReadOnlyDbError: vi.fn(() => false), + loadFTSExtension: loadFTSExtensionMock, +})); + +vi.mock('../../../src/core/lbug/lbug-config.js', () => ({ + // Track concurrently-open Databases: a fresh fake per open, decrement on close. + createLbugDatabase: vi.fn(() => { + openCounter.live += 1; + openCounter.peak = Math.max(openCounter.peak, openCounter.live); + return { + init: vi.fn().mockResolvedValue(undefined), + close: vi.fn().mockImplementation(async () => { + openCounter.live -= 1; + }), + }; + }), + toNativeSafePath: vi.fn((p: string) => p), + isWalCorruptionError: vi.fn(() => false), + WAL_RECOVERY_SUGGESTION: '', +})); + +vi.mock('../../../src/core/lbug/sidecar-recovery.js', () => ({ + preflightLbugSidecars: vi.fn().mockResolvedValue(undefined), + isMissingFsError: vi.fn(() => false), + isMissingShadowSidecarError: vi.fn(() => false), + isReadOnlyShadowReplayError: vi.fn(() => false), + quarantineWalForMissingShadow: vi.fn().mockResolvedValue(''), + renameFailureMessage: vi.fn((p: string) => `rename failed for ${p}`), + statIfExists: vi.fn().mockResolvedValue(null), +})); + +// readRegistry is called in syncGroup's else branch; resolveRepoHandle is +// supplied, so an empty registry is fine (only the meta.json fallback reads it). +vi.mock('../../../src/storage/repo-manager.js', () => ({ + readRegistry: vi.fn().mockResolvedValue([]), +})); + +const { syncGroup } = await import('../../../src/core/group/sync.js'); +const { closeLbug, getMaxResidentRepos } = await import('../../../src/core/lbug/pool-adapter.js'); + +describe('syncGroup windowed resolution bounds pool residency (real pool, #2189)', () => { + let tmpRoot: string; + + beforeEach(() => { + tmpRoot = mkdtempSync(path.join(os.tmpdir(), 'gn-window-resid-')); + loadFTSExtensionMock.mockResolvedValue(true); + openCounter.live = 0; + openCounter.peak = 0; + }); + + afterEach(async () => { + await closeLbug().catch(() => {}); + rmSync(tmpRoot, { recursive: true, force: true }); + }); + + it('never holds more than getMaxResidentRepos() Databases open for a large group', async () => { + const maxResident = getMaxResidentRepos(); + const repoCount = maxResident + 4; // exceed the cap so windowing must split + + const repos: Record = {}; + const links: GroupManifestLink[] = []; + for (let i = 1; i <= repoCount; i++) { + const gp = `app/repo-${i}`; + repos[gp] = `repo-${i}`; + // Star topology: every repo links to repo-1 → many windows reference repo-1. + if (i > 1) { + links.push({ + from: gp, + to: 'app/repo-1', + type: 'http', + contract: `GET::/api/${i}`, + role: 'consumer', + }); + } + } + + const config: GroupConfig = { + version: 1, + name: 'test', + description: '', + repos, + links, + packages: {}, + // All detection off → init loop just opens pools (no extractor file reads). + detect: { + http: false, + grpc: false, + thrift: false, + topics: false, + shared_libs: false, + embedding_fallback: false, + workspace_deps: false, + }, + matching: { bm25_threshold: 0.7, embedding_threshold: 0.65, max_candidates_per_step: 3 }, + }; + + await syncGroup(config, { + resolveRepoHandle: async (_name, groupPath) => { + // Each repo gets a real storage dir with a fake lbug file so fs.stat in + // doInitLbug succeeds; distinct paths → distinct Databases. + const storagePath = path.join(tmpRoot, groupPath); + mkdirSync(storagePath, { recursive: true }); + writeFileSync(path.join(storagePath, 'lbug'), ''); + return { + id: groupPath.replace(/\//g, '-'), + path: groupPath, + repoPath: storagePath, + storagePath, + }; + }, + skipWrite: true, + }); + + // The init loop (no pin) keeps the pool at the LRU cap; windowed resolution + // leases <= maxResident repos per window and releases them. Peak concurrent + // open Databases must stay within the resident cap (+ at most one transient + // overshoot at an init boundary — the pool's documented soft cap). + expect(openCounter.peak).toBeGreaterThan(0); + expect(openCounter.peak).toBeLessThanOrEqual(maxResident + 1); + }); +}); diff --git a/gitnexus/test/unit/group/sync.test.ts b/gitnexus/test/unit/group/sync.test.ts index 4fc320076..061f5cfc4 100644 --- a/gitnexus/test/unit/group/sync.test.ts +++ b/gitnexus/test/unit/group/sync.test.ts @@ -166,20 +166,19 @@ describe('syncGroup', () => { expect(result).toBeDefined(); }); - it('test_syncGroup_closes_only_opened_pools', async () => { + it('test_syncGroup_does_not_force_close_pools (release-not-close, #2191 review)', async () => { + // Post windowed-resolution refactor, syncGroup releases its eviction leases + // and lets the pool's LRU reclaim repos — it does NOT call closeLbug. This + // avoids tearing down a pool entry a concurrent MCP reader may share. const config = makeConfig({ 'app/backend': 'backend-repo', 'app/frontend': 'frontend-repo', }); - const closedIds: string[] = []; - const { vi } = await import('vitest'); const poolAdapter = await import('../../../src/core/lbug/pool-adapter.js'); const initSpy = vi.spyOn(poolAdapter, 'initLbug').mockResolvedValue(undefined); - const closeSpy = vi.spyOn(poolAdapter, 'closeLbug').mockImplementation(async (id?: string) => { - if (id) closedIds.push(id); - }); + const closeSpy = vi.spyOn(poolAdapter, 'closeLbug').mockResolvedValue(undefined); try { await syncGroup(config, { @@ -192,19 +191,8 @@ describe('syncGroup', () => { skipWrite: true, }).catch(() => {}); - // closeLbug must have been called at least once with specific pool ids - expect(closeSpy.mock.calls.length).toBeGreaterThan(0); - expect(closedIds).toContain('app-backend'); - expect(closedIds).toContain('app-frontend'); - - // Every call must have a truthy string id - for (const id of closedIds) { - expect(id).toBeTruthy(); - expect(typeof id).toBe('string'); - } - // No blanket close (no-arg or empty-string or undefined) - const blanketCalls = closeSpy.mock.calls.filter((args) => args.length === 0 || !args[0]); - expect(blanketCalls).toHaveLength(0); + // No closeLbug — repos are left evictable for the LRU to reclaim. + expect(closeSpy.mock.calls.length).toBe(0); } finally { initSpy.mockRestore(); closeSpy.mockRestore(); @@ -534,7 +522,9 @@ service OrderService { }, }); expect(initSpy).toHaveBeenCalledWith('billing-repo', path.join(storageDir, 'lbug')); - expect(closeSpy).toHaveBeenCalledWith('billing-repo'); + // syncGroup no longer force-closes pools (release-not-close, #2191 review); + // repos are left evictable for the LRU. Assert no teardown call here. + expect(closeSpy).not.toHaveBeenCalled(); } finally { initSpy.mockRestore(); closeSpy.mockRestore(); @@ -1044,18 +1034,20 @@ service OrderService { skipWrite: true, }); - // Manifest symbol resolution must run while pools are still open + // Manifest symbol resolution runs against live (leased) pools. expect(manifestResolvedWhilePoolOpen).toBe(true); - expect(closeLbugCalled).toBe(true); // The manifest cross-link must use the real UID from the DB, not synthetic + // — the #2189 fix, now via windowed resolution (the svc/orders↔svc/payments + // link forms one window whose repos are re-inited + leased for resolution). const manifestLinks = result.crossLinks.filter((cl) => cl.matchType === 'manifest'); expect(manifestLinks).toHaveLength(1); expect(manifestLinks[0].to.symbolUid).toBe('real-uid-checkout'); expect(manifestLinks[0].to.symbolUid).not.toContain('manifest::'); - // closeLbug must fire exactly twice (one per repo) - expect(closeSpy).toHaveBeenCalledTimes(2); + // syncGroup no longer force-closes pools (release-not-close, #2191 review). + expect(closeLbugCalled).toBe(false); + expect(closeSpy).not.toHaveBeenCalled(); } finally { initSpy.mockRestore(); closeSpy.mockRestore(); @@ -1105,6 +1097,198 @@ service OrderService { }); }); +// Lifecycle wiring for issue #2189: syncGroup must pin every repo it +// initializes (so a group larger than MAX_POOL_SIZE survives deferred +// manifest/workspace resolution) and release those pins on completion AND on +// error. The eviction-survival MECHANISM itself is proven against real +// evictLRU in test/unit/lbug-pool-pinning.test.ts; these tests prove the sync +// loop drives that mechanism correctly. (A full end-to-end proof through the +// real pool — real symbolUid instead of synthetic after >5 repos — would +// require a real or fully-native-mocked LadybugDB stack; mechanism + wiring +// coverage stands in for it here.) +describe('syncGroup windowed manifest resolution (issue #2189 / PR #2191 review)', () => { + const groupConfig = (count: number, links: GroupManifestLink[] = []): GroupConfig => { + const repos: Record = {}; + for (let i = 1; i <= count; i++) repos[`app/repo-${i}`] = `repo-${i}`; + return { + version: 1, + name: 'test', + description: '', + repos, + links, + packages: {}, + detect: { + http: true, + grpc: false, + thrift: false, + topics: false, + shared_libs: false, + embedding_fallback: false, + workspace_deps: false, + }, + matching: { bm25_threshold: 0.7, embedding_threshold: 0.65, max_candidates_per_step: 3 }, + }; + }; + + const okHandle = async (_name: string, groupPath: string): Promise => ({ + id: groupPath.replace(/\//g, '-'), + path: groupPath, + repoPath: '/tmp/' + groupPath, + storagePath: '/tmp/' + groupPath + '/.gitnexus', + }); + + const httpLink = (from: string, to: string): GroupManifestLink => ({ + from, + to, + type: 'http', + contract: 'GET::/api/x', + role: 'consumer', + }); + + // pinRepo now returns a release disposer; the spy returns a tracked spy fn so + // tests can assert every acquired lease was released. + const setupPoolSpies = async () => { + const poolAdapter = await import('../../../src/core/lbug/pool-adapter.js'); + const releaseSpies: Array> = []; + const initSpy = vi.spyOn(poolAdapter, 'initLbug').mockResolvedValue(undefined); + const execSpy = vi.spyOn(poolAdapter, 'executeParameterized').mockResolvedValue([]); + const pinSpy = vi.spyOn(poolAdapter, 'pinRepo').mockImplementation(() => { + const release = vi.fn(); + releaseSpies.push(release); + return release; + }); + const restore = () => { + initSpy.mockRestore(); + execSpy.mockRestore(); + pinSpy.mockRestore(); + }; + return { releaseSpies, initSpy, execSpy, pinSpy, restore }; + }; + + it('pins only the repos referenced by manifest links, not the whole group', async () => { + const { pinSpy, restore } = await setupPoolSpies(); + try { + await syncGroup(groupConfig(8, [httpLink('app/repo-1', 'app/repo-2')]), { + resolveRepoHandle: okHandle, + skipWrite: true, + }); + const pinnedIds = pinSpy.mock.calls.map((c) => c[0]).sort(); + // Only the windowed (link-referenced) repos are leased — bounded residency, + // not the whole 8-repo group. + expect(pinnedIds).toEqual(['app-repo-1', 'app-repo-2']); + expect(pinnedIds).not.toContain('app-repo-3'); + } finally { + restore(); + } + }); + + it('does not pin during the init loop when there are no manifest links', async () => { + const { pinSpy, restore } = await setupPoolSpies(); + try { + await syncGroup(groupConfig(8, []), { resolveRepoHandle: okHandle, skipWrite: true }); + // The init loop extracts contracts without pinning; with no links there + // are no resolution windows, so nothing is ever pinned. + expect(pinSpy.mock.calls.length).toBe(0); + } finally { + restore(); + } + }); + + it('releases every window lease on successful completion', async () => { + const { releaseSpies, restore } = await setupPoolSpies(); + try { + await syncGroup( + groupConfig(8, [ + httpLink('app/repo-1', 'app/repo-2'), + httpLink('app/repo-7', 'app/repo-8'), + ]), + { resolveRepoHandle: okHandle, skipWrite: true }, + ); + expect(releaseSpies.length).toBeGreaterThan(0); + for (const release of releaseSpies) expect(release).toHaveBeenCalled(); + } finally { + restore(); + } + }); + + it('releases the window leases even when resolution throws mid-window', async () => { + const { ManifestExtractor } = + await import('../../../src/core/group/extractors/manifest-extractor.js'); + const { releaseSpies, restore } = await setupPoolSpies(); + const manifestSpy = vi + .spyOn(ManifestExtractor.prototype, 'extractFromManifest') + .mockRejectedValue(new Error('resolution boom')); + try { + await expect( + syncGroup(groupConfig(8, [httpLink('app/repo-1', 'app/repo-2')]), { + resolveRepoHandle: okHandle, + skipWrite: true, + }), + ).rejects.toThrow('resolution boom'); + // The window's finally released its acquired leases despite the throw. + expect(releaseSpies.length).toBeGreaterThan(0); + for (const release of releaseSpies) expect(release).toHaveBeenCalled(); + } finally { + restore(); + manifestSpy.mockRestore(); + } + }); + + it('does not pin a repo that fails to resolve (no pool handle)', async () => { + const { pinSpy, restore } = await setupPoolSpies(); + try { + await syncGroup(groupConfig(3, [httpLink('app/repo-1', 'app/repo-2')]), { + resolveRepoHandle: async (_name, groupPath) => + groupPath === 'app/repo-2' ? null : okHandle(_name, groupPath), + skipWrite: true, + }); + const pinnedIds = pinSpy.mock.calls.map((c) => c[0]); + // repo-2 has no handle (resolve returned null) → not in knownRepos → + // never windowed, never leased; repo-1 (resolved) is. + expect(pinnedIds).toContain('app-repo-1'); + expect(pinnedIds).not.toContain('app-repo-2'); + } finally { + restore(); + } + }); + + it('releases an already-acquired lease when a later init in the same window throws', async () => { + const poolAdapter = await import('../../../src/core/lbug/pool-adapter.js'); + const releaseSpies: Array> = []; + // Throw on the SECOND init of app-repo-2 — the first is the init-loop + // extraction; the second is the window re-init. This isolates the failure + // to window setup, after app-repo-1's lease was already acquired. + const initCounts = new Map(); + const initSpy = vi.spyOn(poolAdapter, 'initLbug').mockImplementation(async (id: string) => { + const n = (initCounts.get(id) ?? 0) + 1; + initCounts.set(id, n); + if (id === 'app-repo-2' && n === 2) throw new Error('window init boom'); + }); + const execSpy = vi.spyOn(poolAdapter, 'executeParameterized').mockResolvedValue([]); + const pinSpy = vi.spyOn(poolAdapter, 'pinRepo').mockImplementation(() => { + const release = vi.fn(); + releaseSpies.push(release); + return release; + }); + try { + await expect( + syncGroup(groupConfig(2, [httpLink('app/repo-1', 'app/repo-2')]), { + resolveRepoHandle: okHandle, + skipWrite: true, + }), + ).rejects.toThrow('window init boom'); + // Exactly one lease was acquired (app-repo-1) before app-repo-2's init + // threw, and the window finally released it — no leaked lease. + expect(releaseSpies.length).toBe(1); + expect(releaseSpies[0]).toHaveBeenCalled(); + } finally { + initSpy.mockRestore(); + execSpy.mockRestore(); + pinSpy.mockRestore(); + } + }); +}); + describe('stableRepoPoolId', () => { it('returns lowercase name when no collision', () => { const entry: RegistryEntry = { diff --git a/gitnexus/test/unit/lbug-pool-pinning.test.ts b/gitnexus/test/unit/lbug-pool-pinning.test.ts new file mode 100644 index 000000000..914647b09 --- /dev/null +++ b/gitnexus/test/unit/lbug-pool-pinning.test.ts @@ -0,0 +1,223 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { mkdtempSync, writeFileSync, rmSync } from 'node:fs'; + +// Drive the REAL initLbug path (which calls evictLRU) rather than +// initLbugWithDb (which bypasses eviction entirely). The native LadybugDB +// open/connect/FTS stack is mocked exactly as in lbug-pool-fts-load.test.ts, +// plus sidecar-recovery so openReadOnlyDatabase's preflight is a no-op. fs is +// NOT mocked — each repo uses a real temp file so the fs.stat existence check +// in doInitLbug succeeds naturally. +// +// Covers issue #2189: a group sync larger than MAX_POOL_SIZE must keep every +// repo resident through deferred manifest/workspace resolution. Pinning makes +// that resident set survive automatic (LRU + idle) eviction. + +const { loadFTSExtensionMock } = vi.hoisted(() => ({ + loadFTSExtensionMock: vi.fn(), +})); + +vi.mock('@ladybugdb/core', () => ({ + default: { + Database: vi.fn(), + Connection: vi.fn(function (this: any) { + // probeDatabaseForShadowReplay() runs a probe query during the + // read-only open; the result must expose getAll()/close(). + this.query = vi.fn().mockResolvedValue({ + getAll: vi.fn().mockResolvedValue([]), + close: vi.fn(), + }); + this.close = vi.fn().mockResolvedValue(undefined); + }), + }, +})); + +vi.mock('../../src/core/lbug/lbug-adapter.js', () => ({ + isReadOnlyDbError: vi.fn(() => false), + loadFTSExtension: loadFTSExtensionMock, +})); + +vi.mock('../../src/core/lbug/lbug-config.js', () => ({ + // A fresh fake Database per call so distinct dbPaths get distinct entries + // and closeOne's db.close() resolves per repo. + createLbugDatabase: vi.fn(() => ({ + init: vi.fn().mockResolvedValue(undefined), + close: vi.fn().mockResolvedValue(undefined), + })), + toNativeSafePath: vi.fn((p: string) => p), + isWalCorruptionError: vi.fn(() => false), + WAL_RECOVERY_SUGGESTION: '', +})); + +vi.mock('../../src/core/lbug/sidecar-recovery.js', () => ({ + preflightLbugSidecars: vi.fn().mockResolvedValue(undefined), + isMissingFsError: vi.fn(() => false), + isMissingShadowSidecarError: vi.fn(() => false), + isReadOnlyShadowReplayError: vi.fn(() => false), + quarantineWalForMissingShadow: vi.fn().mockResolvedValue(''), + renameFailureMessage: vi.fn((p: string) => `rename failed for ${p}`), + statIfExists: vi.fn().mockResolvedValue(null), +})); + +const { initLbug, closeLbug, isLbugReady, pinRepo, unpinRepo } = + await import('../../src/core/lbug/pool-adapter.js'); + +describe('pool-adapter repo pinning (issue #2189)', () => { + let tmpDir: string; + // Track every repoId touched so afterEach can fully reset module-global state. + const touched = new Set(); + + const dbPathFor = (repoId: string): string => { + const p = path.join(tmpDir, `${repoId}.lbug`); + writeFileSync(p, ''); // real file so fs.stat() in doInitLbug succeeds + return p; + }; + + const init = async (repoId: string): Promise => { + touched.add(repoId); + await initLbug(repoId, dbPathFor(repoId)); + }; + + beforeEach(() => { + tmpDir = mkdtempSync(path.join(os.tmpdir(), 'gn-pin-test-')); + loadFTSExtensionMock.mockResolvedValue(true); + }); + + afterEach(async () => { + vi.useRealTimers(); + await closeLbug().catch(() => {}); + for (const id of touched) unpinRepo(id); + touched.clear(); + loadFTSExtensionMock.mockReset(); + rmSync(tmpDir, { recursive: true, force: true }); + }); + + // MAX_POOL_SIZE is 5; the 6th init triggers evictLRU. + it('characterization: WITHOUT pinning, the earliest repo is LRU-evicted past the cap', async () => { + for (let i = 1; i <= 6; i++) await init(`repo-${i}`); + + // repo-1 had the oldest lastUsed and nothing was checked out, so it is the + // eviction victim — exactly the stale-executor scenario from #2189. + expect(isLbugReady('repo-1')).toBe(false); + // The most recently initialized repo survives. + expect(isLbugReady('repo-6')).toBe(true); + }); + + it('FIX: pinning each repo keeps all of them resident past the cap', async () => { + for (let i = 1; i <= 6; i++) { + await init(`repo-${i}`); + pinRepo(`repo-${i}`); + } + + for (let i = 1; i <= 6; i++) { + expect(isLbugReady(`repo-${i}`)).toBe(true); + } + }); + + it('explicit close beats the pin AND clears it (no cross-operation leak)', async () => { + pinRepo('repo-x'); + await init('repo-x'); + expect(isLbugReady('repo-x')).toBe(true); + + // Explicit teardown closes a pinned repo without needing an unpin first. + await closeLbug('repo-x'); + expect(isLbugReady('repo-x')).toBe(false); + + // The pin must have been cleared on close: re-init repo-x FIRST (oldest + // lastUsed) and fill past the cap. If the pin had leaked, repo-x would be + // un-evictable; instead it is evicted as the LRU victim. + await init('repo-x'); + for (let i = 1; i <= 5; i++) await init(`fresh-${i}`); + expect(isLbugReady('repo-x')).toBe(false); + }); + + it('idle-timeout sweep skips pinned repos but still evicts idle unpinned ones', async () => { + vi.useFakeTimers(); + + await init('pinned-idle'); + pinRepo('pinned-idle'); + await init('unpinned-idle'); + + expect(isLbugReady('pinned-idle')).toBe(true); + expect(isLbugReady('unpinned-idle')).toBe(true); + + // The idle timer runs every 60s and closes entries idle past + // IDLE_TIMEOUT_MS (5 min) with no checked-out connections. advance past + // both thresholds, flushing microtasks between fires. + await vi.advanceTimersByTimeAsync(5 * 60 * 1000 + 60 * 1000); + + expect(isLbugReady('pinned-idle')).toBe(true); + expect(isLbugReady('unpinned-idle')).toBe(false); + }); + + it('unpinRepo re-enables eviction for that repo', async () => { + // Pin five repos and fill the pool; a sixth init evicts nothing (all pinned). + for (let i = 1; i <= 5; i++) { + await init(`p-${i}`); + pinRepo(`p-${i}`); + } + await init('p-6'); // unpinned; pool now holds 6 (soft-cap exceeded) + for (let i = 1; i <= 6; i++) expect(isLbugReady(`p-${i}`)).toBe(true); + + // Unpin the oldest, then init a 7th repo — the now-unpinned p-1 is the LRU + // victim. + unpinRepo('p-1'); + await init('p-7'); + expect(isLbugReady('p-1')).toBe(false); + expect(isLbugReady('p-7')).toBe(true); + }); + + it('reference-counts leases: two pins need two unpins before eviction (Finding 1)', async () => { + // Fill the pool to capacity, all leased. + for (let i = 1; i <= 4; i++) { + await init(`rc-${i}`); + pinRepo(`rc-${i}`); + } + await init('rc-shared'); + pinRepo('rc-shared'); // lease 1 + pinRepo('rc-shared'); // lease 2 (two holders) + + // Release ONE lease — a holder remains, so rc-shared stays exempt even + // under eviction pressure. + unpinRepo('rc-shared'); + await init('rc-extra'); // evictLRU finds no unpinned victim → pool grows + expect(isLbugReady('rc-shared')).toBe(true); + + // Release the LAST lease — now rc-shared (oldest unpinned) is evictable. + unpinRepo('rc-shared'); + await init('rc-extra2'); + expect(isLbugReady('rc-shared')).toBe(false); + }); + + it('unpinRepo floors at zero and tolerates unknown repoIds', () => { + expect(() => { + unpinRepo('never-touched'); // unknown repoId → no-op + pinRepo('floor-x'); + unpinRepo('floor-x'); // count 0 → key deleted + unpinRepo('floor-x'); // already gone → no-op, never a negative count + }).not.toThrow(); + }); + + it('pinRepo returns a disposer that releases exactly once and composes with refcount', async () => { + for (let i = 1; i <= 4; i++) { + await init(`d-${i}`); + pinRepo(`d-${i}`); + } + await init('d-shared'); + const release1 = pinRepo('d-shared'); // lease 1 + const release2 = pinRepo('d-shared'); // lease 2 + + release1(); + release1(); // double-call is a no-op — must NOT decrement lease 2 + + // lease 2 still held → d-shared survives eviction pressure. + await init('d-extra'); + expect(isLbugReady('d-shared')).toBe(true); + + // Release the last lease via its own disposer → now evictable. + release2(); + await init('d-extra2'); + expect(isLbugReady('d-shared')).toBe(false); + }); +});