diff --git a/.changeset/isolate-document-durability-state.md b/.changeset/isolate-document-durability-state.md new file mode 100644 index 00000000..8e251643 --- /dev/null +++ b/.changeset/isolate-document-durability-state.md @@ -0,0 +1,5 @@ +--- +'@inkeep/open-knowledge': patch +--- + +Open Knowledge servers now keep document durability coordination inside each server instance. Running multiple servers in one process with overlapping document names no longer lets one content root's reconciled base, persistence batch, in-flight flush, agent-write marker, or store failure/divergence state affect another server. Disk reconciliation, agent writes, persistence retries, and the staleness watchdog all consume the same instance-owned state, reducing the risk of incorrect merge anchors, skipped flushes, or misleading durability telemetry when desktop, tests, or embedded hosts create more than one server. diff --git a/packages/app/tests/integration/cc1-broadcast.test.ts b/packages/app/tests/integration/cc1-broadcast.test.ts index 00bc80b7..79054ce8 100644 --- a/packages/app/tests/integration/cc1-broadcast.test.ts +++ b/packages/app/tests/integration/cc1-broadcast.test.ts @@ -414,7 +414,12 @@ describe('CC1 broadcast — L1 integration', () => { const beforeTextLen = systemDoc?.getText('source').length ?? 0; expect(() => - applyExternalChange(server.instance.hocuspocus, SYSTEM_DOC_NAME, '# should be ignored'), + applyExternalChange( + server.instance.durabilityState, + server.instance.hocuspocus, + SYSTEM_DOC_NAME, + '# should be ignored', + ), ).not.toThrow(); const afterXmlLen = systemDoc?.getXmlFragment('default').length ?? 0; diff --git a/packages/app/tests/integration/external-change-stale-anchor-interleave.test.ts b/packages/app/tests/integration/external-change-stale-anchor-interleave.test.ts index c39adc41..e3a45967 100644 --- a/packages/app/tests/integration/external-change-stale-anchor-interleave.test.ts +++ b/packages/app/tests/integration/external-change-stale-anchor-interleave.test.ts @@ -92,7 +92,12 @@ async function runInterleave(clientId: number): Promise { // External wholesale replace lands on the server and reaches the live // client only. writeFileSync(join(server.contentDir, `${docName}.md`), REPLACED_CONTENT, 'utf-8'); - applyExternalChange(server.instance.hocuspocus, docName, REPLACED_CONTENT); + applyExternalChange( + server.instance.durabilityState, + server.instance.hocuspocus, + docName, + REPLACED_CONTENT, + ); await pollUntil(() => live.ytext.toString().includes('M6-'), 5000); // Concurrent stale-anchored insert: the paused client still sees the diff --git a/packages/app/tests/integration/persistence-fan-out.test.ts b/packages/app/tests/integration/persistence-fan-out.test.ts index e1d1efed..15e62af0 100644 --- a/packages/app/tests/integration/persistence-fan-out.test.ts +++ b/packages/app/tests/integration/persistence-fan-out.test.ts @@ -150,7 +150,12 @@ describe('persistence L2 fan-out integration (US-014, FR-7)', () => { }); // Simulate a file-watcher external change — registers file-system contributor - applyExternalChange(server.hocuspocus, 'fs-writer-doc', '# Updated from disk\n'); + applyExternalChange( + server.durabilityState, + server.hocuspocus, + 'fs-writer-doc', + '# Updated from disk\n', + ); const doc = server.hocuspocus.documents.get('fs-writer-doc'); doc?.removeDirectConnection(); @@ -195,7 +200,12 @@ describe('persistence L2 fan-out integration (US-014, FR-7)', () => { recordContributor('concurrent-doc', 'agent-s1', 'Session 1', 'agent-s1'); // File-watcher contributor (simulating an external disk change) - applyExternalChange(server.hocuspocus, 'concurrent-doc', '# Updated concurrently\n'); + applyExternalChange( + server.durabilityState, + server.hocuspocus, + 'concurrent-doc', + '# Updated concurrently\n', + ); const doc = server.hocuspocus.documents.get('concurrent-doc'); doc?.removeDirectConnection(); diff --git a/packages/app/tests/stress/bridge-convergence.fuzz.test.ts b/packages/app/tests/stress/bridge-convergence.fuzz.test.ts index e6707b0b..9edec2ea 100644 --- a/packages/app/tests/stress/bridge-convergence.fuzz.test.ts +++ b/packages/app/tests/stress/bridge-convergence.fuzz.test.ts @@ -427,7 +427,12 @@ async function applyOp( // doc. writeFileSync(join(server.contentDir, `${docName}.md`), op.newContent, 'utf-8'); try { - applyExternalChange(server.instance.hocuspocus, docName, op.newContent); + applyExternalChange( + server.instance.durabilityState, + server.instance.hocuspocus, + docName, + op.newContent, + ); } catch { // Non-fatal — no-op early-returns when the doc is unloaded; the // bridge-merge transact race re-throw is survivable for the fuzz. diff --git a/packages/server/src/api-agent-frontmatter.test.ts b/packages/server/src/api-agent-frontmatter.test.ts index 9cb70a67..4adbc41c 100644 --- a/packages/server/src/api-agent-frontmatter.test.ts +++ b/packages/server/src/api-agent-frontmatter.test.ts @@ -26,7 +26,7 @@ import { applyAgentMarkdownWrite, applyAgentUndo, } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; interface CapturedResponse { status: number; diff --git a/packages/server/src/api-agent-patch-ytext-truth.test.ts b/packages/server/src/api-agent-patch-ytext-truth.test.ts index 7775f559..fd25bad6 100644 --- a/packages/server/src/api-agent-patch-ytext-truth.test.ts +++ b/packages/server/src/api-agent-patch-ytext-truth.test.ts @@ -28,7 +28,7 @@ import { AgentSessionManager, applyAgentMarkdownWrite, } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { getMetrics, resetMetrics } from './metrics.ts'; interface CapturedResponse { diff --git a/packages/server/src/api-agent-patch.test.ts b/packages/server/src/api-agent-patch.test.ts index dd66f50d..eb0af839 100644 --- a/packages/server/src/api-agent-patch.test.ts +++ b/packages/server/src/api-agent-patch.test.ts @@ -11,7 +11,7 @@ import { AgentSessionManager, applyAgentMarkdownWrite, } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; interface CapturedResponse { status: number; diff --git a/packages/server/src/api-agent-write-file-index.test.ts b/packages/server/src/api-agent-write-file-index.test.ts index a6afd338..957ff12a 100644 --- a/packages/server/src/api-agent-write-file-index.test.ts +++ b/packages/server/src/api-agent-write-file-index.test.ts @@ -21,7 +21,7 @@ import { AgentSessionManager, applyAgentMarkdownWrite, } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { createContentFilter } from './content-filter.ts'; import type { DiskEvent, FileIndexEntry } from './file-watcher.ts'; diff --git a/packages/server/src/api-agent-write-summary.test.ts b/packages/server/src/api-agent-write-summary.test.ts index 3e6e40ba..00a449b8 100644 --- a/packages/server/src/api-agent-write-summary.test.ts +++ b/packages/server/src/api-agent-write-summary.test.ts @@ -6,7 +6,7 @@ import { Readable } from 'node:stream'; import { Hocuspocus } from '@hocuspocus/server'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; import { AgentSessionManager, applyAgentMarkdownWrite } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { __formatContributorsForTests as formatContributorsForTest, __resetContributorsForTests as resetContributorsForTest, diff --git a/packages/server/src/api-asset.test.ts b/packages/server/src/api-asset.test.ts index e8989312..4bd3446e 100644 --- a/packages/server/src/api-asset.test.ts +++ b/packages/server/src/api-asset.test.ts @@ -4,7 +4,7 @@ import { createServer, type Server } from 'node:http'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import type { ContentFilter } from './content-filter.ts'; import { listenOnLoopback } from './loopback-rig-test-helpers.ts'; diff --git a/packages/server/src/api-backlinks.test.ts b/packages/server/src/api-backlinks.test.ts index 3ecd1af4..6041270b 100644 --- a/packages/server/src/api-backlinks.test.ts +++ b/packages/server/src/api-backlinks.test.ts @@ -4,7 +4,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Readable } from 'node:stream'; import { describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { type ContentFilter, createContentFilter } from './content-filter.ts'; import type { FileIndexEntry } from './file-watcher.ts'; diff --git a/packages/server/src/api-config.test.ts b/packages/server/src/api-config.test.ts index a07b52ae..a873b916 100644 --- a/packages/server/src/api-config.test.ts +++ b/packages/server/src/api-config.test.ts @@ -4,7 +4,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Readable } from 'node:stream'; import { describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { getLocalDir } from './config/paths.ts'; import { acquireServerLock } from './server-lock.ts'; diff --git a/packages/server/src/api-create-page.test.ts b/packages/server/src/api-create-page.test.ts index 88ab2567..7999222f 100644 --- a/packages/server/src/api-create-page.test.ts +++ b/packages/server/src/api-create-page.test.ts @@ -13,7 +13,7 @@ import { join } from 'node:path'; import { Readable } from 'node:stream'; import type { Principal } from '@inkeep/open-knowledge-core'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { contributorCount, hasContributor, swapContributors } from './contributor-tracker.ts'; import type { FileIndexEntry } from './file-watcher.ts'; diff --git a/packages/server/src/api-docname-validation.test.ts b/packages/server/src/api-docname-validation.test.ts index 0afdac31..175bfd3b 100644 --- a/packages/server/src/api-docname-validation.test.ts +++ b/packages/server/src/api-docname-validation.test.ts @@ -15,7 +15,7 @@ import { Readable } from 'node:stream'; import { Hocuspocus } from '@hocuspocus/server'; import { describe, expect, test } from 'vitest'; import { AgentSessionManager } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; interface CapturedResponse { status: number; diff --git a/packages/server/src/api-document-list-ready-gate.test.ts b/packages/server/src/api-document-list-ready-gate.test.ts index 301c5fb6..e1707d2e 100644 --- a/packages/server/src/api-document-list-ready-gate.test.ts +++ b/packages/server/src/api-document-list-ready-gate.test.ts @@ -23,7 +23,7 @@ import { join } from 'node:path'; import { Readable } from 'node:stream'; import { Hocuspocus } from '@hocuspocus/server'; import { describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import type { FileIndexEntry } from './file-watcher.ts'; function makeReq(url: string): IncomingMessage { diff --git a/packages/server/src/api-empty-docname.test.ts b/packages/server/src/api-empty-docname.test.ts index 7c89c3eb..95bf56f4 100644 --- a/packages/server/src/api-empty-docname.test.ts +++ b/packages/server/src/api-empty-docname.test.ts @@ -17,7 +17,7 @@ import { Readable } from 'node:stream'; import { Hocuspocus } from '@hocuspocus/server'; import { describe, expect, test } from 'vitest'; import { AgentSessionManager } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; interface CapturedResponse { status: number; diff --git a/packages/server/src/api-extension.ok-init.test.ts b/packages/server/src/api-extension.ok-init.test.ts index ae3daf27..afc1b4cd 100644 --- a/packages/server/src/api-extension.ok-init.test.ts +++ b/packages/server/src/api-extension.ok-init.test.ts @@ -74,7 +74,7 @@ async function bootRig(): Promise { const { Hocuspocus } = await import('@hocuspocus/server'); const { AgentSessionManager } = await import('./agent-sessions.ts'); - const { createApiExtension } = await import('./api-extension.ts'); + const { createApiExtension } = await import('./api-extension.test-helper.ts'); const hocuspocus = new Hocuspocus({ quiet: true }); const sessionManager = new AgentSessionManager(hocuspocus); diff --git a/packages/server/src/api-extension.test-helper.ts b/packages/server/src/api-extension.test-helper.ts new file mode 100644 index 00000000..451b4b99 --- /dev/null +++ b/packages/server/src/api-extension.test-helper.ts @@ -0,0 +1,14 @@ +import type { ApiExtensionOptions } from './api-extension.ts'; +import { createApiExtension as createApiExtensionBase } from './api-extension.ts'; +import { DocumentDurabilityState } from './document-durability-state.ts'; + +export * from './api-extension.ts'; + +export function createApiExtension( + options: Omit, +): ReturnType { + return createApiExtensionBase({ + ...options, + durabilityState: new DocumentDurabilityState(), + }); +} diff --git a/packages/server/src/api-extension.test.ts b/packages/server/src/api-extension.test.ts index 2c3687a6..ac954315 100644 --- a/packages/server/src/api-extension.test.ts +++ b/packages/server/src/api-extension.test.ts @@ -3,7 +3,11 @@ import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join, resolve } from 'node:path'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { resumeSyncOnAuthEvent, safeSubdir, sanitizeFilename } from './api-extension.ts'; +import { + resumeSyncOnAuthEvent, + safeSubdir, + sanitizeFilename, +} from './api-extension.test-helper.ts'; import type { AuthEvent } from './local-ops/types.ts'; import { listenOnLoopback } from './loopback-rig-test-helpers.ts'; import type { SyncEngine } from './sync-engine.ts'; @@ -186,7 +190,7 @@ describe('handleUploadAsset', () => { const { Hocuspocus } = await import('@hocuspocus/server'); const { AgentSessionManager } = await import('./agent-sessions.ts'); - const { createApiExtension } = await import('./api-extension.ts'); + const { createApiExtension } = await import('./api-extension.test-helper.ts'); const hocuspocus = new Hocuspocus({ quiet: true }); const sessionManager = new AgentSessionManager(hocuspocus); @@ -538,7 +542,7 @@ describe('handleUploadAsset — same-dir sha256 dedup (FR-2)', () => { const { Hocuspocus } = await import('@hocuspocus/server'); const { AgentSessionManager } = await import('./agent-sessions.ts'); - const { createApiExtension } = await import('./api-extension.ts'); + const { createApiExtension } = await import('./api-extension.test-helper.ts'); const hocuspocus = new Hocuspocus({ quiet: true }); const sessionManager = new AgentSessionManager(hocuspocus); diff --git a/packages/server/src/api-extension.ts b/packages/server/src/api-extension.ts index 666b973c..b2186376 100644 --- a/packages/server/src/api-extension.ts +++ b/packages/server/src/api-extension.ts @@ -478,6 +478,7 @@ import { isConfigDoc, isSystemDoc } from './cc1-broadcast.ts'; import { withHiddenWindowsConsole } from './child-process-windows-hide.ts'; import type { ResolveStrategy } from './conflict-storage.ts'; import type { ContentFilter } from './content-filter.ts'; +import { isWithinContentDir, safeContentPath } from './content-path.ts'; import { docNameToRelativePath, forgetDocExtension, @@ -488,6 +489,7 @@ import { SUPPORTED_DOC_EXTENSIONS, stripDocExtension, } from './doc-extensions.ts'; +import type { DocumentDurabilityState, StoreFailure } from './document-durability-state.ts'; import { type ReconcileBeforeWriteResult, reconcileDiskBeforeAgentWrite, @@ -584,14 +586,6 @@ import { } from './metrics.ts'; import { precomputeParse } from './parse-pool.ts'; import { isWithinDir, toPosix } from './path-utils.ts'; -import { - deleteReconciledBase, - getActiveBranch, - isWithinContentDir, - type StoreFailure, - safeContentPath, - setReconciledBase, -} from './persistence.ts'; import { appendRenameLogEntry, createAncestorShaSetCache, @@ -2607,6 +2601,7 @@ async function renameTrackedPathInGit( export interface ApiExtensionOptions { hocuspocus: Hocuspocus; + durabilityState: DocumentDurabilityState; sessionManager: AgentSessionManager; contentDir: string; /** @@ -2703,28 +2698,6 @@ export interface ApiExtensionOptions { * is NOT yet ported from origin/main. */ flushContributors?: () => Promise; - /** - * Read-and-clear the last disk-store failure for a docName. Wired to - * `persistence.takeStoreFailure`. A write handler force-flushes the store - * then calls this to report disk truth instead of a false success when the - * persistence step threw (ENOSPC / EACCES / EROFS, etc.). - */ - takeStoreFailure?: (docName: string) => StoreFailure | null; - /** - * Read-and-clear whether a docName's most recent agent-triggered store was - * reverted by the L3 disk-divergence backstop. Wired to - * `persistence.takeStoreDivergence`. A write handler force-flushes the store - * then calls this to return `urn:ok:error:disk-divergence` instead of a false - * success when disk diverged and the overwrite was aborted (disk won). - */ - takeStoreDivergence?: (docName: string) => boolean; - /** - * Mark a docName's next store as agent-write-triggered (L3 - * gate). Wired to `persistence.markAgentWriteStore`. `flushDiskAndDetectOutcome` - * calls it immediately before force-flushing, so only agent-handler-forced - * stores can disk-wins-revert on divergence (human-editor stores are excluded). - */ - markAgentWriteStore?: (docName: string) => void; /** Accessor for the current branch from the HEAD watcher. Returns null when unknown. */ getCurrentBranch?: () => string | null; /** @@ -2991,6 +2964,7 @@ function applyDiskEventToLiveAllFilesIndex( } export function createApiExtension(options: ApiExtensionOptions): Extension { + const { durabilityState } = options; const { hocuspocus, sessionManager, @@ -3017,9 +2991,6 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { shadowRef, flushGitCommit, flushContributors, - takeStoreFailure, - takeStoreDivergence, - markAgentWriteStore, getCurrentBranch, getDiskAckSVs, contentRoot, @@ -3419,12 +3390,12 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { // fires (Hocuspocus passes a null transaction origin for agent // DirectConnection writes, so the origin can't gate it). `storeDocumentNow` // read-and-clears the marker. - markAgentWriteStore?.(docName); + durabilityState.markAgentWriteStore(docName); await hocuspocus.debouncer.executeNow(debounceId); } - const failure = takeStoreFailure?.(docName) ?? null; + const failure = durabilityState.takeStoreFailure(docName); if (failure) return { kind: 'failure', failure }; - if (takeStoreDivergence?.(docName)) return { kind: 'divergence' }; + if (durabilityState.takeStoreDivergence(docName)) return { kind: 'divergence' }; return null; } @@ -3731,7 +3702,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { for (const docName of docNames) { const document = hocuspocus.documents.get(docName); - deleteReconciledBase(docName); + durabilityState.deleteReconciledBase(docName); // Forget the managed-artifact LKG too (no-op for ordinary docs). The LKG // is the verbatim bytes last persisted; leaving it set lets an identical- // content re-create after a delete be classed a no-op and never re-land on @@ -3810,7 +3781,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { tracedMkdirSync(dirname(filePath), { recursive: true }); writeFileIfContentDiffers(filePath, markdown); registerWrite(filePath, contentHash(markdown)); - setReconciledBase(docName, markdown); + durabilityState.setReconciledBase(docName, markdown); mutateFileIndex?.({ kind: 'update', path: filePath, docName, content: markdown }); } @@ -4247,7 +4218,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { span.setAttribute('rename.rewrite_candidates', pendingRewrites.length); assertRewriteTargetsNotConflicted(pendingRewrites.map((entry) => entry.docName)); - reconcileDiskBeforeAgentWrite(hocuspocus, sourceDocName, contentDir); + reconcileDiskBeforeAgentWrite(durabilityState, hocuspocus, sourceDocName, contentDir); if (recentlyRemovedDocs && !isSystemDoc(sourceDocName) && !isConfigDoc(sourceDocName)) { recentlyRemovedDocs.setDeleted(sourceDocName); } @@ -4508,7 +4479,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { // losslessly through the rename re-serialize and re-resolves on the // next normal load/reconcile. The extension-level resolveEmbed is // also shadowed by this function's own options param. - reconcileDiskBeforeAgentWrite(hocuspocus, docName, contentDir); + reconcileDiskBeforeAgentWrite(durabilityState, hocuspocus, docName, contentDir); const content = readCurrentDocumentContent(docName); if (typeof content === 'string') { snapshotContents.set(docName, content); @@ -4808,7 +4779,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { [{ fromDocName, toDocName }], new Map([[fromDocName, renamedSource.markdown]]), ); - setReconciledBase(toDocName, renamedSource.markdown); + durabilityState.setReconciledBase(toDocName, renamedSource.markdown); mutateFileIndex?.({ kind: 'rename', @@ -5178,6 +5149,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { // content handlers. Separate FILE_WATCHER_ORIGIN transact before the // agent's session.origin transact below. const agentWriteReconcile = reconcileDiskBeforeAgentWrite( + durabilityState, hocuspocus, docName, contentDir, @@ -5372,6 +5344,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { // can't clobber it. Runs its own FILE_WATCHER_ORIGIN transact BEFORE // the agent's session.origin transact below — never nested. const writeMdReconcile = reconcileDiskBeforeAgentWrite( + durabilityState, hocuspocus, resolvedDocName, contentDir, @@ -5819,6 +5792,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { }); const reconcile = reconcileDiskBeforeAgentWrite( + durabilityState, hocuspocus, resolvedDocName, contentDir, @@ -6061,6 +6035,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { // live (disk-reflecting) frontmatter, not a stale loaded copy. Separate // FILE_WATCHER_ORIGIN transact BEFORE the agent's session.origin transact. const fmReconcile = reconcileDiskBeforeAgentWrite( + durabilityState, hocuspocus, resolvedDocName, contentDir, @@ -7500,6 +7475,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { // result). Separate FILE_WATCHER_ORIGIN transact BEFORE the agent's // session.origin transact below. const patchReconcile = reconcileDiskBeforeAgentWrite( + durabilityState, hocuspocus, docName, contentDir, @@ -9208,7 +9184,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { * Gated on `ready` for the same reason `handleDocumentList` is: the * boot-time `switchReconciledBaseScope(startupBranch)` lives inside * `initAsync` (server-factory.ts), and a renderer that fetches before - * it runs would observe the module-level `'main'` default instead of + * it runs would observe this server's initial `'main'` default instead of * the actual HEAD branch. The renderer's `current-branch-store` is * fire-once and only updates from CC1 `branch-switched`, so a stale * cold-start fetch sticks until a real cross-branch checkout. @@ -9238,7 +9214,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { try { // Park until `initAsync` has called `switchReconciledBaseScope` with // the resolved HEAD branch. Without this gate, a renderer that fetches - // during the boot window reads the persistence module's `'main'` + // during the boot window reads this server's initial `'main'` // default and caches it in `current-branch-store` for the lifetime of // the session. Mirrors the `handleDocumentList` gate; `.catch()` keeps // the handler responsive on a degraded boot. @@ -9250,7 +9226,7 @@ export function createApiExtension(options: ApiExtensionOptions): Extension { ); }); } - const currentBranch = getActiveBranch(); + const currentBranch = durabilityState.getActiveBranch(); // `getDiskAckSVs` is wired by standalone boot; plugin mode (dev // server) doesn't have a CC1Broadcaster and omits the field. The // schema's `.optional()` keeps the response shape valid in both diff --git a/packages/server/src/api-file-ops.test.ts b/packages/server/src/api-file-ops.test.ts index 17e8d8e3..c658c4d6 100644 --- a/packages/server/src/api-file-ops.test.ts +++ b/packages/server/src/api-file-ops.test.ts @@ -17,7 +17,7 @@ import { Hocuspocus } from '@hocuspocus/server'; import simpleGit from 'simple-git'; import { afterEach, describe, expect, test } from 'vitest'; import type * as Y from 'yjs'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { type ContentFilter, createContentFilter } from './content-filter.ts'; import { _resetDocExtensionsForTests, getDocExtension } from './doc-extensions.ts'; diff --git a/packages/server/src/api-folder-delete-duplicate-disk-enum.test.ts b/packages/server/src/api-folder-delete-duplicate-disk-enum.test.ts index 76319399..7b703a2a 100644 --- a/packages/server/src/api-folder-delete-duplicate-disk-enum.test.ts +++ b/packages/server/src/api-folder-delete-duplicate-disk-enum.test.ts @@ -25,7 +25,7 @@ import { tmpdir } from 'node:os'; import { dirname, join } from 'node:path'; import { Readable } from 'node:stream'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { _resetDocExtensionsForTests } from './doc-extensions.ts'; import { RecentlyRemovedDocs } from './recently-removed-docs.ts'; diff --git a/packages/server/src/api-folder-rename-disk-enum.test.ts b/packages/server/src/api-folder-rename-disk-enum.test.ts index f5d6679d..08b283e1 100644 --- a/packages/server/src/api-folder-rename-disk-enum.test.ts +++ b/packages/server/src/api-folder-rename-disk-enum.test.ts @@ -19,7 +19,7 @@ import { tmpdir } from 'node:os'; import { dirname, join } from 'node:path'; import { Readable } from 'node:stream'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { _resetDocExtensionsForTests } from './doc-extensions.ts'; diff --git a/packages/server/src/api-link-preview.test.ts b/packages/server/src/api-link-preview.test.ts index ed64a54f..8c92f19e 100644 --- a/packages/server/src/api-link-preview.test.ts +++ b/packages/server/src/api-link-preview.test.ts @@ -4,7 +4,7 @@ import { createServer, type Server } from 'node:http'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, type Mock, test, vi } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import type { GuardRejectReason } from './link-preview/guarded-fetch.ts'; import type { GuardedFetch } from './link-preview/metadata.ts'; import { PinoLogger } from './logger.ts'; diff --git a/packages/server/src/api-ok-reserved-guard.test.ts b/packages/server/src/api-ok-reserved-guard.test.ts index 91c75cda..9645a79a 100644 --- a/packages/server/src/api-ok-reserved-guard.test.ts +++ b/packages/server/src/api-ok-reserved-guard.test.ts @@ -4,7 +4,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Readable } from 'node:stream'; import { afterEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { _resetDocExtensionsForTests } from './doc-extensions.ts'; diff --git a/packages/server/src/api-pages.test.ts b/packages/server/src/api-pages.test.ts index 81b51775..ece6a43f 100644 --- a/packages/server/src/api-pages.test.ts +++ b/packages/server/src/api-pages.test.ts @@ -10,7 +10,11 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Readable } from 'node:stream'; import { describe, expect, test } from 'vitest'; -import { createApiExtension, extractHeadings, extractPageTitle } from './api-extension.ts'; +import { + createApiExtension, + extractHeadings, + extractPageTitle, +} from './api-extension.test-helper.ts'; import type { FileIndexEntry } from './file-watcher.ts'; describe('extractPageTitle', () => { diff --git a/packages/server/src/api-rename-crash-injection.test.ts b/packages/server/src/api-rename-crash-injection.test.ts index 4f7bc520..48c178e1 100644 --- a/packages/server/src/api-rename-crash-injection.test.ts +++ b/packages/server/src/api-rename-crash-injection.test.ts @@ -34,7 +34,7 @@ import { join, resolve } from 'node:path'; import { Readable } from 'node:stream'; import simpleGit from 'simple-git'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { swapContributors } from './contributor-tracker.ts'; import type { FileIndexEntry } from './file-watcher.ts'; diff --git a/packages/server/src/api-rename-log-emission.test.ts b/packages/server/src/api-rename-log-emission.test.ts index bb92289f..196c47dd 100644 --- a/packages/server/src/api-rename-log-emission.test.ts +++ b/packages/server/src/api-rename-log-emission.test.ts @@ -23,7 +23,7 @@ import { join, resolve } from 'node:path'; import { Readable } from 'node:stream'; import simpleGit from 'simple-git'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { recordContributor, swapContributors } from './contributor-tracker.ts'; import { _resetDocExtensionsForTests } from './doc-extensions.ts'; diff --git a/packages/server/src/api-rename-race-conditions.test.ts b/packages/server/src/api-rename-race-conditions.test.ts index b9873f81..98eecbcd 100644 --- a/packages/server/src/api-rename-race-conditions.test.ts +++ b/packages/server/src/api-rename-race-conditions.test.ts @@ -21,7 +21,7 @@ import { tmpdir } from 'node:os'; import { join, resolve } from 'node:path'; import { Readable } from 'node:stream'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { swapContributors } from './contributor-tracker.ts'; import { _resetDocExtensionsForTests } from './doc-extensions.ts'; diff --git a/packages/server/src/api-rename-rollback-summary.test.ts b/packages/server/src/api-rename-rollback-summary.test.ts index 95d59bc0..527fe4c8 100644 --- a/packages/server/src/api-rename-rollback-summary.test.ts +++ b/packages/server/src/api-rename-rollback-summary.test.ts @@ -24,7 +24,7 @@ import { Readable } from 'node:stream'; import { setImmediate } from 'node:timers/promises'; import type { Principal } from '@inkeep/open-knowledge-core'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { __formatContributorsForTests as formatContributorsForTest, diff --git a/packages/server/src/api-rename-telemetry.test.ts b/packages/server/src/api-rename-telemetry.test.ts index 149c2483..92d50df4 100644 --- a/packages/server/src/api-rename-telemetry.test.ts +++ b/packages/server/src/api-rename-telemetry.test.ts @@ -22,7 +22,10 @@ import { PeriodicExportingMetricReader, } from '@opentelemetry/sdk-metrics'; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, test } from 'vitest'; -import { __resetRenameTelemetryForTesting, createApiExtension } from './api-extension.ts'; +import { + __resetRenameTelemetryForTesting, + createApiExtension, +} from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { swapContributors } from './contributor-tracker.ts'; import type { FileIndexEntry } from './file-watcher.ts'; diff --git a/packages/server/src/api-rollback-actor-identity.test.ts b/packages/server/src/api-rollback-actor-identity.test.ts index f81f040e..c7767a2f 100644 --- a/packages/server/src/api-rollback-actor-identity.test.ts +++ b/packages/server/src/api-rollback-actor-identity.test.ts @@ -7,7 +7,7 @@ import type { Principal } from '@inkeep/open-knowledge-core'; import simpleGit from 'simple-git'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; import * as Y from 'yjs'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { type ContributorEntry, diff --git a/packages/server/src/api-rollback-rename-history.test.ts b/packages/server/src/api-rollback-rename-history.test.ts index 7f1ce72e..954310be 100644 --- a/packages/server/src/api-rollback-rename-history.test.ts +++ b/packages/server/src/api-rollback-rename-history.test.ts @@ -6,7 +6,7 @@ import { Readable } from 'node:stream'; import simpleGit from 'simple-git'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; import * as Y from 'yjs'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import { swapContributors } from './contributor-tracker.ts'; import { diff --git a/packages/server/src/api-search-semantic.test.ts b/packages/server/src/api-search-semantic.test.ts index 294f7f2b..da716aed 100644 --- a/packages/server/src/api-search-semantic.test.ts +++ b/packages/server/src/api-search-semantic.test.ts @@ -13,7 +13,7 @@ import { join } from 'node:path'; import { Readable } from 'node:stream'; import { createWorkspaceSearchDocument } from '@inkeep/open-knowledge-core'; import { describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { createConceptEmbedder, type Embedder, SemanticSearchService } from './embeddings/index.ts'; import type { FileIndexEntry } from './file-watcher.ts'; diff --git a/packages/server/src/api-search.test.ts b/packages/server/src/api-search.test.ts index 677477ab..ab73efaa 100644 --- a/packages/server/src/api-search.test.ts +++ b/packages/server/src/api-search.test.ts @@ -8,7 +8,7 @@ import { createApiExtension, DEFAULT_SEARCH_MAX_ENTRIES, getSearchMaxEntries, -} from './api-extension.ts'; +} from './api-extension.test-helper.ts'; import type { FileIndexEntry } from './file-watcher.ts'; interface CapturedResponse { diff --git a/packages/server/src/api-suggest-links.test.ts b/packages/server/src/api-suggest-links.test.ts index bf5c32a2..a79ad783 100644 --- a/packages/server/src/api-suggest-links.test.ts +++ b/packages/server/src/api-suggest-links.test.ts @@ -5,7 +5,7 @@ import { join } from 'node:path'; import { Readable } from 'node:stream'; import type { Hocuspocus } from '@hocuspocus/server'; import { describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import type { FileIndexEntry } from './file-watcher.ts'; interface CapturedResponse { diff --git a/packages/server/src/api-tags.test.ts b/packages/server/src/api-tags.test.ts index fa178c2c..b6d938e8 100644 --- a/packages/server/src/api-tags.test.ts +++ b/packages/server/src/api-tags.test.ts @@ -4,7 +4,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Readable } from 'node:stream'; import { describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import type { FileIndexEntry } from './file-watcher.ts'; import { TagIndex } from './tag-index.ts'; diff --git a/packages/server/src/api-template-get.test.ts b/packages/server/src/api-template-get.test.ts index bc469243..e3932709 100644 --- a/packages/server/src/api-template-get.test.ts +++ b/packages/server/src/api-template-get.test.ts @@ -15,7 +15,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Readable } from 'node:stream'; import { describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; function makeReq(url: string): IncomingMessage { const readable = Readable.from(Buffer.from('')) as unknown as IncomingMessage; diff --git a/packages/server/src/api-test-rescan-files.test.ts b/packages/server/src/api-test-rescan-files.test.ts index f24861e5..6273d464 100644 --- a/packages/server/src/api-test-rescan-files.test.ts +++ b/packages/server/src/api-test-rescan-files.test.ts @@ -22,7 +22,7 @@ import { tmpdir } from 'node:os'; import { join } from 'node:path'; import { Readable } from 'node:stream'; import { describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import type { FileIndexEntry } from './file-watcher.ts'; interface CapturedResponse { diff --git a/packages/server/src/api-trash-cleanup.test.ts b/packages/server/src/api-trash-cleanup.test.ts index cf12eea1..dbefce80 100644 --- a/packages/server/src/api-trash-cleanup.test.ts +++ b/packages/server/src/api-trash-cleanup.test.ts @@ -31,7 +31,7 @@ import { join, resolve } from 'node:path'; import { Readable } from 'node:stream'; import type { Principal } from '@inkeep/open-knowledge-core'; import { afterEach, describe, expect, test } from 'vitest'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { BacklinkIndex } from './backlink-index.ts'; import type { FileIndexEntry, FolderIndexEntry } from './file-watcher.ts'; import { RecentlyRemovedDocs } from './recently-removed-docs.ts'; diff --git a/packages/server/src/asset-references.ts b/packages/server/src/asset-references.ts index 82152a42..a9a1fae0 100644 --- a/packages/server/src/asset-references.ts +++ b/packages/server/src/asset-references.ts @@ -7,10 +7,10 @@ import { mediaKindForSidebarAssetExtension, resolveAssetProjectPath, } from '@inkeep/open-knowledge-core'; +import { isWithinContentDir } from './content-path.ts'; import type { FileIndexEntry } from './file-watcher.ts'; import { matchMarkdownLinks, matchWikiLinks } from './link-syntax.ts'; import { getLogger } from './logger.ts'; -import { isWithinContentDir } from './persistence.ts'; const log = getLogger('asset-references'); diff --git a/packages/server/src/content-path.ts b/packages/server/src/content-path.ts new file mode 100644 index 00000000..f4b79f74 --- /dev/null +++ b/packages/server/src/content-path.ts @@ -0,0 +1,19 @@ +import { resolve } from 'node:path'; +import { docNameToRelativePath } from './doc-extensions.ts'; +import { isWithinDir } from './path-utils.ts'; + +export function safeContentPath(documentName: string, contentDir: string): string { + if (documentName.includes('\x00')) { + throw new Error(`Invalid document name: ${documentName}`); + } + const relativePath = docNameToRelativePath(documentName); + const filePath = resolve(contentDir, relativePath); + if (!isWithinDir(filePath, contentDir)) { + throw new Error(`Invalid document name: ${documentName}`); + } + return filePath; +} + +export function isWithinContentDir(p: string, contentDir: string): boolean { + return isWithinDir(p, contentDir); +} diff --git a/packages/server/src/disk-content-intake.ts b/packages/server/src/disk-content-intake.ts new file mode 100644 index 00000000..98622ee5 --- /dev/null +++ b/packages/server/src/disk-content-intake.ts @@ -0,0 +1,30 @@ +import type * as Y from 'yjs'; +import { composeAndWriteRawBody } from './bridge-intake.ts'; +import type { PairedWriteOrigin } from './server-observers.ts'; + +/** + * Identity-stable transaction origin for file-watcher disk-to-CRDT writes. + * The paired marker opts into the bridge observers' paired-write fast path; + * skipStoreHooks prevents persistence from re-saving bytes just read from disk. + */ +export const FILE_WATCHER_ORIGIN = { + source: 'local', + skipStoreHooks: true, + context: { origin: 'file-watcher', paired: true }, +} as const satisfies PairedWriteOrigin; + +/** + * Apply raw disk bytes through the shared paired-write primitive. The caller + * owns the outer `document.transact(..., FILE_WATCHER_ORIGIN)` boundary. + */ +export function applyDiskContentToDoc( + document: Y.Doc, + content: string, + resolveEmbed?: (basename: string, sourcePath: string) => string | null, + sourcePath?: string, + resolveSize?: (basename: string, sourcePath: string) => number | null, +): void { + const embedResolver = + resolveEmbed && sourcePath ? { resolveEmbed, resolveSize, sourcePath } : undefined; + composeAndWriteRawBody(document, content, 'file-watcher', embedResolver); +} diff --git a/packages/server/src/document-durability-state.test.ts b/packages/server/src/document-durability-state.test.ts new file mode 100644 index 00000000..23d758e6 --- /dev/null +++ b/packages/server/src/document-durability-state.test.ts @@ -0,0 +1,92 @@ +import { describe, expect, test } from 'vitest'; +import { DocumentDurabilityState } from './document-durability-state.ts'; + +describe('DocumentDurabilityState', () => { + test('starts with an empty main scope and no transient coordination state', () => { + const state = new DocumentDurabilityState(); + + expect(state.getActiveBranch()).toBe('main'); + expect(state.getReconciledBase('doc')).toBeUndefined(); + expect(state.peekInFlightFlush('doc')).toBeUndefined(); + expect(state.isBatchInProgress()).toBe(false); + expect(state.consumeAgentWriteStore('doc')).toBe(false); + expect(state.takeStoreFailure('doc')).toBeNull(); + expect(state.takeStoreDivergence('doc')).toBe(false); + }); + + test('retains reconciled bases independently for each visited branch', () => { + const state = new DocumentDurabilityState(); + state.setReconciledBase('doc', 'main bytes'); + state.switchReconciledBaseScope('feature'); + state.setReconciledBase('doc', 'feature bytes'); + + expect(state.getReconciledBase('doc')).toBe('feature bytes'); + state.switchReconciledBaseScope('main'); + expect(state.getReconciledBase('doc')).toBe('main bytes'); + }); + + test('deletes a reconciled base only from the active branch', () => { + const state = new DocumentDurabilityState(); + state.setReconciledBase('doc', 'main bytes'); + state.switchReconciledBaseScope('feature'); + state.setReconciledBase('doc', 'feature bytes'); + + state.switchReconciledBaseScope('main'); + state.deleteReconciledBase('doc'); + expect(state.getReconciledBase('doc')).toBeUndefined(); + state.switchReconciledBaseScope('feature'); + expect(state.getReconciledBase('doc')).toBe('feature bytes'); + }); + + test('isolates every owned coordination channel between instances', () => { + const first = new DocumentDurabilityState(); + const second = new DocumentDurabilityState(); + first.setReconciledBase('doc', 'first'); + first.setBatchInProgress(true); + first.beginInFlightFlush('doc', 'first flush'); + first.markAgentWriteStore('doc'); + first.recordStoreFailure('doc', { code: 'ENOSPC', message: 'full' }); + first.recordStoreDivergence('doc'); + + expect(second.getReconciledBase('doc')).toBeUndefined(); + expect(second.isBatchInProgress()).toBe(false); + expect(second.peekInFlightFlush('doc')).toBeUndefined(); + expect(second.consumeAgentWriteStore('doc')).toBe(false); + expect(second.takeStoreFailure('doc')).toBeNull(); + expect(second.takeStoreDivergence('doc')).toBe(false); + }); + + test('does not let an older flush clear a newer in-flight snapshot', () => { + const state = new DocumentDurabilityState(); + state.beginInFlightFlush('doc', 'older'); + state.beginInFlightFlush('doc', 'newer'); + state.finishInFlightFlush('doc', 'older'); + expect(state.peekInFlightFlush('doc')).toBe('newer'); + state.finishInFlightFlush('doc', 'newer'); + expect(state.peekInFlightFlush('doc')).toBeUndefined(); + }); + + test('consumes agent markers, failures, and divergences once', () => { + const state = new DocumentDurabilityState(); + state.markAgentWriteStore('doc'); + state.recordStoreFailure('doc', { message: 'write failed' }); + state.recordStoreDivergence('doc'); + + expect(state.consumeAgentWriteStore('doc')).toBe(true); + expect(state.consumeAgentWriteStore('doc')).toBe(false); + expect(state.takeStoreFailure('doc')).toEqual({ message: 'write failed' }); + expect(state.takeStoreFailure('doc')).toBeNull(); + expect(state.takeStoreDivergence('doc')).toBe(true); + expect(state.takeStoreDivergence('doc')).toBe(false); + }); + + test('clears a store failure without consuming another document failure', () => { + const state = new DocumentDurabilityState(); + state.recordStoreFailure('cleared', { code: 'ENOSPC', message: 'full' }); + state.recordStoreFailure('retained', { message: 'readonly' }); + + state.clearStoreFailure('cleared'); + expect(state.takeStoreFailure('cleared')).toBeNull(); + expect(state.takeStoreFailure('retained')).toEqual({ message: 'readonly' }); + }); +}); diff --git a/packages/server/src/document-durability-state.ts b/packages/server/src/document-durability-state.ts new file mode 100644 index 00000000..38a0925e --- /dev/null +++ b/packages/server/src/document-durability-state.ts @@ -0,0 +1,109 @@ +/** + * A disk-store failure captured out-of-band so an agent write handler can + * report disk truth after Hocuspocus swallows a rejected store hook. + */ +export interface StoreFailure { + code?: string; + message: string; +} + +export class DocumentDurabilityState { + /** Last known-good markdown for each document, retained independently by branch. */ + private readonly reconciledBaseByBranch = new Map>(); + /** + * Normalized snapshots for disk flushes that have started but not settled. + * These let reconciliation distinguish this server's own just-written bytes + * from a foreign edit during the rename-to-base-advance window. + */ + private readonly inFlightFlushByDoc = new Map(); + private readonly agentWriteStores = new Set(); + private readonly storeFailures = new Map(); + private readonly storeDivergences = new Set(); + private activeBranch: string; + private batchInProgress = false; + + constructor(initialBranch = 'main') { + this.activeBranch = initialBranch; + this.reconciledBaseByBranch.set(initialBranch, new Map()); + } + + switchReconciledBaseScope(branch: string): void { + this.activeBranch = branch; + if (!this.reconciledBaseByBranch.has(branch)) { + this.reconciledBaseByBranch.set(branch, new Map()); + } + } + + getActiveBranch(): string { + return this.activeBranch; + } + + getReconciledBase(docName: string): string | undefined { + return this.reconciledBaseByBranch.get(this.activeBranch)?.get(docName); + } + + setReconciledBase(docName: string, content: string): void { + let bases = this.reconciledBaseByBranch.get(this.activeBranch); + if (!bases) { + bases = new Map(); + this.reconciledBaseByBranch.set(this.activeBranch, bases); + } + bases.set(docName, content); + } + + deleteReconciledBase(docName: string): void { + this.reconciledBaseByBranch.get(this.activeBranch)?.delete(docName); + } + + beginInFlightFlush(docName: string, normalizedMarkdown: string): void { + this.inFlightFlushByDoc.set(docName, normalizedMarkdown); + } + + peekInFlightFlush(docName: string): string | undefined { + return this.inFlightFlushByDoc.get(docName); + } + + finishInFlightFlush(docName: string, expectedNormalizedMarkdown: string): void { + if (this.inFlightFlushByDoc.get(docName) === expectedNormalizedMarkdown) { + this.inFlightFlushByDoc.delete(docName); + } + } + + setBatchInProgress(value: boolean): void { + this.batchInProgress = value; + } + + isBatchInProgress(): boolean { + return this.batchInProgress; + } + + markAgentWriteStore(docName: string): void { + this.agentWriteStores.add(docName); + } + + consumeAgentWriteStore(docName: string): boolean { + return this.agentWriteStores.delete(docName); + } + + recordStoreFailure(docName: string, failure: StoreFailure): void { + this.storeFailures.set(docName, failure); + } + + clearStoreFailure(docName: string): void { + this.storeFailures.delete(docName); + } + + takeStoreFailure(docName: string): StoreFailure | null { + const failure = this.storeFailures.get(docName) ?? null; + this.storeFailures.delete(docName); + return failure; + } + + recordStoreDivergence(docName: string): void { + this.storeDivergences.add(docName); + } + + takeStoreDivergence(docName: string): boolean { + return this.storeDivergences.delete(docName); + } +} diff --git a/packages/server/src/embed-probe.test.ts b/packages/server/src/embed-probe.test.ts index 4d88a9cb..16b95326 100644 --- a/packages/server/src/embed-probe.test.ts +++ b/packages/server/src/embed-probe.test.ts @@ -12,7 +12,7 @@ import { Readable } from 'node:stream'; import { Hocuspocus } from '@hocuspocus/server'; import { describe, expect, test } from 'vitest'; import { AgentSessionManager } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { deriveDetection, EMBED_PROBE_CAPACITY, diff --git a/packages/server/src/external-change.test.ts b/packages/server/src/external-change.test.ts index 004c099d..b6eee3f4 100644 --- a/packages/server/src/external-change.test.ts +++ b/packages/server/src/external-change.test.ts @@ -18,9 +18,9 @@ import { } from '@inkeep/open-knowledge-core'; import { beforeEach, describe, expect, test, vi } from 'vitest'; import type * as Y from 'yjs'; +import { DocumentDurabilityState } from './document-durability-state.ts'; import { applyExternalChange, createExternalChangeHandler } from './external-change.ts'; import { getLogger } from './logger.ts'; -import { getReconciledBase, setReconciledBase } from './persistence.ts'; type Conn = Awaited>; @@ -32,14 +32,16 @@ function getDoc(conn: Conn): Y.Doc { describe('applyExternalChange — throwing helper', () => { let hp: Hocuspocus; + let durabilityState: DocumentDurabilityState; beforeEach(() => { hp = new Hocuspocus({ quiet: true }); + durabilityState = new DocumentDurabilityState(); }); test('(a) document-missing early return — no throw, no mutations', () => { expect(() => { - applyExternalChange(hp, 'nonexistent-doc', '# Hello\n\nWorld\n'); + applyExternalChange(durabilityState, hp, 'nonexistent-doc', '# Hello\n\nWorld\n'); }).not.toThrow(); expect(hp.documents.get('nonexistent-doc')).toBeUndefined(); }); @@ -51,7 +53,7 @@ describe('applyExternalChange — throwing helper', () => { const fullContent = '---\ntitle: Test\ntags: [a, b]\n---\n# Hello\n\nParagraph text.\n'; - applyExternalChange(hp, docName, fullContent); + applyExternalChange(durabilityState, hp, docName, fullContent); // Y.Text contains the FULL content (FM region IS the FM source of truth). const ytext = doc.getText('source'); @@ -77,7 +79,7 @@ describe('applyExternalChange — throwing helper', () => { const doc = getDoc(conn); const content = '---\ntitle: Stable\nstatus: draft\n---\n# Body\n'; - applyExternalChange(hp, docName, content); + applyExternalChange(durabilityState, hp, docName, content); let textMutations = 0; const ytext = doc.getText('source'); @@ -86,7 +88,7 @@ describe('applyExternalChange — throwing helper', () => { }; ytext.observe(observer); - applyExternalChange(hp, docName, content); + applyExternalChange(durabilityState, hp, docName, content); ytext.unobserve(observer); expect(textMutations).toBe(0); @@ -100,7 +102,7 @@ describe('applyExternalChange — throwing helper', () => { const doc = getDoc(conn); const malformed = '---\ntitle: [unterminated\nstatus: published\n---\n# Body\n'; - applyExternalChange(hp, docName, malformed); + applyExternalChange(durabilityState, hp, docName, malformed); // Y.Text holds the malformed bytes verbatim — no defensive revert. expect(doc.getText('source').toString()).toBe(malformed); @@ -123,7 +125,7 @@ describe('applyExternalChange — throwing helper', () => { const doc = getDoc(conn); const onDisk = '---\ntags:\n - characters\n - air-nomads\n---\n\n# Aang\n'; - applyExternalChange(hp, docName, onDisk); + applyExternalChange(durabilityState, hp, docName, onDisk); const ytext = doc.getText('source').toString(); // FM region byte-identical to disk (preserves indent + scalar style). @@ -148,7 +150,7 @@ describe('applyExternalChange — throwing helper', () => { const conn = await hp.openDirectConnection(docName); const doc = getDoc(conn); - applyExternalChange(hp, docName, '---\n'); + applyExternalChange(durabilityState, hp, docName, '---\n'); const ytext = doc.getText('source').toString(); // Under contract: ytext holds the raw disk bytes — `---\n` survives. @@ -164,7 +166,7 @@ describe('applyExternalChange — throwing helper', () => { const content = '# Hello\n\nWorld\n'; - applyExternalChange(hp, docName, content); + applyExternalChange(durabilityState, hp, docName, content); expect(doc.getText('source').toString()).toBe(content); let textMutations = 0; @@ -174,7 +176,7 @@ describe('applyExternalChange — throwing helper', () => { }; ytext.observe(observer); - applyExternalChange(hp, docName, content); + applyExternalChange(durabilityState, hp, docName, content); ytext.unobserve(observer); expect(textMutations).toBe(0); @@ -199,7 +201,7 @@ describe('applyExternalChange — throwing helper', () => { } }); - applyExternalChange(hp, docName, '# Test\n'); + applyExternalChange(durabilityState, hp, docName, '# Test\n'); expect(capturedOrigin).toEqual({ source: 'local', @@ -232,10 +234,10 @@ describe('applyExternalChange — throwing helper', () => { const doc = getDoc(conn); // Establish initial state via a successful apply. - applyExternalChange(hp, docName, '# Original\n'); - setReconciledBase(docName, '# Original\n'); + applyExternalChange(durabilityState, hp, docName, '# Original\n'); + durabilityState.setReconciledBase(docName, '# Original\n'); expect(doc.getText('source').toString()).toBe('# Original\n'); - expect(getReconciledBase(docName)).toBe('# Original\n'); + expect(durabilityState.getReconciledBase(docName)).toBe('# Original\n'); // Wrap doc.transact so it runs the inner mutations normally, then // throws after — simulating "applyFastDiff succeeded, updateYFragment @@ -250,7 +252,7 @@ describe('applyExternalChange — throwing helper', () => { }) as typeof doc.transact; expect(() => { - applyExternalChange(hp, docName, '# After-Mutation\n'); + applyExternalChange(durabilityState, hp, docName, '# After-Mutation\n'); }).toThrow(/synthetic/); // ytext WAS mutated by applyFastDiff before the synthetic throw. @@ -259,7 +261,7 @@ describe('applyExternalChange — throwing helper', () => { // the fix, reconciledBase would still equal '# Original\n' (the post- // transact setReconciledBase('# After-Mutation\n') is skipped on // throw), and persistence would see ytext ≠ reconciledBase and write. - expect(getReconciledBase(docName)).toBe('# After-Mutation\n'); + expect(durabilityState.getReconciledBase(docName)).toBe('# After-Mutation\n'); doc.transact = originalTransact as typeof doc.transact; await conn.disconnect(); @@ -268,16 +270,18 @@ describe('applyExternalChange — throwing helper', () => { describe('createExternalChangeHandler — error-swallowing factory', () => { let hp: Hocuspocus; + let durabilityState: DocumentDurabilityState; beforeEach(() => { hp = new Hocuspocus({ quiet: true }); + durabilityState = new DocumentDurabilityState(); }); test('factory wrapper catches and logs when applyExternalChange throws', async () => { const errorSpy = vi.spyOn(getLogger('file-watcher'), 'error'); try { - const handler = createExternalChangeHandler(hp); + const handler = createExternalChangeHandler(durabilityState, hp); const docName = 'test-throw-path'; const conn = await hp.openDirectConnection(docName); @@ -314,7 +318,7 @@ describe('createExternalChangeHandler — error-swallowing factory', () => { const errorSpy = vi.spyOn(getLogger('file-watcher'), 'error'); try { - const handler = createExternalChangeHandler(hp); + const handler = createExternalChangeHandler(durabilityState, hp); const docName = 'test-bridge-violation-rethrow'; const conn = await hp.openDirectConnection(docName); @@ -359,7 +363,7 @@ describe('createExternalChangeHandler — error-swallowing factory', () => { console.error = errorSpy; try { - const handler = createExternalChangeHandler(hp); + const handler = createExternalChangeHandler(durabilityState, hp); const docName = 'test-merge-loss-rethrow'; const conn = await hp.openDirectConnection(docName); diff --git a/packages/server/src/external-change.ts b/packages/server/src/external-change.ts index a567a091..87ba406f 100644 --- a/packages/server/src/external-change.ts +++ b/packages/server/src/external-change.ts @@ -16,11 +16,12 @@ import { stripFrontmatter, } from '@inkeep/open-knowledge-core'; import { formatReconcileSubject } from '@inkeep/open-knowledge-core/shadow-repo-layout'; -import type * as Y from 'yjs'; -import { composeAndWriteRawBody } from './bridge-intake.ts'; import { isConfigDoc, isMermaidDoc, isSystemDoc } from './cc1-broadcast.ts'; import { isDocInConflict } from './conflict-errors.ts'; +import { isWithinContentDir, safeContentPath } from './content-path.ts'; import { recordContributor } from './contributor-tracker.ts'; +import { applyDiskContentToDoc, FILE_WATCHER_ORIGIN } from './disk-content-intake.ts'; +import type { DocumentDurabilityState } from './document-durability-state.ts'; import { recordFrontmatterEditSurface } from './frontmatter-telemetry.ts'; import { getLogger } from './logger.ts'; import { @@ -28,74 +29,10 @@ import { incrementReconcileInFlightFallthroughs, incrementReconcileOwnFlushSkips, } from './metrics.ts'; -import { - getReconciledBase, - isWithinContentDir, - peekInFlightFlush, - safeContentPath, - setReconciledBase, -} from './persistence.ts'; import { reconcile } from './reconciliation.ts'; -import type { PairedWriteOrigin } from './server-observers.ts'; import { FILE_SYSTEM_WRITER } from './shadow-repo.ts'; -/** - * Transaction origin for file-watcher disk→CRDT bridge operations. - * - * Exported so the bridge-invariant watcher can include it in its - * enforcing-origins Set by identity (not by string literal). Y.js transaction - * matching uses `Set.has(tx.origin)` which is identity-based for objects; - * a string literal `'file-watcher'` would never match this object. - * - * skipStoreHooks: true — prevents persistence from re-saving a file we just - * loaded from disk (feedback loop prevention). - * - * paired: true — `applyExternalChange` atomically writes BOTH XmlFragment and - * Y.Text inside one `doc.transact(..., FILE_WATCHER_ORIGIN)` block. Server - * Observer A/B match via `context.paired === true` and short-circuit - * symmetrically. - */ -export const FILE_WATCHER_ORIGIN = { - source: 'local', - skipStoreHooks: true, - context: { origin: 'file-watcher', paired: true }, -} as const satisfies PairedWriteOrigin; - -/** - * Apply file content to a live Y.Doc through the shared - * `composeAndWriteRawBody` primitive. Pure CRDT update — no contributor - * recording, no reconciledBase advance, no `Hocuspocus` lookup. - * - * Y.Text-is-truth contract: disk bytes land in Y.Text - * verbatim. The fragment derives from `parse(body)` via the primitive. - * No canonicalize-write-back step — markdown forms (e.g. doc-start `---` - * thematic breaks) survive in Y.Text in the user's source bytes; - * the bridge invariant tolerates any difference between raw bytes and - * `serialize(fragment)` via `normalizeBridge`'s equivalence classes. - * - * Used both by the file-watcher path (`applyExternalChange`, which adds - * contributor + reconciledBase side effects) and by the persistence - * tripwire reset path (which must NOT advance attribution or the - * reconciled base because no disk write happened). - * - * Atomicity boundary: caller MUST wrap this in - * `document.transact(..., FILE_WATCHER_ORIGIN)` so paired-write origin - * identity (precedent #24) reaches the observer guards. Calling transact - * inside the function would either lose origin identity (nested transacts - * pick the outer's) or fragment the atomicity contract for callers that - * already wrap. - */ -export function applyDiskContentToDoc( - document: Y.Doc, - content: string, - resolveEmbed?: (basename: string, sourcePath: string) => string | null, - sourcePath?: string, - resolveSize?: (basename: string, sourcePath: string) => number | null, -): void { - const embedResolver = - resolveEmbed && sourcePath ? { resolveEmbed, resolveSize, sourcePath } : undefined; - composeAndWriteRawBody(document, content, 'file-watcher', embedResolver); -} +export { FILE_WATCHER_ORIGIN } from './disk-content-intake.ts'; /** * Apply external file content to a live Y.Doc — the throwing core of the @@ -127,6 +64,7 @@ export function applyDiskContentToDoc( * so dev/test surfaces regressions loudly. */ export function applyExternalChange( + durabilityState: DocumentDurabilityState, hocuspocus: Hocuspocus, docName: string, content: string, @@ -180,7 +118,7 @@ export function applyExternalChange( // B settlement, or user mutation. Re-throwing preserves the existing // outer error-handling contract (`createExternalChangeHandler` // increments the error counter and logs). - setReconciledBase(docName, document.getText('source').toString()); + durabilityState.setReconciledBase(docName, document.getText('source').toString()); throw err; } @@ -211,7 +149,7 @@ export function applyExternalChange( // Set the reconciled base so persistence does not re-serialize and re-write // the same content on next flush. - setReconciledBase(docName, content); + durabilityState.setReconciledBase(docName, content); } /** @@ -227,13 +165,14 @@ export function applyExternalChange( * here would silently subvert the test-mode loud-failure gates. */ export function createExternalChangeHandler( + durabilityState: DocumentDurabilityState, hocuspocus: Hocuspocus, resolveEmbed?: (basename: string, sourcePath: string) => string | null, resolveSize?: (basename: string, sourcePath: string) => number | null, ): (docName: string, content: string) => Promise { return async (docName: string, content: string): Promise => { try { - applyExternalChange(hocuspocus, docName, content, resolveEmbed, resolveSize); + applyExternalChange(durabilityState, hocuspocus, docName, content, resolveEmbed, resolveSize); getLogger('file-watcher').info({ docName }, 'applied external change'); } catch (err) { if ( @@ -359,6 +298,7 @@ export function serializeYDocSource(document: { } export function reconcileDiskBeforeAgentWrite( + durabilityState: DocumentDurabilityState, hocuspocus: Hocuspocus, docName: string, contentDir: string, @@ -374,7 +314,7 @@ export function reconcileDiskBeforeAgentWrite( const document = hocuspocus.documents.get(docName); if (document && isDocInConflict(document)) return NOT_RECONCILED; - const base = getReconciledBase(docName); + const base = durabilityState.getReconciledBase(docName); if (base === undefined) return NOT_RECONCILED; let canonical: string; @@ -424,7 +364,7 @@ export function reconcileDiskBeforeAgentWrite( // flush continuation is the single owner of the base advance and runs // moments later. A foreign byte sequence landing inside the window does not // match and still falls through to the three-way merge below. - const inFlightFlush = peekInFlightFlush(docName); + const inFlightFlush = durabilityState.peekInFlightFlush(docName); if (inFlightFlush !== undefined) { if (normalizeBridge(diskContent) === inFlightFlush) { incrementReconcileOwnFlushSkips(); @@ -503,13 +443,13 @@ export function reconcileDiskBeforeAgentWrite( // FILE_WATCHER_ORIGIN transact and advances reconciledBase // synchronously. const ingest = outcome.kind === 'clean' ? diskContent : outcome.newContent; - applyExternalChange(hocuspocus, docName, ingest, resolveEmbed); + applyExternalChange(durabilityState, hocuspocus, docName, ingest, resolveEmbed); if (outcome.kind === 'merged') { // The base must track the DISK bytes, not the memory-only merged // content: the L3 store backstop and the watcher's queued event for // this same disk write both compare disk against the base, and the // caller's forced flush is what lands the merged content on disk. - setReconciledBase(docName, diskContent); + durabilityState.setReconciledBase(docName, diskContent); } // UTF-8 byte counts (Buffer.byteLength), matching the sibling // ContentDivergenceWarning's `*Bytes` semantics on the shared diff --git a/packages/server/src/file-watcher.test.ts b/packages/server/src/file-watcher.test.ts index ab42fd15..d81ad47f 100644 --- a/packages/server/src/file-watcher.test.ts +++ b/packages/server/src/file-watcher.test.ts @@ -442,7 +442,7 @@ describe('startWatcher file index', () => { const handle = await startWatcher(contentDir, async () => {}); try { const { getDocExtension } = await import('./doc-extensions.ts'); - const { safeContentPath } = await import('./persistence.ts'); + const { safeContentPath } = await import('./content-path.ts'); expect(getDocExtension('Upper')).toBe('.MD'); expect(getDocExtension('Mixed')).toBe('.MdX'); diff --git a/packages/server/src/file-watcher.ts b/packages/server/src/file-watcher.ts index 55c6f920..575c048c 100644 --- a/packages/server/src/file-watcher.ts +++ b/packages/server/src/file-watcher.ts @@ -20,6 +20,7 @@ import { dirname, extname, join, relative } from 'node:path'; import { LINKABLE_ASSET_EXTENSIONS } from '@inkeep/open-knowledge-core'; import { isConfigDoc, isReservedForUserTree, isSystemDoc } from './cc1-broadcast.ts'; import type { ContentFilter } from './content-filter.ts'; +import { isWithinContentDir } from './content-path.ts'; import { forgetDocExtension, isSupportedAssetFile, @@ -32,7 +33,6 @@ import { errnoCode } from './http/handler-utils.ts'; import { getLogger } from './logger.ts'; import { extractPageIcon, extractPageTitle } from './page-identity.ts'; import { toPosix } from './path-utils.ts'; -import { isWithinContentDir } from './persistence.ts'; import { containsConflictMarkers } from './reconciliation.ts'; import { getMeter, withSpan } from './telemetry.ts'; diff --git a/packages/server/src/git-preflight-boot.test.ts b/packages/server/src/git-preflight-boot.test.ts index d46aa6c6..100f908c 100644 --- a/packages/server/src/git-preflight-boot.test.ts +++ b/packages/server/src/git-preflight-boot.test.ts @@ -354,6 +354,9 @@ describe('bootServer git-preflight', () => { }, }); expect(booted.port).toBeGreaterThan(0); + // This test covers preflight gating, not shutdown during initialization. + // Let async initialization settle before exercising the teardown path. + await booted.ready; } finally { process.stderr.write = originalStderrWrite; if (booted) await booted.destroy(); diff --git a/packages/server/src/handoff-api.test.ts b/packages/server/src/handoff-api.test.ts index 4c20af88..0ee81030 100644 --- a/packages/server/src/handoff-api.test.ts +++ b/packages/server/src/handoff-api.test.ts @@ -466,7 +466,7 @@ describe('GET /api/installed-agents (integration — real HTTP + real createApiE const { Hocuspocus } = await import('@hocuspocus/server'); const { AgentSessionManager } = await import('./agent-sessions.ts'); - const { createApiExtension } = await import('./api-extension.ts'); + const { createApiExtension } = await import('./api-extension.test-helper.ts'); const hocuspocus = new Hocuspocus({ quiet: true }); const sessionManager = new AgentSessionManager(hocuspocus); diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index dd9ae049..66faa534 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -105,6 +105,7 @@ export { createContentFilterAsync, type RebuildResult as ContentFilterRebuildResult, } from './content-filter.ts'; +export { safeContentPath } from './content-path.ts'; export { // Back-compat public export; new code should use swapContributors(). // oxlint-disable-next-line typescript/no-deprecated @@ -122,6 +123,8 @@ export { type DetectClaudeDesktopOptions, detectClaudeDesktopPresence, } from './detect-claude-desktop.ts'; +export { FILE_WATCHER_ORIGIN } from './disk-content-intake.ts'; +export { DocumentDurabilityState, type StoreFailure } from './document-durability-state.ts'; export { clearEmbeddingsKeyFromAllBackends, createEmbeddingsSecretStore, @@ -139,7 +142,6 @@ export { export { applyExternalChange, createExternalChangeHandler, - FILE_WATCHER_ORIGIN, } from './external-change.ts'; export { createFileLogger, flushFileLogger, getLogFilePath, getLogsDir } from './file-logger.ts'; export { @@ -345,7 +347,6 @@ export { createPersistenceExtension, type PersistenceHandle, type PersistenceOptions, - safeContentPath, } from './persistence.ts'; export { loadPrincipal } from './principal.ts'; export { isProcessAlive, isValidLockPid } from './process-alive.ts'; diff --git a/packages/server/src/lint/audit.ts b/packages/server/src/lint/audit.ts index bd1e5c25..17430e11 100644 --- a/packages/server/src/lint/audit.ts +++ b/packages/server/src/lint/audit.ts @@ -17,7 +17,7 @@ import { } from '@inkeep/open-knowledge-core'; import { SymlinkEscapeError } from '../apply-managed-rename.ts'; import { createContentFilter } from '../content-filter.ts'; -import { isWithinContentDir } from '../persistence.ts'; +import { isWithinContentDir } from '../content-path.ts'; import { resolveEffectiveLinterConfig } from './resolve-config.ts'; export interface FileLintResult { diff --git a/packages/server/src/live-derived-index.test.ts b/packages/server/src/live-derived-index.test.ts index 8f1b719b..07891fc2 100644 --- a/packages/server/src/live-derived-index.test.ts +++ b/packages/server/src/live-derived-index.test.ts @@ -2,6 +2,7 @@ import { setTimeout as wait } from 'node:timers/promises'; import { Hocuspocus } from '@hocuspocus/server'; import { beforeEach, describe, expect, test, vi } from 'vitest'; import type * as Y from 'yjs'; +import { DocumentDurabilityState } from './document-durability-state.ts'; import { applyExternalChange } from './external-change.ts'; import { createLiveDerivedIndexExtension } from './live-derived-index.ts'; import { getLogger } from './logger.ts'; @@ -37,9 +38,11 @@ function makeOnChangePayload( describe('createLiveDerivedIndexExtension', () => { let hp: Hocuspocus; + let durabilityState: DocumentDurabilityState; beforeEach(() => { hp = new Hocuspocus({ quiet: true }); + durabilityState = new DocumentDurabilityState(); }); test('skips file-watcher origin transactions', async () => { @@ -53,7 +56,7 @@ describe('createLiveDerivedIndexExtension', () => { const conn = await hp.openDirectConnection('skip-file-watcher'); const doc = getDoc(conn); - applyExternalChange(hp, 'skip-file-watcher', '# Hello\n\n[[beta]]\n'); + applyExternalChange(durabilityState, hp, 'skip-file-watcher', '# Hello\n\n[[beta]]\n'); await extension.onChange?.( makeOnChangePayload(hp, doc, 'skip-file-watcher', { source: 'local', @@ -78,7 +81,12 @@ describe('createLiveDerivedIndexExtension', () => { const conn = await hp.openDirectConnection('debounced-doc'); const doc = getDoc(conn); - applyExternalChange(hp, 'debounced-doc', '---\ntitle: Debounced\n---\n# Hello\n\n[[beta]]\n'); + applyExternalChange( + durabilityState, + hp, + 'debounced-doc', + '---\ntitle: Debounced\n---\n# Hello\n\n[[beta]]\n', + ); const payload = makeOnChangePayload(hp, doc, 'debounced-doc', { source: 'local', context: { origin: 'agent-write' }, @@ -112,7 +120,7 @@ describe('createLiveDerivedIndexExtension', () => { const conn = await hp.openDirectConnection('tag-derived-doc'); const doc = getDoc(conn); - applyExternalChange(hp, 'tag-derived-doc', '# Hello\n\nA #typescript note.\n'); + applyExternalChange(durabilityState, hp, 'tag-derived-doc', '# Hello\n\nA #typescript note.\n'); const payload = makeOnChangePayload(hp, doc, 'tag-derived-doc', { source: 'local', context: { origin: 'agent-write' }, @@ -137,7 +145,7 @@ describe('createLiveDerivedIndexExtension', () => { const conn = await hp.openDirectConnection('unload-doc'); const doc = getDoc(conn); - applyExternalChange(hp, 'unload-doc', '# Hello\n'); + applyExternalChange(durabilityState, hp, 'unload-doc', '# Hello\n'); await extension.onChange?.( makeOnChangePayload(hp, doc, 'unload-doc', { source: 'local', @@ -166,8 +174,8 @@ describe('createLiveDerivedIndexExtension', () => { const firstDoc = getDoc(first); const secondDoc = getDoc(second); - applyExternalChange(hp, 'destroy-a', '# A\n'); - applyExternalChange(hp, 'destroy-b', '# B\n'); + applyExternalChange(durabilityState, hp, 'destroy-a', '# A\n'); + applyExternalChange(durabilityState, hp, 'destroy-b', '# B\n'); await extension.onChange?.( makeOnChangePayload(hp, firstDoc, 'destroy-a', { source: 'local', @@ -210,7 +218,7 @@ describe('createLiveDerivedIndexExtension', () => { // lands raw bytes in both ytext and fragment. Under contract, ytext keeps // the CRLF; fragment is derived via parse(body) so its serialize output // would emit LF. - applyExternalChange(hp, 'crlf-doc', '# Title\r\n\r\nLine A\r\nLine B\r\n'); + applyExternalChange(durabilityState, hp, 'crlf-doc', '# Title\r\n\r\nLine A\r\nLine B\r\n'); await extension.onChange?.( makeOnChangePayload(hp, doc, 'crlf-doc', { source: 'local', @@ -242,7 +250,7 @@ describe('createLiveDerivedIndexExtension', () => { const conn = await hp.openDirectConnection('thematic-doc'); const doc = getDoc(conn); - applyExternalChange(hp, 'thematic-doc', '---\n# Title\n'); + applyExternalChange(durabilityState, hp, 'thematic-doc', '---\n# Title\n'); await extension.onChange?.( makeOnChangePayload(hp, doc, 'thematic-doc', { source: 'local', @@ -272,7 +280,12 @@ describe('createLiveDerivedIndexExtension', () => { const conn = await hp.openDirectConnection('autolink-doc'); const doc = getDoc(conn); - applyExternalChange(hp, 'autolink-doc', '# Page\n\nVisit for info\n'); + applyExternalChange( + durabilityState, + hp, + 'autolink-doc', + '# Page\n\nVisit for info\n', + ); await extension.onChange?.( makeOnChangePayload(hp, doc, 'autolink-doc', { source: 'local', @@ -303,7 +316,7 @@ describe('createLiveDerivedIndexExtension', () => { const errorSpy = vi.spyOn(getLogger('live-derived-index'), 'error'); try { - applyExternalChange(hp, 'error-doc', '# Error\n'); + applyExternalChange(durabilityState, hp, 'error-doc', '# Error\n'); await extension.onChange?.( makeOnChangePayload(hp, doc, 'error-doc', { source: 'local', diff --git a/packages/server/src/managed-artifact-persistence.ts b/packages/server/src/managed-artifact-persistence.ts index 4c8aa9f8..6655be35 100644 --- a/packages/server/src/managed-artifact-persistence.ts +++ b/packages/server/src/managed-artifact-persistence.ts @@ -42,7 +42,7 @@ import { withFileLock, } from '@inkeep/open-knowledge-core/server'; import type * as Y from 'yjs'; -import { applyDiskContentToDoc, FILE_WATCHER_ORIGIN } from './external-change.ts'; +import { applyDiskContentToDoc, FILE_WATCHER_ORIGIN } from './disk-content-intake.ts'; import { tracedAtomicFs, tracedMkdir } from './fs-traced.ts'; import { getLogger } from './logger.ts'; diff --git a/packages/server/src/managed-rename-journal.ts b/packages/server/src/managed-rename-journal.ts index ef525655..8a8d064b 100644 --- a/packages/server/src/managed-rename-journal.ts +++ b/packages/server/src/managed-rename-journal.ts @@ -1,6 +1,7 @@ import { existsSync, readdirSync, readFileSync } from 'node:fs'; import { dirname, resolve, sep } from 'node:path'; import { getLocalDir } from './config/paths.ts'; +import { safeContentPath } from './content-path.ts'; import { tracedMkdirSync, tracedRenameSync, @@ -9,7 +10,6 @@ import { tracedWriteFileSync, } from './fs-traced.ts'; import { getLogger } from './logger.ts'; -import { safeContentPath } from './persistence.ts'; const log = getLogger('managed-rename'); diff --git a/packages/server/src/on-agent-write.test.ts b/packages/server/src/on-agent-write.test.ts index dc6b3880..1f51eb05 100644 --- a/packages/server/src/on-agent-write.test.ts +++ b/packages/server/src/on-agent-write.test.ts @@ -18,7 +18,7 @@ import { AgentSessionManager, applyAgentMarkdownWrite, } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; interface CapturedResponse { status: number; diff --git a/packages/server/src/persistence-deferred-store.test.ts b/packages/server/src/persistence-deferred-store.test.ts index 98d797a2..3d62b00d 100644 --- a/packages/server/src/persistence-deferred-store.test.ts +++ b/packages/server/src/persistence-deferred-store.test.ts @@ -6,16 +6,24 @@ import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; import * as Y from 'yjs'; import { composeAndWriteRawBody } from './bridge-intake.ts'; import { __setQuiescentOverrideForTests } from './bridge-quiescence.ts'; +import { DocumentDurabilityState } from './document-durability-state.ts'; import * as fsTraced from './fs-traced.ts'; import { getMetrics, resetMetrics } from './metrics.ts'; import { classifyDeferredStoreError, - createPersistenceExtension, - getReconciledBase, - setBatchInProgress, - switchReconciledBaseScope, + createPersistenceExtension as createPersistenceExtensionBase, + type PersistenceOptions, } from './persistence.ts'; +let durabilityState = new DocumentDurabilityState(); +function createPersistenceExtension(options: PersistenceOptions) { + return createPersistenceExtensionBase({ ...options, durabilityState }); +} +const getReconciledBase = (docName: string) => durabilityState.getReconciledBase(docName); +const setBatchInProgress = (value: boolean) => durabilityState.setBatchInProgress(value); +const switchReconciledBaseScope = (branch: string) => + durabilityState.switchReconciledBaseScope(branch); + const BROWSER_ORIGIN = { source: 'connection', connection: { context: { principalId: 'principal-test' } }, @@ -88,6 +96,7 @@ describe('batch-gated L1 persistence', () => { let tmpDir: string; beforeEach(() => { + durabilityState = new DocumentDurabilityState(); tmpDir = mkdtempSync(join(tmpdir(), 'ok-deferred-store-')); mkdirSync(tmpDir, { recursive: true }); setBatchInProgress(false); @@ -95,8 +104,6 @@ describe('batch-gated L1 persistence', () => { }); afterEach(() => { - setBatchInProgress(false); - switchReconciledBaseScope('main'); rmSync(tmpDir, { recursive: true, force: true }); }); @@ -341,6 +348,7 @@ describe('quiescence gate — deferCount cleanup on disk-write error', () => { let tmpDir: string; beforeEach(() => { + durabilityState = new DocumentDurabilityState(); tmpDir = mkdtempSync(join(tmpdir(), 'ok-defer-disk-error-')); mkdirSync(tmpDir, { recursive: true }); setBatchInProgress(false); @@ -348,8 +356,6 @@ describe('quiescence gate — deferCount cleanup on disk-write error', () => { }); afterEach(() => { - setBatchInProgress(false); - switchReconciledBaseScope('main'); rmSync(tmpDir, { recursive: true, force: true }); }); @@ -475,6 +481,7 @@ describe('Y.Text-is-truth wiring (FR-33 / FR-35)', () => { let tmpDir: string; beforeEach(() => { + durabilityState = new DocumentDurabilityState(); tmpDir = mkdtempSync(join(tmpdir(), 'ok-fr33-wiring-')); mkdirSync(tmpDir, { recursive: true }); setBatchInProgress(false); @@ -482,8 +489,6 @@ describe('Y.Text-is-truth wiring (FR-33 / FR-35)', () => { }); afterEach(() => { - setBatchInProgress(false); - switchReconciledBaseScope('main'); rmSync(tmpDir, { recursive: true, force: true }); }); @@ -582,6 +587,7 @@ describe('FR-9 — deferred-store-failed event + counter', () => { } beforeEach(() => { + durabilityState = new DocumentDurabilityState(); tmpDir = mkdtempSync(join(tmpdir(), 'ok-fr9-deferred-drain-')); mkdirSync(tmpDir, { recursive: true }); setBatchInProgress(false); @@ -594,8 +600,6 @@ describe('FR-9 — deferred-store-failed event + counter', () => { }); afterEach(() => { - setBatchInProgress(false); - switchReconciledBaseScope('main'); warnSpy.mockRestore(); delete process.env.OK_TELEMETRY_VERBOSE; rmSync(tmpDir, { recursive: true, force: true }); @@ -948,15 +952,12 @@ describe('forceStore dispatch (staleness watchdog entry point)', () => { let tmpDir: string; beforeEach(() => { + durabilityState = new DocumentDurabilityState(); tmpDir = mkdtempSync(join(tmpdir(), 'ok-force-store-')); mkdirSync(tmpDir, { recursive: true }); - setBatchInProgress(false); - switchReconciledBaseScope('main'); }); afterEach(() => { - setBatchInProgress(false); - switchReconciledBaseScope('main'); rmSync(tmpDir, { recursive: true, force: true }); }); diff --git a/packages/server/src/persistence-fan-out.test.ts b/packages/server/src/persistence-fan-out.test.ts index f260aba4..2f0ce44d 100644 --- a/packages/server/src/persistence-fan-out.test.ts +++ b/packages/server/src/persistence-fan-out.test.ts @@ -145,7 +145,12 @@ describe('persistence L2 fan-out (US-014)', () => { }); // Simulate file-watcher external change — registers file-system contributor - applyExternalChange(server.hocuspocus, 'fs-writer-doc', '# Updated from disk\n'); + applyExternalChange( + server.durabilityState, + server.hocuspocus, + 'fs-writer-doc', + '# Updated from disk\n', + ); const doc = server.hocuspocus.documents.get('fs-writer-doc'); doc?.removeDirectConnection(); @@ -191,7 +196,12 @@ describe('persistence L2 fan-out (US-014)', () => { recordContributor('concurrent-doc', 'agent-s1', 'Session 1', 'agent-s1'); // Simulate file-watcher change to the same doc — registers file-system contributor - applyExternalChange(server.hocuspocus, 'concurrent-doc', '# Updated concurrently\n'); + applyExternalChange( + server.durabilityState, + server.hocuspocus, + 'concurrent-doc', + '# Updated concurrently\n', + ); const doc = server.hocuspocus.documents.get('concurrent-doc'); doc?.removeDirectConnection(); diff --git a/packages/server/src/persistence-staleness-watchdog.ts b/packages/server/src/persistence-staleness-watchdog.ts index b6c6bd52..9ffdadd3 100644 --- a/packages/server/src/persistence-staleness-watchdog.ts +++ b/packages/server/src/persistence-staleness-watchdog.ts @@ -55,12 +55,7 @@ import { incrementPersistenceStalenessForcedStores, incrementPersistenceStalenessStoodDown, } from './metrics.ts'; -import { - getReconciledBase, - isBatchInProgress, - normalizedSourceForm, - peekInFlightFlush, -} from './persistence.ts'; +import { normalizedSourceForm } from './persistence.ts'; const log = getLogger('persistence-staleness'); @@ -113,10 +108,11 @@ export interface StalenessWatchdogOptions { sweepIntervalMs?: number; /** Test seam for deterministic clocks. */ now?: () => number; - /** Test seams — default to the real persistence/quiescence surfaces. */ - getBase?: (documentName: string) => string | undefined; - isBatchActive?: () => boolean; - peekInFlight?: (documentName: string) => string | undefined; + /** Per-server durability state readers, passed by the composition root. */ + getBase: (documentName: string) => string | undefined; + isBatchActive: () => boolean; + peekInFlight: (documentName: string) => string | undefined; + /** Test seam — defaults to the real quiescence surface. */ msSinceLastUserTx?: (doc: Y.Doc, nowMs: number) => number | null; } @@ -161,9 +157,7 @@ export function createPersistenceStalenessWatchdog( const graceMs = options.graceMs ?? DEFAULT_STALENESS_GRACE_MS; const sweepIntervalMs = options.sweepIntervalMs ?? DEFAULT_STALENESS_SWEEP_INTERVAL_MS; const now = options.now ?? Date.now; - const getBase = options.getBase ?? getReconciledBase; - const isBatchActive = options.isBatchActive ?? isBatchInProgress; - const peekInFlight = options.peekInFlight ?? peekInFlightFlush; + const { getBase, isBatchActive, peekInFlight } = options; const msSinceLastUserTx = options.msSinceLastUserTx ?? getMsSinceLastUserTx; const attempts = new Map(); diff --git a/packages/server/src/persistence-tripwire-read-failure.test.ts b/packages/server/src/persistence-tripwire-read-failure.test.ts index ac2cd1d9..dc3e6c06 100644 --- a/packages/server/src/persistence-tripwire-read-failure.test.ts +++ b/packages/server/src/persistence-tripwire-read-failure.test.ts @@ -29,12 +29,9 @@ import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; import * as Y from 'yjs'; import { composeAndWriteRawBody } from './bridge-intake.ts'; +import { DocumentDurabilityState } from './document-durability-state.ts'; import { getLogger } from './logger.ts'; -import { - createPersistenceExtension, - setReconciledBase, - switchReconciledBaseScope, -} from './persistence.ts'; +import { createPersistenceExtension } from './persistence.ts'; const BROWSER_ORIGIN = { source: 'connection', @@ -62,14 +59,14 @@ async function storeDocument( describe('tripwire reset readFileSync failure', () => { let contentDir: string; + let durabilityState: DocumentDurabilityState; beforeEach(() => { contentDir = realpathSync(mkdtempSync(join(tmpdir(), 'ok-tripwire-readfail-'))); - switchReconciledBaseScope('main'); + durabilityState = new DocumentDurabilityState(); }); afterEach(() => { - switchReconciledBaseScope('main'); rmSync(contentDir, { recursive: true, force: true }); }); @@ -86,11 +83,12 @@ describe('tripwire reset readFileSync failure', () => { contentDir, projectDir: contentDir, gitEnabled: false, + durabilityState, }); const document = new Y.Doc(); composeAndWriteRawBody(document, doubledMarkdown, 'agent'); - setReconciledBase(docName, baseMarkdown); + durabilityState.setReconciledBase(docName, baseMarkdown); const warnSpy = vi.spyOn(getLogger('persistence'), 'warn'); try { diff --git a/packages/server/src/persistence-tripwire-symlink.test.ts b/packages/server/src/persistence-tripwire-symlink.test.ts index 9db1853d..65eea87e 100644 --- a/packages/server/src/persistence-tripwire-symlink.test.ts +++ b/packages/server/src/persistence-tripwire-symlink.test.ts @@ -31,12 +31,9 @@ import { join } from 'node:path'; import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest'; import * as Y from 'yjs'; import { composeAndWriteRawBody } from './bridge-intake.ts'; +import { DocumentDurabilityState } from './document-durability-state.ts'; import { getLogger } from './logger.ts'; -import { - createPersistenceExtension, - setReconciledBase, - switchReconciledBaseScope, -} from './persistence.ts'; +import { createPersistenceExtension } from './persistence.ts'; const BROWSER_ORIGIN = { source: 'connection', @@ -64,6 +61,7 @@ async function storeDocument( describe('tripwire reset symlink-escape', () => { let contentDir: string; + let durabilityState: DocumentDurabilityState; let outsideDir: string; let secretPath: string; const secretContent = '# SECRET\n\nThis content lives outside the content root.\n'; @@ -73,11 +71,10 @@ describe('tripwire reset symlink-escape', () => { outsideDir = realpathSync(mkdtempSync(join(tmpdir(), 'ok-tripwire-outside-'))); secretPath = join(outsideDir, 'secret.md'); writeFileSync(secretPath, secretContent, 'utf-8'); - switchReconciledBaseScope('main'); + durabilityState = new DocumentDurabilityState(); }); afterEach(() => { - switchReconciledBaseScope('main'); rmSync(contentDir, { recursive: true, force: true }); rmSync(outsideDir, { recursive: true, force: true }); }); @@ -98,6 +95,7 @@ describe('tripwire reset symlink-escape', () => { contentDir, projectDir: contentDir, gitEnabled: false, + durabilityState, }); // Hand-construct the Y.Doc state the tripwire path expects: XmlFragment @@ -106,7 +104,7 @@ describe('tripwire reset symlink-escape', () => { // route into the reset disk-read. const document = new Y.Doc(); composeAndWriteRawBody(document, doubledMarkdown, 'agent'); - setReconciledBase(docName, baseMarkdown); + durabilityState.setReconciledBase(docName, baseMarkdown); const warnSpy = vi.spyOn(getLogger('persistence'), 'warn'); try { diff --git a/packages/server/src/persistence-ytext-truth.test.ts b/packages/server/src/persistence-ytext-truth.test.ts index 89b5d53c..0e3ad407 100644 --- a/packages/server/src/persistence-ytext-truth.test.ts +++ b/packages/server/src/persistence-ytext-truth.test.ts @@ -23,7 +23,7 @@ * structurally avoided). */ -import { describe as _bunDescribe, afterEach, beforeEach, expect, test, vi } from 'vitest'; +import { describe as _vitestDescribe, afterEach, beforeEach, expect, test, vi } from 'vitest'; // Skip-on-CI gate (oven-sh/bun#11892 — child-process reaping bug). The full // contract surface lives in this file: every test boots a real @@ -44,7 +44,7 @@ import { describe as _bunDescribe, afterEach, beforeEach, expect, test, vi } fro // a full canonical-gate run on ubuntu-latest GHA is green for ≥5 consecutive // runs of this file. Track via the shared CI-skip pattern across server tests // (~49 files). -const describe = process.env.CI ? _bunDescribe.skip : _bunDescribe; +const describe = process.env.CI ? _vitestDescribe.skip : _vitestDescribe; import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; import { tmpdir } from 'node:os'; @@ -54,7 +54,6 @@ import simpleGit from 'simple-git'; import { __resetQuiescenceForTests, __setQuiescentOverrideForTests } from './bridge-quiescence.ts'; import { __resetBridgeWatchdogForTests } from './bridge-watchdog.ts'; import { getMetrics, resetMetrics } from './metrics.ts'; -import { getReconciledBase } from './persistence.ts'; import { createServer } from './server-factory.ts'; interface Fixture { @@ -256,9 +255,9 @@ describe('FR-35: cold-load setReconciledBase stores raw disk bytes', () => { const conn = await server.hocuspocus.openDirectConnection(docName); // Wait for onLoadDocument to populate the doc + set reconciledBase. - await waitForCondition(() => getReconciledBase(docName) !== undefined); + await waitForCondition(() => server.durabilityState.getReconciledBase(docName) !== undefined); // reconciledBase is the raw disk content verbatim. - expect(getReconciledBase(docName)).toBe(rawDiskContent); + expect(server.durabilityState.getReconciledBase(docName)).toBe(rawDiskContent); conn.disconnect(); } finally { @@ -291,7 +290,7 @@ describe('FR-35: cold-load setReconciledBase stores raw disk bytes', () => { await server.ready; const conn = await server.hocuspocus.openDirectConnection(docName); // Wait for cold-load to settle. - await waitForCondition(() => getReconciledBase(docName) !== undefined); + await waitForCondition(() => server.durabilityState.getReconciledBase(docName) !== undefined); // Wait briefly to give the debounce a chance to fire if it would. await new Promise((r) => setTimeout(r, 250)); @@ -338,7 +337,7 @@ describe('FR-33: full round-trip preserves user-form bytes', () => { const serverDoc = server.hocuspocus.documents.get(docName); if (!serverDoc) return; - await waitForCondition(() => getReconciledBase(docName) !== undefined); + await waitForCondition(() => server.durabilityState.getReconciledBase(docName) !== undefined); const ytextAfterLoad = serverDoc.getText('source').toString(); // ytext should hold the disk content verbatim post-cold-load. expect(ytextAfterLoad).toBe(initialContent); @@ -424,7 +423,7 @@ describe('Quiescence gate via direct counter manipulation', () => { const serverDoc = server.hocuspocus.documents.get(docName); if (!serverDoc) return; - await waitForCondition(() => getReconciledBase(docName) !== undefined); + await waitForCondition(() => server.durabilityState.getReconciledBase(docName) !== undefined); // Pin the predicate to non-quiescent for this doc. Yjs's // afterAllTransactions fires synchronously after every drain, so @@ -499,7 +498,7 @@ describe('Quiescence gate via direct counter manipulation', () => { const conn = await server.hocuspocus.openDirectConnection(docName); const serverDoc = server.hocuspocus.documents.get(docName); if (!serverDoc) return; - await waitForCondition(() => getReconciledBase(docName) !== undefined); + await waitForCondition(() => server.durabilityState.getReconciledBase(docName) !== undefined); __setQuiescentOverrideForTests(serverDoc, false); @@ -572,7 +571,7 @@ describe('Quiescence gate via direct counter manipulation', () => { const conn = await server.hocuspocus.openDirectConnection(docName); const serverDoc = server.hocuspocus.documents.get(docName); if (!serverDoc) return; - await waitForCondition(() => getReconciledBase(docName) !== undefined); + await waitForCondition(() => server.durabilityState.getReconciledBase(docName) !== undefined); const userOrigin = { source: 'connection' as const, diff --git a/packages/server/src/persistence.test.ts b/packages/server/src/persistence.test.ts index e38a5fe2..8ed1fcd7 100644 --- a/packages/server/src/persistence.test.ts +++ b/packages/server/src/persistence.test.ts @@ -14,13 +14,9 @@ import { tmpdir } from 'node:os'; import { dirname, join, resolve, sep } from 'node:path'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; import * as Y from 'yjs'; +import { isWithinContentDir, safeContentPath } from './content-path.ts'; import { contentHash, isSelfWrite, registerWrite } from './file-watcher'; -import { - captureDocSnapshotForPersistence, - isWithinContentDir, - resolveWriterFromOrigin, - safeContentPath, -} from './persistence'; +import { captureDocSnapshotForPersistence, resolveWriterFromOrigin } from './persistence'; import { FILE_SYSTEM_WRITER, GIT_UPSTREAM_WRITER, SERVICE_WRITER } from './shadow-repo'; describe('safeContentPath', () => { diff --git a/packages/server/src/persistence.ts b/packages/server/src/persistence.ts index e2eca027..d22ab238 100644 --- a/packages/server/src/persistence.ts +++ b/packages/server/src/persistence.ts @@ -54,6 +54,7 @@ import { } from './cc1-broadcast.ts'; import { type ConfigPersistenceCtx, loadConfigDoc, storeConfigDoc } from './config-persistence.ts'; import { frozenDocLifecycleStatus } from './conflict-errors.ts'; +import { isWithinContentDir, safeContentPath } from './content-path.ts'; import type { ContributorEntry } from './contributor-tracker.ts'; import { contributorCount, @@ -63,8 +64,8 @@ import { restoreContributors, swapContributors, } from './contributor-tracker.ts'; -import { docNameToRelativePath } from './doc-extensions.ts'; -import { applyDiskContentToDoc, FILE_WATCHER_ORIGIN } from './external-change.ts'; +import { applyDiskContentToDoc, FILE_WATCHER_ORIGIN } from './disk-content-intake.ts'; +import { DocumentDurabilityState, type StoreFailure } from './document-durability-state.ts'; import { contentHash, registerWrite } from './file-watcher.ts'; import { tracedMkdir, tracedRename, tracedUnlinkSync, tracedWriteFile } from './fs-traced.ts'; import { errnoCode } from './http/handler-utils.ts'; @@ -92,7 +93,7 @@ import { incrementPersistenceSkipNonQuiescent, incrementPersistenceStoreRemovedDoc, } from './metrics.ts'; -import { isWithinDir, toPosix } from './path-utils.ts'; +import { toPosix } from './path-utils.ts'; import { classifyDuplication } from './persistence-tripwire.ts'; import { backfillRenameLogCommitSha, getOrLoadRenameLogIndex } from './rename-log.ts'; import { OBSERVER_SYNC_ORIGIN } from './server-observers.ts'; @@ -276,6 +277,12 @@ export function classifyDeferredStoreError(err: unknown): DeferredStoreErrorClas export interface PersistenceOptions { contentDir: string; projectDir: string; + /** + * Durability registry shared with other server extensions. Standalone + * persistence callers may omit it; composition callers must give every + * durability consumer the same instance exposed on `PersistenceHandle`. + */ + durabilityState?: DocumentDurabilityState; gitEnabled?: boolean; commitDebounceMs?: number; wipRef?: string; @@ -453,128 +460,6 @@ export function normalizedSourceForm(rawYText: string): string { const { frontmatter, body } = stripFrontmatter(rawYText); return normalizeBridge(prependFrontmatter(frontmatter, body)); } - -export function safeContentPath(documentName: string, contentDir: string): string { - if (documentName.includes('\x00')) { - throw new Error(`Invalid document name: ${documentName}`); - } - const relativePath = docNameToRelativePath(documentName); - const filePath = resolve(contentDir, relativePath); - if (!isWithinDir(filePath, contentDir)) { - throw new Error(`Invalid document name: ${documentName}`); - } - return filePath; -} - -export function isWithinContentDir(p: string, contentDir: string): boolean { - return isWithinDir(p, contentDir); -} - -/** - * Reconciled base: last known-good markdown for each document, scoped by branch. - * Updated on load, store, and reconciliation. Used as the merge base - * for three-way reconciliation. - * - * Outer key = branch name (e.g. "main", "feature/xyz", "detached-abc123def456") - * Inner key = docName, value = last-synced markdown content - */ -const reconciledBaseByBranch = new Map>(); - -/** Active branch scope for reconciledBase lookups. Defaults to 'main'. */ -let activeBranch = 'main'; - -/** Switch the active branch scope. Creates a fresh scope if first visit. */ -export function switchReconciledBaseScope(branch: string): void { - activeBranch = branch; - if (!reconciledBaseByBranch.has(branch)) { - reconciledBaseByBranch.set(branch, new Map()); - } -} - -/** Get the active branch name for reconciledBase. */ -export function getActiveBranch(): string { - return activeBranch; -} - -/** Get the reconciledBase value for a doc in the active branch scope. */ -export function getReconciledBase(docName: string): string | undefined { - return reconciledBaseByBranch.get(activeBranch)?.get(docName); -} - -/** Set the reconciledBase value for a doc in the active branch scope. */ -export function setReconciledBase(docName: string, content: string): void { - if (!reconciledBaseByBranch.has(activeBranch)) { - reconciledBaseByBranch.set(activeBranch, new Map()); - } - reconciledBaseByBranch.get(activeBranch)?.set(docName, content); -} - -/** Delete the reconciledBase entry for a doc in the active branch scope. */ -export function deleteReconciledBase(docName: string): void { - reconciledBaseByBranch.get(activeBranch)?.delete(docName); -} - -/** - * In-flight flush snapshots, keyed by docName, value = `normalizeBridge`d - * markdown of the flush currently committing to disk. The disk commit in - * `storeDocumentNow` is non-atomic w.r.t. concurrent readers: the rename - * lands first, `setReconciledBase` runs later in the promise continuation. - * A reader comparing disk against the base inside that gap sees the - * server's OWN just-flushed bytes against a stale base — phantom foreign - * divergence. This map is the producer-owned discriminator: set before the - * commit's first await, cleared (only if still ours — overlapping flushes - * for one doc overwrite the entry) after the base advances. On the L3 - * divergence-abort path the snapshot briefly advertises bytes that never - * reached disk — benign: disk holds non-matching foreign content there, so - * no false own-write match is possible; the .finally clears it. - * - * Not branch-scoped, unlike `reconciledBaseByBranch`: entries live for the - * milliseconds of one disk commit. A cross-branch switch does not drain a - * flush already mid-commit (onBatchEnd only defers NEW stores via - * `isBatchInProgress` → `deferStore`), but a stale entry outliving the - * switch only suppresses a reconcile when the NEW branch's disk bytes are - * byte-identical to the old flush snapshot — identical content, where the - * skip is a no-op and the next flush advances the base anyway. - */ -const inFlightFlushByDoc = new Map(); - -/** - * Read-only peek at the in-flight flush snapshot for a doc (normalized - * markdown), or undefined when no flush is mid-commit. Consumers compare - * `normalizeBridge(disk)` against it to recognize the server's own bytes; - * they must never mutate the entry — persistence alone owns its lifecycle. - */ -export function peekInFlightFlush(docName: string): string | undefined { - return inFlightFlushByDoc.get(docName); -} - -/** Batch-in-progress flag — gates L1 writes and L2 commits during coordinated git operations. */ -let batchInProgress = false; - -export function setBatchInProgress(value: boolean): void { - batchInProgress = value; -} - -export function isBatchInProgress(): boolean { - return batchInProgress; -} - -/** - * A disk-store failure captured out-of-band. `storeDocumentNow` records one - * when its atomic write throws (ENOSPC / EDQUOT / EACCES / EROFS / read-only - * FS, etc.) and rethrows — Hocuspocus's `storeDocumentHooks` then catches the - * rethrow, logs "stays in memory to avoid data loss", and resolves WITHOUT - * signaling the write that triggered the flush. A handler that force-flushes - * the store (`executeNow`) and then calls `takeStoreFailure` is the only way to - * learn the bytes never reached disk and report disk truth rather than a false - * "Written successfully". - */ -export interface StoreFailure { - /** Node errno code when available (e.g. `ENOSPC`, `EACCES`, `EROFS`). */ - code?: string; - message: string; -} - /** * Extract a {@link StoreFailure} from a thrown value without touching it in a * way that can itself throw. Disk-write rejections are normally @@ -616,37 +501,9 @@ export interface PersistenceQueueDepths { export interface PersistenceHandle { extension: Extension; + readonly durabilityState: DocumentDurabilityState; flushDeferredStores: (mode?: 'within-branch' | 'discard-stale') => Promise; flushPendingGitCommit: () => Promise; - /** - * Read-and-clear the last recorded disk-store failure for `documentName`, - * or null if the most recent store reached disk. - * - * Precondition: call this ONLY immediately after force-flushing THIS doc's - * `onStoreDocument` debounce (`debouncer.executeNow('onStoreDocument-')`). - * The flush records the failure (or clears it on success) synchronously, so - * the read reflects that store. Calling it without a preceding force-flush of - * the same doc risks reading a stale cross-write residue. - */ - takeStoreFailure: (documentName: string) => StoreFailure | null; - /** - * Read-and-clear whether `documentName`'s most recent agent-triggered store - * was REVERTED by the L3 disk-divergence backstop — disk diverged - * from the reconciled base, so the overwrite was aborted and disk won. - * - * Same precondition as {@link takeStoreFailure}: call ONLY immediately after - * force-flushing THIS doc's `onStoreDocument` debounce. Mutually exclusive with - * a store failure for the same flush (L3 returns before the atomic write). - */ - takeStoreDivergence: (documentName: string) => boolean; - /** - * Mark `documentName`'s next store as agent-write-triggered (L3 - * gate). Call IMMEDIATELY before force-flushing this doc's `onStoreDocument` - * debounce from an agent write handler. `storeDocumentNow` read-and-clears it; - * only a marked store can disk-wins-revert on divergence, so human-editor - * stores (which never route through a handler's force-flush) are excluded. - */ - markAgentWriteStore: (documentName: string) => void; /** * Force a drain of the contributor map regardless of timer state. Used by * write surfaces that mutate `pendingContributors` outside any Y.Doc transact @@ -693,6 +550,7 @@ export interface PersistenceHandle { } export function createPersistenceExtension(options?: PersistenceOptions): PersistenceHandle { + const durabilityState = options?.durabilityState ?? new DocumentDurabilityState(); const contentDirRaw = options?.contentDir ?? process.cwd(); let contentDir: string; try { @@ -742,8 +600,8 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis projectDir, homedirOverride: options?.configHomedirOverride, lkgCache: managedArtifactLkgCache, - setReconciledBase, - getReconciledBase, + setReconciledBase: (docName, content) => durabilityState.setReconciledBase(docName, content), + getReconciledBase: (docName) => durabilityState.getReconciledBase(docName), }; // Mermaid (`.mmd`/`.mermaid`) persistence ctx. Y.Text-only, content-dir paths, @@ -782,40 +640,6 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis const QUIESCENCE_MAX_DEFER = 8; const persistenceDeferCounts = new Map(); - // Last disk-store failure per docName. Set when `storeDocumentNow`'s atomic - // write throws (before it rethrows for Hocuspocus to keep the doc in memory); - // cleared on the next successful store and read-and-cleared via - // `takeStoreFailure`. See `StoreFailure` for why this out-of-band channel is - // necessary — Hocuspocus swallows the rethrow without signaling the caller. - const storeFailures = new Map(); - - // Docs whose most recent agent-triggered store was REVERTED by the L3 - // disk-divergence backstop instead of written: the store detected - // that disk diverged from the reconciled base since L1's check (the residual - // TOCTOU), aborted the overwrite, and ingested disk (disk-wins). The agent's - // edit was discarded. Out-of-band like `storeFailures` because Hocuspocus's - // store hook gives the triggering handler no return channel; the handler - // force-flushes then read-clears via `takeStoreDivergence` and returns - // `urn:ok:error:disk-divergence`. Mutually exclusive with `storeFailures` for - // the same flush — L3 returns before the write, so no atomic-write throw can - // also record a failure for that store. - const storeDivergences = new Set(); - - // Docs whose pending store was forced by an agent write handler's awaited - // flush (L3 gate). Set by `markAgentWriteStore` immediately before - // the handler calls `executeNow`, consumed (deleted) at the top of - // `storeDocumentNow`. This is the agent-triggered signal: Hocuspocus passes - // `lastTransactionOrigin: null` for agent DirectConnection writes, so the - // origin can't gate L3 — but the handler KNOWS it's an agent write. Human- - // editor (browser) stores fire via the natural debounce and never route - // through a handler's `executeNow`, so they never get marked → L3 never - // disk-wins-reverts a human's in-progress typing. - const agentWriteStores = new Set(); - - // reconciledBase and batchInProgress use the module-level systems - // (reconciledBaseByBranch via get/setReconciledBase, and isBatchInProgress) - // so that server-factory.ts and persistence stay in sync. - const gitEnabled = options?.gitEnabled ?? true; const commitDebounceMs = options?.commitDebounceMs ?? 15_000; const wipRef = options?.wipRef ?? 'refs/wip/main'; @@ -1120,7 +944,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis function scheduleGitCommit(): void { if (!gitEnabled) return; - if (isBatchInProgress()) return; + if (durabilityState.isBatchInProgress()) return; if (gitCommitTimer) clearTimeout(gitCommitTimer); gitCommitTimer = setTimeout(() => { gitCommitTimer = null; @@ -1307,7 +1131,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis // store forced by an agent write handler's awaited flush? Read-and-clear // here so it can't leak to a later human-editor store of the same doc // (e.g. if this store no-ops at the unchanged-base skip below). - const agentTriggeredStore = agentWriteStores.delete(documentName); + const agentTriggeredStore = durabilityState.consumeAgentWriteStore(documentName); // Lifecycle guard: when the file watcher saw an external delete or // rename, the disk-event handler in standalone.ts sets the doc's @@ -1523,7 +1347,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis // class as a no-op skips both the disk write AND the principal // safety-net below, preventing phantom commits attributed to the // browser's principal when a later agent write triggers the L2 fan-out. - const currentBase = getReconciledBase(documentName); + const currentBase = durabilityState.getReconciledBase(documentName); // Routed through the shared helper so this compare and the // staleness watchdog's divergence predicate stay one derivation. const normalizedMarkdown = normalizedSourceForm(ytextSnapshot); @@ -1729,7 +1553,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis // lifecycle synchronous from the hook's entry, which the overlap // test relies on to deterministically pin clear-only-if-still-ours. inFlightFlushValue = normalizeBridge(markdown); - inFlightFlushByDoc.set(documentName, inFlightFlushValue); + durabilityState.beginInFlightFlush(documentName, inFlightFlushValue); const requestedPath = safeContentPath(documentName, contentDir); await tracedMkdir(dirname(requestedPath), { recursive: true }); @@ -1854,12 +1678,12 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis applyDiskContent(document, diskContent); }, FILE_WATCHER_ORIGIN); } catch (err) { - storeFailures.set(documentName, toStoreFailure(err)); + durabilityState.recordStoreFailure(documentName, toStoreFailure(err)); persistenceDeferCounts.delete(documentName); throw err; } - setReconciledBase(documentName, diskContent); - storeDivergences.add(documentName); + durabilityState.setReconciledBase(documentName, diskContent); + durabilityState.recordStoreDivergence(documentName); persistenceDeferCounts.delete(documentName); return; } @@ -1896,7 +1720,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis registerWrite(canonicalPath, contentHash(markdown)); // The bytes reached disk — clear any prior recorded failure so a // later force-flush + `takeStoreFailure` reflects the success. - storeFailures.delete(documentName); + durabilityState.clearStoreFailure(documentName); // Increment disk-write counter after the atomic rename succeeds. // Regression gate: if OBSERVER_SYNC_ORIGIN drops skipStoreHooks, // observer writes trigger onStoreDocument and produce amplified @@ -1946,7 +1770,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis // `toStoreFailure` reads `e` defensively so a throwing-getter error // shape can't replace `e` here and rob the deferred-store classifier // downstream of the original throw. - storeFailures.set(documentName, toStoreFailure(e)); + durabilityState.recordStoreFailure(documentName, toStoreFailure(e)); log.error({ err: e, documentName }, `[persistence] Failed to save ${documentName}`); throw e; } @@ -1956,7 +1780,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis ); // Update reconciled base after successful store - setReconciledBase(documentName, markdown); + durabilityState.setReconciledBase(documentName, markdown); tripwireResetFailedDocs.delete(documentName); persistenceDeferCounts.delete(documentName); @@ -1979,11 +1803,8 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis // (possibly failed) flush must not delete that newer signal. On a // commit throw the entry clears here too, so the signal can never // stay stuck set past the flush that owns it. - if ( - inFlightFlushValue !== undefined && - inFlightFlushByDoc.get(documentName) === inFlightFlushValue - ) { - inFlightFlushByDoc.delete(documentName); + if (inFlightFlushValue !== undefined) { + durabilityState.finishInFlightFlush(documentName, inFlightFlushValue); } // doc.name deliberately NOT recorded on the histogram — per-doc cardinality // would blow up Prometheus label storage at scale. The span carries it. @@ -2001,7 +1822,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis lastTransactionOrigin: unknown; }): void { deferredStores.set(documentName, { - branch: getActiveBranch(), + branch: durabilityState.getActiveBranch(), document, lastTransactionOrigin, }); @@ -2024,7 +1845,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis if (drainMode !== 'discard-stale') { for (const [documentName, entry] of entries) { - if (entry.branch !== getActiveBranch()) continue; + if (entry.branch !== durabilityState.getActiveBranch()) continue; try { await storeDocumentNow({ document: entry.document, @@ -2264,7 +2085,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis // and the bridge invariant comparator (`assertBridgeInvariant` // and `attachBridgeInvariantWatcher`) tolerates the same classes, // so no false-positive write fires on mere file open. - setReconciledBase(documentName, raw); + durabilityState.setReconciledBase(documentName, raw); }, ).finally(() => { // doc.name deliberately NOT recorded on the histogram — per-doc cardinality @@ -2354,7 +2175,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis await storeMermaidDoc(document, documentName, lastTransactionOrigin, mermaidPersistenceCtx); return; } - if (isBatchInProgress()) { + if (durabilityState.isBatchInProgress()) { deferStore({ document, documentName, lastTransactionOrigin }); return; } @@ -2403,22 +2224,6 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis await commitInFlight; } - function takeStoreFailure(documentName: string): StoreFailure | null { - const failure = storeFailures.get(documentName); - if (failure) storeFailures.delete(documentName); - return failure ?? null; - } - - function takeStoreDivergence(documentName: string): boolean { - if (!storeDivergences.has(documentName)) return false; - storeDivergences.delete(documentName); - return true; - } - - function markAgentWriteStore(documentName: string): void { - agentWriteStores.add(documentName); - } - function getQueueDepths(): PersistenceQueueDepths { return { branchDeferred: deferredStores.size, @@ -2430,7 +2235,7 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis if (isPersistenceExcludedDoc(documentName)) { return; } - if (isBatchInProgress()) { + if (durabilityState.isBatchInProgress()) { // Same parking the debounced hook applies; the batch's own // `flushDeferredStores` replays it when the coordinated operation ends. deferStore({ document, documentName, lastTransactionOrigin: null }); @@ -2444,13 +2249,11 @@ export function createPersistenceExtension(options?: PersistenceOptions): Persis return { extension, + durabilityState, flushDeferredStores, flushPendingGitCommit, flushContributors, waitForPendingCommits, - takeStoreFailure, - takeStoreDivergence, - markAgentWriteStore, getQueueDepths, forceStore, configPersistenceCtx, diff --git a/packages/server/src/prd-6654-e2e-repro.test.ts b/packages/server/src/prd-6654-e2e-repro.test.ts index 54f895a2..83b2cf51 100644 --- a/packages/server/src/prd-6654-e2e-repro.test.ts +++ b/packages/server/src/prd-6654-e2e-repro.test.ts @@ -23,7 +23,7 @@ import { getSchema } from '@tiptap/core'; import { describe, expect, test } from 'vitest'; import * as Y from 'yjs'; import { AGENT_WRITE_ORIGIN, AgentSessionManager } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { composeAndWriteRawBody } from './bridge-intake.ts'; import { createServerObserverExtension } from './server-observer-extension.ts'; diff --git a/packages/server/src/reconcile-own-flush-window.test.ts b/packages/server/src/reconcile-own-flush-window.test.ts index 4f705f7a..c865bb8a 100644 --- a/packages/server/src/reconcile-own-flush-window.test.ts +++ b/packages/server/src/reconcile-own-flush-window.test.ts @@ -38,17 +38,12 @@ import { normalizeBridge } from '@inkeep/open-knowledge-core'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; import * as Y from 'yjs'; import { isDocInConflict } from './conflict-errors.ts'; +import { DocumentDurabilityState } from './document-durability-state.ts'; import { type ReconcileBeforeWriteResult, reconcileDiskBeforeAgentWrite, } from './external-change.ts'; -import { - createPersistenceExtension, - getReconciledBase, - peekInFlightFlush, - setBatchInProgress, - switchReconciledBaseScope, -} from './persistence.ts'; +import { createPersistenceExtension } from './persistence.ts'; const BROWSER_ORIGIN = { source: 'connection', @@ -137,6 +132,7 @@ async function drivePhantomDivergence( tmpDir: string, docName: string, document: Y.Doc, + durabilityState: DocumentDurabilityState, options: { diskContentInWindow?: string } = {}, ): Promise { const docPath = join(tmpDir, `${docName}.md`); @@ -154,18 +150,19 @@ async function drivePhantomDivergence( contentDir: tmpDir, projectDir: tmpDir, gitEnabled: false, + durabilityState, onDiskFlush: (name) => { if (name !== docName || windowFired) return; windowFired = true; // Live doc moves past the flush snapshot while the commit is mid-flight // (in production: further agent/user edits landing during the flush). document.transact(() => replaceDocParagraphs(document, LIVE_PARAGRAPHS), BROWSER_ORIGIN); - probe.baseSeenInWindow = getReconciledBase(docName); + probe.baseSeenInWindow = durabilityState.getReconciledBase(docName); // Pins set-before-window ordering: the signal must already be set when // a guard read can first observe the renamed bytes. Without this, a // reorder moving the set after the rename would silently disable the // own-flush discrimination while the suite stays green. - probe.inFlightSeenInWindow = peekInFlightFlush(docName); + probe.inFlightSeenInWindow = durabilityState.peekInFlightFlush(docName); // A FOREIGN writer landing inside the same window (bytes ≠ the // in-flight snapshot). if (options.diskContentInWindow !== undefined) { @@ -174,6 +171,7 @@ async function drivePhantomDivergence( // The agent write's guard reads disk inside the server's own // flush-commit window — the production race, hit deterministically. probe.windowResult = reconcileDiskBeforeAgentWrite( + durabilityState, fakeHocuspocusWith(docName, document), docName, tmpDir, @@ -196,6 +194,7 @@ async function drivePhantomDivergence( describe('reconcileDiskBeforeAgentWrite — own persistence flush is not foreign divergence', () => { let tmpDir: string; let document: Y.Doc; + let durabilityState: DocumentDurabilityState; beforeEach(() => { // realpathSync: macOS /var → /private/var; without it the guard's @@ -203,21 +202,18 @@ describe('reconcileDiskBeforeAgentWrite — own persistence flush is not foreign // and the test would pass vacuously. tmpDir = realpathSync(mkdtempSync(join(tmpdir(), 'ok-own-flush-window-'))); mkdirSync(tmpDir, { recursive: true }); - setBatchInProgress(false); - switchReconciledBaseScope('main'); + durabilityState = new DocumentDurabilityState(); document = new Y.Doc(); }); afterEach(() => { document.destroy(); - setBatchInProgress(false); - switchReconciledBaseScope('main'); rmSync(tmpDir, { recursive: true, force: true }); }); test("an agent write inside the server's own flush-commit window is not refused and does not latch lifecycle conflict", async () => { const docName = 'own-flush-window'; - const probe = await drivePhantomDivergence(tmpDir, docName, document); + const probe = await drivePhantomDivergence(tmpDir, docName, document, durabilityState); // The disk bytes the guard read were the server's OWN in-flight flush — // not an out-of-band edit. Latching `lifecycle.status = 'conflict'` here @@ -244,7 +240,7 @@ describe('reconcileDiskBeforeAgentWrite — own persistence flush is not foreign // normalization), the foreign edit is misclassified as own-write, the // guard skips, and this test fails on reconciled === false. const FOREIGN_CONTENT = 'alpha FOREIGN EDIT\n\nbeta\n'; - const probe = await drivePhantomDivergence(tmpDir, docName, document, { + const probe = await drivePhantomDivergence(tmpDir, docName, document, durabilityState, { diskContentInWindow: FOREIGN_CONTENT, }); @@ -261,14 +257,15 @@ describe('reconcileDiskBeforeAgentWrite — own persistence flush is not foreign test('no permanent 409 wedge: after the flush settles, subsequent agent writes are not refused', async () => { const docName = 'own-flush-wedge'; - await drivePhantomDivergence(tmpDir, docName, document); + await drivePhantomDivergence(tmpDir, docName, document, durabilityState); // The flush has fully committed: disk == base == the flushed bytes; the // only possible residue of the phantom divergence is the lifecycle latch. - expect(getReconciledBase(docName)).toBe(FLUSHED_CONTENT); + expect(durabilityState.getReconciledBase(docName)).toBe(FLUSHED_CONTENT); // A later agent write runs the same guard first… const laterGuard = reconcileDiskBeforeAgentWrite( + durabilityState, fakeHocuspocusWith(docName, document), docName, tmpDir, @@ -294,6 +291,7 @@ describe('reconcileDiskBeforeAgentWrite — own persistence flush is not foreign contentDir: tmpDir, projectDir: tmpDir, gitEnabled: false, + durabilityState, }); await loadDocument(persistence, document, docName); document.transact(() => replaceDocParagraphs(document, FLUSHED_PARAGRAPHS), BROWSER_ORIGIN); @@ -314,13 +312,13 @@ describe('reconcileDiskBeforeAgentWrite — own persistence flush is not foreign // A stuck signal would make the guard treat any FUTURE foreign disk edit // that happens to match the failed flush's bytes as the server's own. - expect(peekInFlightFlush(docName)).toBeUndefined(); + expect(durabilityState.peekInFlightFlush(docName)).toBeUndefined(); // Symmetric invariant: the base must NOT have advanced to the never-landed // flush. setReconciledBase runs only after a successful rename, so the // faulted store leaves the base at the last good content. A spuriously // advanced base would make the next real disk edit look in-sync and skip // reconcile. - expect(getReconciledBase(docName)).toBe(BASE_CONTENT); + expect(durabilityState.getReconciledBase(docName)).toBe(BASE_CONTENT); }); test("an earlier overlapping flush settling does not clear a later flush's in-flight signal", async () => { @@ -338,6 +336,7 @@ describe('reconcileDiskBeforeAgentWrite — own persistence flush is not foreign contentDir: tmpDir, projectDir: tmpDir, gitEnabled: false, + durabilityState, onDiskFlush: (name) => { if (name !== docName || windowFired) return; windowFired = true; @@ -346,7 +345,7 @@ describe('reconcileDiskBeforeAgentWrite — own persistence flush is not foreign // it in storeDocumentNow), overwriting A's entry. document.transact(() => replaceDocParagraphs(document, OVERLAP_PARAGRAPHS), BROWSER_ORIGIN); laterFlush = storeDocument(persistence, document, docName); - peekAfterLaterStart = peekInFlightFlush(docName); + peekAfterLaterStart = durabilityState.peekInFlightFlush(docName); }, }); @@ -361,10 +360,10 @@ describe('reconcileDiskBeforeAgentWrite — own persistence flush is not foreign // must NOT have cleared B's still-live signal: the clear is guarded on // the entry still being A's own. Dropping that guard in a "simplify" // refactor reintroduces the unguarded-window wedge this suite pins. - expect(peekInFlightFlush(docName)).toBe(normalizeBridge(OVERLAP_CONTENT)); + expect(durabilityState.peekInFlightFlush(docName)).toBe(normalizeBridge(OVERLAP_CONTENT)); await laterFlush; // B's own settle clears its entry — no stuck signal. - expect(peekInFlightFlush(docName)).toBeUndefined(); + expect(durabilityState.peekInFlightFlush(docName)).toBeUndefined(); }); }); diff --git a/packages/server/src/save-version.test.ts b/packages/server/src/save-version.test.ts index b07989e8..b5d552c3 100644 --- a/packages/server/src/save-version.test.ts +++ b/packages/server/src/save-version.test.ts @@ -16,7 +16,7 @@ import simpleGit from 'simple-git'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; import * as Y from 'yjs'; import { AgentSessionManager } from './agent-sessions.ts'; -import { createApiExtension } from './api-extension.ts'; +import { createApiExtension } from './api-extension.test-helper.ts'; import { commitWip, initShadowRepo, diff --git a/packages/server/src/serialize-doc-ytext.test.ts b/packages/server/src/serialize-doc-ytext.test.ts index ebd4fe98..c18c056d 100644 --- a/packages/server/src/serialize-doc-ytext.test.ts +++ b/packages/server/src/serialize-doc-ytext.test.ts @@ -33,10 +33,9 @@ import { existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'no import { tmpdir } from 'node:os'; import { join } from 'node:path'; import simpleGit from 'simple-git'; -import { describe as _bunDescribe, afterEach, beforeEach, expect, test, vi } from 'vitest'; +import { describe as _vitestDescribe, afterEach, beforeEach, expect, test, vi } from 'vitest'; import { __resetQuiescenceForTests } from './bridge-quiescence.ts'; import { resetMetrics } from './metrics.ts'; -import { getReconciledBase } from './persistence.ts'; import { createServer } from './server-factory.ts'; // Skip-on-CI gate (oven-sh/bun#11892 — child-process reaping bug). Same @@ -56,10 +55,10 @@ import { createServer } from './server-factory.ts'; // Re-enable condition: drop this gate when oven-sh/bun#11892 is closed AND // a full canonical-gate run on ubuntu-latest GHA is green for ≥5 consecutive // runs. -const describe = process.env.CI ? _bunDescribe.skip : _bunDescribe; +const describe = process.env.CI ? _vitestDescribe.skip : _vitestDescribe; // File-watcher startup + reconcile fire is parcel-watcher latency-bound; -// 5s Bun default isn't enough headroom under suite contention. Tests +// Vitest's default timeout isn't enough headroom under suite contention. Tests // inside still bound their own waits via `waitForCondition`. vi.setConfig({ testTimeout: 20_000, hookTimeout: 20_000 }); @@ -136,8 +135,8 @@ describe('FR-34: serializeDoc returns ytext bytes verbatim', () => { await server.ready; const conn = await server.hocuspocus.openDirectConnection(docName); // Cold-load completed; reconciledBase is raw disk bytes. - await waitForCondition(() => getReconciledBase(docName) !== undefined); - expect(getReconciledBase(docName)).toBe(initialContent); + await waitForCondition(() => server.durabilityState.getReconciledBase(docName) !== undefined); + expect(server.durabilityState.getReconciledBase(docName)).toBe(initialContent); // External change: write new bytes to disk. The file-watcher fires // a within-branch update event → reconcile case 'update' → @@ -147,10 +146,13 @@ describe('FR-34: serializeDoc returns ytext bytes verbatim', () => { writeFileSync(docPath, updatedContent, 'utf-8'); // Wait for reconcile to fire and advance the reconciledBase. - await waitForCondition(() => getReconciledBase(docName) === updatedContent, { - timeoutMs: 8_000, - }); - expect(getReconciledBase(docName)).toBe(updatedContent); + await waitForCondition( + () => server.durabilityState.getReconciledBase(docName) === updatedContent, + { + timeoutMs: 8_000, + }, + ); + expect(server.durabilityState.getReconciledBase(docName)).toBe(updatedContent); // The doc-start `---\n` survived the full chain: cold-load → ytext // → composeAndWriteRawBody → fragment derives via parse → file- @@ -159,14 +161,14 @@ describe('FR-34: serializeDoc returns ytext bytes verbatim', () => { // chain would have canonicalized `---` to `***` somewhere along // the way (specifically: serializeDoc returning serialize(fragment) // bytes for `ours`). - expect(getReconciledBase(docName)).toContain('---\n'); - expect(getReconciledBase(docName)).not.toContain('***\n'); + expect(server.durabilityState.getReconciledBase(docName)).toContain('---\n'); + expect(server.durabilityState.getReconciledBase(docName)).not.toContain('***\n'); conn.disconnect(); } finally { await server.destroy(); } - }); + }, 20_000); // The in-flight ytext-edit visibility case isn't covered here: a // non-paired user-origin Y.Text mutation triggers Observer B Phase 1 @@ -222,22 +224,27 @@ describe('FR-35: setReconciledBase stores raw bytes uniformly across all paths', try { await server.ready; const conn = await server.hocuspocus.openDirectConnection(docName); - await waitForCondition(() => getReconciledBase(docName) === initialContent); + await waitForCondition( + () => server.durabilityState.getReconciledBase(docName) === initialContent, + ); // Trigger reconcile via external change. const updatedContent = '---\n# Title\n\nA __strong__ paragraph.\n\nNew block.\n'; writeFileSync(docPath, updatedContent, 'utf-8'); - await waitForCondition(() => getReconciledBase(docName) === updatedContent, { - timeoutMs: 8_000, - }); + await waitForCondition( + () => server.durabilityState.getReconciledBase(docName) === updatedContent, + { + timeoutMs: 8_000, + }, + ); // After reconcile, reconciledBase advances to the merge output. // Both `__strong__` (source-form) and `---\n` ( // doc-start) survived the full cycle. // these would have been replaced with canonical-form bytes // somewhere along the way. - const finalBase = getReconciledBase(docName); + const finalBase = server.durabilityState.getReconciledBase(docName); expect(finalBase).toBe(updatedContent); expect(finalBase).toContain('---\n'); expect(finalBase).toContain('__strong__'); @@ -253,5 +260,5 @@ describe('FR-35: setReconciledBase stores raw bytes uniformly across all paths', } finally { await server.destroy(); } - }); + }, 20_000); }); diff --git a/packages/server/src/server-factory.test.ts b/packages/server/src/server-factory.test.ts index 34e0b06c..c67da0da 100644 --- a/packages/server/src/server-factory.test.ts +++ b/packages/server/src/server-factory.test.ts @@ -7,6 +7,7 @@ import shellQuote from 'shell-quote'; import simpleGit from 'simple-git'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; import * as Y from 'yjs'; +import { applyExternalChange } from './external-change.ts'; import type { CheckPushPermissionOptions, DetectGhFn, @@ -109,6 +110,78 @@ function captureAllLoggers(): { // ─── Test suite ───────────────────────────────────────────────────────────── +describe('createServer() — document durability state isolation', () => { + test('keeps same-named documents, branch scope, batch state, and disk intake per server', async () => { + const projectA = await mkdtemp(join(tmpdir(), 'ok-durability-a-')); + const projectB = await mkdtemp(join(tmpdir(), 'ok-durability-b-')); + const docName = 'same-doc'; + const diskA = '# Disk A\n'; + const diskB = '# Disk B\n'; + let serverA: ServerInstance | null = null; + let serverB: ServerInstance | null = null; + + try { + writeFileSync(join(projectA, `${docName}.md`), diskA, 'utf-8'); + writeFileSync(join(projectB, `${docName}.md`), diskB, 'utf-8'); + serverA = createServer({ + contentDir: projectA, + projectDir: projectA, + gitEnabled: false, + quiet: true, + }); + serverB = createServer({ + contentDir: projectB, + projectDir: projectB, + gitEnabled: false, + quiet: true, + }); + await Promise.all([serverA.ready, serverB.ready]); + + const [connectionA, connectionB] = await Promise.all([ + serverA.hocuspocus.openDirectConnection(docName), + serverB.hocuspocus.openDirectConnection(docName), + ]); + + expect(serverA.durabilityState).not.toBe(serverB.durabilityState); + expect(serverA.durabilityState.getReconciledBase(docName)).toBe(diskA); + expect(serverB.durabilityState.getReconciledBase(docName)).toBe(diskB); + expect(serverA.hocuspocus.documents.get(docName)?.getText('source').toString()).toBe(diskA); + expect(serverB.hocuspocus.documents.get(docName)?.getText('source').toString()).toBe(diskB); + + serverA.durabilityState.switchReconciledBaseScope('feature-a'); + serverA.durabilityState.setReconciledBase(docName, 'A feature'); + serverA.durabilityState.setBatchInProgress(true); + serverA.durabilityState.beginInFlightFlush(docName, 'A flush'); + + expect(serverA.durabilityState.getReconciledBase(docName)).toBe('A feature'); + expect(serverB.durabilityState.getActiveBranch()).toBe('main'); + expect(serverB.durabilityState.getReconciledBase(docName)).toBe(diskB); + expect(serverA.durabilityState.isBatchInProgress()).toBe(true); + expect(serverB.durabilityState.isBatchInProgress()).toBe(false); + expect(serverB.durabilityState.peekInFlightFlush(docName)).toBeUndefined(); + + const externalA = '# External A\n'; + applyExternalChange(serverA.durabilityState, serverA.hocuspocus, docName, externalA); + + expect(serverA.hocuspocus.documents.get(docName)?.getText('source').toString()).toBe( + externalA, + ); + expect(serverA.durabilityState.getReconciledBase(docName)).toBe(externalA); + expect(serverB.hocuspocus.documents.get(docName)?.getText('source').toString()).toBe(diskB); + expect(serverB.durabilityState.getReconciledBase(docName)).toBe(diskB); + + await Promise.all([connectionA.disconnect(), connectionB.disconnect()]); + } finally { + await serverA?.destroy(); + await serverB?.destroy(); + await Promise.all([ + rm(projectA, { recursive: true, force: true }), + rm(projectB, { recursive: true, force: true }), + ]); + } + }); +}); + describe('createServer().destroy() — graceful shutdown flush', () => { let tmpDir: string; let logCapture: ReturnType; diff --git a/packages/server/src/server-factory.ts b/packages/server/src/server-factory.ts index b3f7b4a3..72cf2543 100644 --- a/packages/server/src/server-factory.ts +++ b/packages/server/src/server-factory.ts @@ -75,9 +75,12 @@ import { isDocInConflict } from './conflict-errors.ts'; import { createConflictLifecycleSeedExtension } from './conflict-lifecycle-seed.ts'; import { resolveProjectTemplates } from './content/templates-resolver.ts'; import { type ContentFilter, createContentFilter } from './content-filter.ts'; +import { isWithinContentDir, safeContentPath } from './content-path.ts'; import { dropPendingDocs, recordContributor } from './contributor-tracker.ts'; +import { applyDiskContentToDoc } from './disk-content-intake.ts'; import { docNameToRelativePath, getDocExtension, stripDocExtension } from './doc-extensions.ts'; import { runDocLineageGuard } from './doc-lineage-guard.ts'; +import { DocumentDurabilityState } from './document-durability-state.ts'; import { DEFAULT_EMBEDDINGS_DIMENSIONS, type Embedder, @@ -90,7 +93,6 @@ import { secretsFilePath, } from './embeddings/index.ts'; import { - applyDiskContentToDoc, applyExternalChange, FILE_WATCHER_ORIGIN, serializeYDocSource, @@ -140,19 +142,7 @@ import { } from './metrics.ts'; import { destroyParsePool } from './parse-pool.ts'; import { isWithinDir, toPosix } from './path-utils.ts'; -import { - createPersistenceExtension, - deleteReconciledBase, - getActiveBranch, - getReconciledBase, - isBatchInProgress, - isWithinContentDir, - type PersistenceOptions, - safeContentPath, - setBatchInProgress, - setReconciledBase, - switchReconciledBaseScope, -} from './persistence.ts'; +import { createPersistenceExtension, type PersistenceOptions } from './persistence.ts'; import { createPersistenceStalenessWatchdog, type StalenessWatchdogHandle, @@ -381,6 +371,7 @@ export interface ServerInstance { * server Y.Doc. Part of the CRDT server-restart recovery defense. */ readonly serverInstanceId: string; + readonly durabilityState: DocumentDurabilityState; destroy: () => Promise; /** Resolves when async init (shadow repo, file watcher subscription) is complete. */ ready: Promise; @@ -588,6 +579,16 @@ export function createServer(options: ServerOptions): ServerInstance { } = options; const log = getLogger('server'); + const durabilityState = new DocumentDurabilityState(); + const getActiveBranch = () => durabilityState.getActiveBranch(); + const getReconciledBase = (docName: string) => durabilityState.getReconciledBase(docName); + const setReconciledBase = (docName: string, content: string) => + durabilityState.setReconciledBase(docName, content); + const deleteReconciledBase = (docName: string) => durabilityState.deleteReconciledBase(docName); + const switchReconciledBaseScope = (branch: string) => + durabilityState.switchReconciledBaseScope(branch); + const setBatchInProgress = (value: boolean) => durabilityState.setBatchInProgress(value); + const isBatchInProgress = () => durabilityState.isBatchInProgress(); function readProjectAttachmentFolderPath(): string { const project = readConfigSafely({ @@ -1269,7 +1270,7 @@ export function createServer(options: ServerOptions): ServerInstance { mdManager: options.mdManager, }; - persistence = createPersistenceExtension(persistenceOpts); + persistence = createPersistenceExtension({ ...persistenceOpts, durabilityState }); hocuspocus = new Hocuspocus({ quiet, @@ -1318,6 +1319,9 @@ export function createServer(options: ServerOptions): ServerInstance { stalenessWatchdog = createPersistenceStalenessWatchdog({ getLoadedDocuments: () => hp.documents, forceStore: (document, documentName) => persistence.forceStore(document, documentName), + getBase: (documentName) => durabilityState.getReconciledBase(documentName), + isBatchActive: () => durabilityState.isBatchInProgress(), + peekInFlight: (documentName) => durabilityState.peekInFlightFlush(documentName), // Realpath symlink-escape gate + open-byte-limit cap, matching the // load-path read discipline. Not atomic — three syscalls — but // ENOENT from any of them maps to `null` (out-of-band delete) @@ -1746,6 +1750,7 @@ export function createServer(options: ServerOptions): ServerInstance { const apiExtension = createApiExtension({ hocuspocus, + durabilityState, sessionManager, contentDir, contentFilter, @@ -1763,9 +1768,6 @@ export function createServer(options: ServerOptions): ServerInstance { shadowRef, flushGitCommit: () => persistence.flushPendingGitCommit(), flushContributors: () => persistence.flushContributors(), - takeStoreFailure: (docName: string) => persistence.takeStoreFailure(docName), - takeStoreDivergence: (docName: string) => persistence.takeStoreDivergence(docName), - markAgentWriteStore: (docName: string) => persistence.markAgentWriteStore(docName), getCurrentBranch: () => headWatcher?.getLastKnownBranch() ?? null, // CC1 broadcaster is initialized after persistence but captured by // closure reference (same pattern as `onAgentCommit` + `onDiskFlush` @@ -1920,7 +1922,7 @@ export function createServer(options: ServerOptions): ServerInstance { /** Apply markdown content to Y.Doc — delegates to the shared throwing helper. */ const applyToDoc = (docName: string, content: string): void => - applyExternalChange(hocuspocus, docName, content, resolveEmbed, resolveSize); + applyExternalChange(durabilityState, hocuspocus, docName, content, resolveEmbed, resolveSize); /** * Clear the conflict status set by `case 'conflict'` once a subsequent @@ -3482,13 +3484,8 @@ export function createServer(options: ServerOptions): ServerInstance { degraded.push('ignore-files-watcher'); } - // Reset branch-scoped state to match THIS project's current HEAD before - // anything reads/writes it. `persistence.activeBranch` and the - // `BacklinkIndex.activeBranch` are mutable state; in single-process test - // runners these leak across test files, so a prior test that - // triggered `switchReconciledBaseScope` leaves state at the wrong branch - // for the next server's reads. Detecting the actual HEAD here and - // normalizing both scopes in lock-step closes the leak. + // Align this server's branch-scoped durability state and backlink index + // with the project's current HEAD before either is read or written. const startupBranch = readProjectHeadState(projectDir).branch ?? 'main'; switchReconciledBaseScope(startupBranch); backlinkIndex.switchBranch(startupBranch); @@ -4331,6 +4328,7 @@ export function createServer(options: ServerOptions): ServerInstance { return { hocuspocus, + durabilityState, sessionManager, cc1Broadcaster, agentFocusBroadcaster, diff --git a/packages/server/src/share/construct-url.test.ts b/packages/server/src/share/construct-url.test.ts index 42cab171..521eb55b 100644 --- a/packages/server/src/share/construct-url.test.ts +++ b/packages/server/src/share/construct-url.test.ts @@ -42,7 +42,7 @@ async function bootRig( const { Hocuspocus } = await import('@hocuspocus/server'); const { AgentSessionManager } = await import('../agent-sessions.ts'); - const { createApiExtension } = await import('../api-extension.ts'); + const { createApiExtension } = await import('../api-extension.test-helper.ts'); const hocuspocus = new Hocuspocus({ quiet: true }); const sessionManager = new AgentSessionManager(hocuspocus); diff --git a/packages/server/src/share/endpoint-http.test-helper.ts b/packages/server/src/share/endpoint-http.test-helper.ts index ba7dc9fc..455bdbe8 100644 --- a/packages/server/src/share/endpoint-http.test-helper.ts +++ b/packages/server/src/share/endpoint-http.test-helper.ts @@ -34,7 +34,7 @@ export async function bootEndpointServer(opts: { const { Hocuspocus } = await import('@hocuspocus/server'); const { AgentSessionManager } = await import('../agent-sessions.ts'); - const { createApiExtension } = await import('../api-extension.ts'); + const { createApiExtension } = await import('../api-extension.test-helper.ts'); const hocuspocus = new Hocuspocus({ quiet: true }); const sessionManager = new AgentSessionManager(hocuspocus); diff --git a/packages/server/src/share/publish.integration.test.ts b/packages/server/src/share/publish.integration.test.ts index 3d3386ad..1033a944 100644 --- a/packages/server/src/share/publish.integration.test.ts +++ b/packages/server/src/share/publish.integration.test.ts @@ -72,7 +72,7 @@ async function bootRig(options: RigOptions): Promise { const { Hocuspocus } = await import('@hocuspocus/server'); const { AgentSessionManager } = await import('../agent-sessions.ts'); - const { createApiExtension } = await import('../api-extension.ts'); + const { createApiExtension } = await import('../api-extension.test-helper.ts'); const hocuspocus = new Hocuspocus({ quiet: true }); const sessionManager = new AgentSessionManager(hocuspocus); diff --git a/packages/server/src/suggest-links.test.ts b/packages/server/src/suggest-links.test.ts index 8e4df12e..ffc0323a 100644 --- a/packages/server/src/suggest-links.test.ts +++ b/packages/server/src/suggest-links.test.ts @@ -5,6 +5,7 @@ import { Hocuspocus } from '@hocuspocus/server'; import { afterEach, beforeEach, describe, expect, test } from 'vitest'; import type * as Y from 'yjs'; import { AGENT_WRITE_ORIGIN, applyAgentMarkdownWrite } from './agent-sessions.ts'; +import { DocumentDurabilityState } from './document-durability-state.ts'; import { applyExternalChange } from './external-change.ts'; import type { FileIndexEntry } from './file-watcher.ts'; import { installTestLoggers, loggerFactory } from './logger.ts'; @@ -250,6 +251,7 @@ describe('suggestLinks', () => { const contentDir = join(projectDir, 'content'); mkdirSync(contentDir, { recursive: true }); const hocuspocus = new Hocuspocus({ quiet: true }); + const durabilityState = new DocumentDurabilityState(); let conn: Conn | null = null; try { @@ -261,7 +263,12 @@ describe('suggestLinks', () => { // verbatim and fragment derives via parse. conn = await hocuspocus.openDirectConnection('notes'); const doc = getDoc(conn); - applyExternalChange(hocuspocus, 'notes', '---\n# Notes\n\nProject Alpha discussion.\n'); + applyExternalChange( + durabilityState, + hocuspocus, + 'notes', + '---\n# Notes\n\nProject Alpha discussion.\n', + ); // ytext byte-equal: user form preserved. const yText = doc.getText('source').toString();