diff --git a/.changeset/docs-fetch-merged-head.md b/.changeset/docs-fetch-merged-head.md new file mode 100644 index 0000000000..ab15d57421 --- /dev/null +++ b/.changeset/docs-fetch-merged-head.md @@ -0,0 +1,20 @@ +--- +id: docs-fetch-merged-head +tasks: [T13842] +kind: fix +summary: docs fetch now returns the merged head after concurrent edits, not the last writer's raw input +--- + +`docs update` wrote the doc row (what `docs fetch` and `docs view` read) +from the incoming bytes, then projected the edit through the section CRDT. +After a stale-base merge the conduit head held both writers' edits, but the +row held only the last writer's input, so readers lost the other edit even +when the merge itself was correct. The merge is now previewed first (the +real projection, rolled back), the row is written from the resulting head +text (the merged text when the write advances the head, the current head's +text when it loses the head race), and the projection is committed only +after the row write succeeds, so a failed blob or row write never leaves the +head advanced over an old row. A failed preview writes the incoming bytes and +leaves the head unchanged; a failed commit after the row write is reported +with the head uid (`W_DOC_PROJECTION_FAILED`). The body schema is checked on +the merged text. diff --git a/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts b/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts index f159c3142d..06cbfa406a 100644 --- a/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts +++ b/packages/core/src/docs/__tests__/docs-update-concurrency.test.ts @@ -57,11 +57,38 @@ */ import { createHash } from 'node:crypto'; -import { readFileSync, writeFileSync } from 'node:fs'; +import { mkdirSync, readFileSync, writeFileSync } from 'node:fs'; import { mkdtemp, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; + +/** + * T13842 test seams over the REAL section write path: `forcedHlc` runs the + * preview and the projection with an explicit HLC (a minted one always + * advances the head); `failPreview` / `failCommit` make that step throw. + */ +let forcedHlc: string | undefined; +let failPreview = false; +let failCommit = false; +vi.mock('../section-write-path.js', async (importOriginal) => { + const actual = await importOriginal(); + type Input = Parameters[1]; + const withHlc = (input: Input): Input => + forcedHlc !== undefined ? { ...input, hlc: forcedHlc } : input; + return { + ...actual, + previewDocUpdateThroughSections: (root: string, input: Input) => { + if (failPreview) throw new Error('injected preview failure'); + return actual.previewDocUpdateThroughSections(root, withHlc(input)); + }, + projectDocUpdateThroughSections: (root: string, input: Input) => { + if (failCommit) throw new Error('injected commit failure'); + return actual.projectDocUpdateThroughSections(root, withHlc(input)); + }, + }; +}); + import { getConduitAttachmentVersion, getConduitDocHead, @@ -680,6 +707,151 @@ async function seedLegacyDoc(root: string, slug: string): Promise { .run(); } +describe("docs fetch serves the merged head, not the last writer's raw input (T13842)", () => { + const SEED = '# Plan\n\npre.\n\n## Alpha\n\nalpha body\n\n## Beta\n\nbeta body'; + + async function withStore(fn: (root: string) => Promise): Promise { + const root = await mkdtemp(join(tmpdir(), 'cleo-docs-fetch-head-')); + process.env['CLEO_DIR'] = join(root, '.cleo'); + closeDb(); + try { + await seedLegacyDoc(root, SLUG); + expect((await updateDocBySlug(root, { slug: SLUG, content: SEED })).ok).toBe(true); + await fn(root); + } finally { + forcedHlc = undefined; + failPreview = false; + failCommit = false; + closeDb(); + delete process.env['CLEO_DIR']; + await rm(root, { recursive: true, force: true }); + } + } + + async function fetchedBody(root: string): Promise { + const model = new DocsReadModel({ projectRoot: root }); + const doc = await model.resolveBySlug(SLUG); + return doc === null ? null : model.fetchContent(doc); + } + + async function headText(root: string): Promise { + const docUid = (await resolveConduitDocUid(root, SLUG)) as string; + const head = await getConduitDocHead(root, docUid); + const version = await getConduitAttachmentVersion(root, head?.headVersionUid ?? ''); + return readFileSync( + join(root, '.cleo', 'attachments', 'sha256', version?.storageKey ?? ''), + 'utf-8', + ); + } + + it("a stale-base append merge: the fetched body carries both writers' lines", async () => { + await withStore(async (root) => { + const docUid = (await resolveConduitDocUid(root, SLUG)) as string; + const base = (await getConduitDocHead(root, docUid))?.headVersionUid; + const a = await updateDocBySlug(root, { + slug: SLUG, + content: SEED.replace('alpha body', 'alpha body\n\nA line'), + }); + expect(a.ok).toBe(true); + const b = await updateDocBySlug(root, { + slug: SLUG, + content: SEED.replace('beta body', 'beta body\n\nB line'), + baseVersionUid: base, + }); + expect(b.ok && b.result.staleBaseMerged).toBe(true); + + const body = await fetchedBody(root); + expect(body).toContain('A line'); + expect(body).toContain('B line'); + expect(body).toBe(await headText(root)); + }); + }); + + it('a write that loses the head race: the fetched body is the head, not the losing input', async () => { + await withStore(async (root) => { + const winner = SEED.replace('alpha body', 'alpha body\n\nwinner line'); + forcedHlc = '9999999999999-000000-f4d5b7e5-4a36-4c39-9f8b-8e2d6a1f2c40'; + expect((await updateDocBySlug(root, { slug: SLUG, content: winner })).ok).toBe(true); + // A later write with a lower HLC loses the head race; its version stays in history. + forcedHlc = '0000000000001-000000-f4d5b7e5-4a36-4c39-9f8b-8e2d6a1f2c40'; + const loser = await updateDocBySlug(root, { + slug: SLUG, + content: SEED.replace('beta body', 'beta body\n\nloser line'), + }); + expect(loser.ok).toBe(true); + + const head = await headText(root); + expect(head).toContain('winner line'); + expect(await fetchedBody(root)).toBe(head); + }); + }); + + /** Land A's Alpha append on the seed head and return the base B fetched before it. */ + async function landA(root: string): Promise<{ docUid: string; base: string; aText: string }> { + const docUid = (await resolveConduitDocUid(root, SLUG)) as string; + const base = (await getConduitDocHead(root, docUid))?.headVersionUid as string; + const aText = SEED.replace('alpha body', 'alpha body\n\nA line'); + expect((await updateDocBySlug(root, { slug: SLUG, content: aText })).ok).toBe(true); + return { docUid, base, aText }; + } + + const B_TEXT = SEED.replace('beta body', 'beta body\n\nB line'); + + it('a blob write that fails after the merge leaves the head and history untouched', async () => { + await withStore(async (root) => { + const { docUid, base, aText } = await landA(root); + const headBefore = (await getConduitDocHead(root, docUid))?.headVersionUid; + const versionsBefore = (await listConduitAttachmentVersions(root, docUid)).length; + // Block the row blob of the merged text: a directory where its file goes. + const merged = aText.replace('beta body', 'beta body\n\nB line'); + const sha = createHash('sha256').update(merged, 'utf-8').digest('hex'); + mkdirSync( + join(root, '.cleo', 'attachments', 'sha256', sha.slice(0, 2), `${sha.slice(2)}.txt`), + { + recursive: true, + }, + ); + + const b = await updateDocBySlug(root, { slug: SLUG, content: B_TEXT, baseVersionUid: base }); + + expect(b.ok).toBe(false); + if (b.ok) return; + expect(b.error.code).toBe('E_FILE_ERROR'); + expect((await getConduitDocHead(root, docUid))?.headVersionUid).toBe(headBefore); + expect((await listConduitAttachmentVersions(root, docUid)).length).toBe(versionsBefore); + expect(await fetchedBody(root)).toBe(aText); + }); + }); + + it('a failed merge preview writes the incoming bytes and leaves the head unchanged', async () => { + await withStore(async (root) => { + const { docUid, base } = await landA(root); + const headBefore = (await getConduitDocHead(root, docUid))?.headVersionUid; + failPreview = true; + + const b = await updateDocBySlug(root, { slug: SLUG, content: B_TEXT, baseVersionUid: base }); + + expect(b.ok).toBe(true); + expect(await fetchedBody(root)).toBe(B_TEXT); + expect((await getConduitDocHead(root, docUid))?.headVersionUid).toBe(headBefore); + }); + }); + + it('a failed projection commit after the row write keeps the merged row and the old head', async () => { + await withStore(async (root) => { + const { docUid, base, aText } = await landA(root); + const headBefore = (await getConduitDocHead(root, docUid))?.headVersionUid; + failCommit = true; + + const b = await updateDocBySlug(root, { slug: SLUG, content: B_TEXT, baseVersionUid: base }); + + expect(b.ok).toBe(true); + expect(await fetchedBody(root)).toBe(aText.replace('beta body', 'beta body\n\nB line')); + expect((await getConduitDocHead(root, docUid))?.headVersionUid).toBe(headBefore); + }); + }); +}); + describe('first update on two devices adopts one doc uid (T13586)', () => { it('two stores holding the same synced legacy row converge on one conduit uid through updateDocBySlug', async () => { const legacyBirth = '2026-10-01T12:00:00.000Z'; diff --git a/packages/core/src/docs/docs-update.ts b/packages/core/src/docs/docs-update.ts index c473405b4b..ff5bd276c4 100644 --- a/packages/core/src/docs/docs-update.ts +++ b/packages/core/src/docs/docs-update.ts @@ -43,6 +43,7 @@ import { attachmentRefs, docsAttachments } from '../store/tasks-schema.js'; import { assertTwinCollapseWritable } from '../store/twin-collapse.js'; import { writeFileAtomic } from '../tools/fs.js'; import { deriveDocLinks, linksJsonOrNull } from './derive-links.js'; +import type { DocUpdatePreview, ProjectDocUpdateInput } from './section-write-path.js'; import { validateDocBody } from './validate-body.js'; import { getCanonicalCleoVersion } from './version-ssot.js'; @@ -588,6 +589,100 @@ export async function updateDocBySlug( const previousAttachmentId = oldRow.id; const previousSha256 = oldRow.sha256; + const incomingSha256 = buf === null ? previousSha256 : sha256Of(buf); + + // T13842: the row written below is what `docs fetch` reads, so it must hold + // the HEAD text, not the incoming bytes — after a stale-base merge the head + // carries a concurrent writer's edit the incoming bytes lack. The merge is + // PREVIEWED here (the real projection, rolled back), the row is written from + // its result, and the projection is committed only after the row write + // succeeded, so a failed blob or row write never leaves the head advanced + // over an old row. Text bodies only (T13688); binary files stay whole-doc. + // A failed preview writes the incoming bytes and leaves the head unchanged. + let projectionInput: ProjectDocUpdateInput | null = null; + if ( + text !== null && + buf !== null && + params.dryRun !== true && + incomingSha256 !== previousSha256 + ) { + try { + const { previewDocUpdateThroughSections } = await import('./section-write-path.js'); + // T13734: link the pre-update bytes to the doc before the first projection. + const priorVersionUid = await recordPriorVersionOnAdopt( + projectRoot, + slug, + oldRow, + attachedBy, + ); + // T13356 AC4: the real session id drives lease acquire/renew, so a + // lease dies honestly with its session (TTL bound, holder-only renew). + // Gate 16: the bound session is resolved via resolveBoundSession — + // never a bare getActiveSession(). + const boundSession = await resolveBoundSession(projectRoot); + const input: ProjectDocUpdateInput = { + slug, + authorAgentId: attachedBy, + content: text, + ...(boundSession !== null ? { sessionId: boundSession.session.id } : {}), + ...(params.message !== undefined ? { changeSummary: params.message } : {}), + ...(priorVersionUid !== null ? { parentVersionUid: priorVersionUid } : {}), + // T13674: the declared base replaces the previous head as the + // single-writer/concurrent discriminator, so a stale base merges + // (append) or conflicts (non-append) instead of re-seeding over a + // concurrent writer's landed edit. + ...(baseSections !== undefined ? { baseSections } : {}), + }; + // T13881: the preview ROLLS BACK on the process-wide shared native handle, + // so a write another in-process writer lands on that handle during its + // awaits would be rolled back with it. Two guards: (1) every tracked + // background op (graph nodes, hook dispatches, mirrors) settles first, + // and the preview is skipped if one is still running; (2) the preview + // runs on the handle's writer queue (scheduleTaskBackground), so accessor + // transactions and committed background effects queued meanwhile start + // only after its rollback. + const { settleBackgroundOps } = await import('../store/background-ops.js'); + const pending = await settleBackgroundOps(PREVIEW_SETTLE_BUDGET_MS); + if (pending > 0) { + // @sync-invariant none:input-shape caught just below: the preview is skipped and the incoming bytes are written; nothing is rejected + throw new Error(`${pending} background write(s) still running; the preview is skipped`); + } + const native = getNativeTasksDb(projectRoot); + if (!native) { + // @sync-invariant none:input-shape caught just below: without the store handle the preview is skipped and the incoming bytes are written; nothing is rejected + throw new Error('tasks database is not initialised'); + } + const { scheduleTaskBackground } = await import('../store/sqlite-data-accessor.js'); + const queued: { preview?: DocUpdatePreview } = {}; + await scheduleTaskBackground(native, async () => { + queued.preview = await previewDocUpdateThroughSections(projectRoot, input); + }); + const preview = queued.preview; + if (preview === undefined) { + // @sync-invariant none:input-shape caught just below: an empty preview skips the projection and the incoming bytes are written; nothing is rejected + throw new Error('the merge preview returned nothing'); + } + // The head is the merged text when this write would advance it; when it + // loses the head race (an explicit lower HLC) the head stays where it is. + const headText = preview.wouldAdvance + ? preview.mergedText + : await readCurrentHeadText(projectRoot, slug); + if (headText === null) { + // @sync-invariant none:input-shape caught just below: an unreadable head skips the projection and the incoming bytes are written; nothing is rejected + throw new Error('the current conduit head text is unreadable'); + } + if (headText !== text) buf = Buffer.from(headText, 'utf-8'); + projectionInput = input; + } catch (previewErr) { + pushWarning({ + // @sync-invariant none:input-shape the warning only reports that the section projection was skipped; the incoming bytes are written as before and nothing is rejected + code: 'W_DOC_PROJECTION_FAILED', + message: + `section merge preview failed for slug '${slug}' — the incoming bytes are written and the conduit head is unchanged: ` + + (previewErr instanceof Error ? previewErr.message : String(previewErr)), + }); + } + } const newSha256 = buf === null ? previousSha256 : sha256Of(buf); const wouldChange = newSha256 !== previousSha256 || oldRow.lifecycleStatus !== status; @@ -646,7 +741,7 @@ export async function updateDocBySlug( // update if the caller asked for one, and still write an audit entry so // operators can see the noop attempt. A lifecycle-only update (no new // bytes, T12654) always takes this path. - if (buf === null || newSha256 === previousSha256) { + if (buf === null || incomingSha256 === previousSha256) { // Only touch lifecycle_status if it actually changes. if (oldRow.lifecycleStatus !== status) { await db @@ -948,45 +1043,29 @@ export async function updateDocBySlug( ); } - // T13355: project the update through the section CRDT merge layer (spec - // `cleodocs-rearch-design` §2 option b). Text bodies only — inline content - // or a UTF-8 text file (T13688); binary files stay on the whole-doc path. - // Best-effort: a projection failure must not fail the completed update — - // the whole-row write above is the fallback when no section state exists. + // T13842: commit the previewed projection now that the row holds its text. // T13604: a recorded conflict is threaded into the result envelope and a // stderr warning names the sections — the update reports changed:true, so // without this the caller would never learn their edit is not in the head. let sectionConflicts: DocsSectionConflict[] | undefined; let leaseNotices: DocsSectionLeaseNotice[] | undefined; let staleBaseMerged = false; - if (text !== null) { + if (projectionInput !== null) { try { const { projectDocUpdateThroughSections } = await import('./section-write-path.js'); - // T13734: link the pre-update bytes to the doc before the first projection. - const priorVersionUid = await recordPriorVersionOnAdopt( - projectRoot, - slug, - oldRow, - attachedBy, - ); - // T13356 AC4: the real session id drives lease acquire/renew, so a - // lease dies honestly with its session (TTL bound, holder-only renew). - // Gate 16: the bound session is resolved via resolveBoundSession — - // never a bare getActiveSession(). - const boundSession = await resolveBoundSession(projectRoot); - const projection = await projectDocUpdateThroughSections(projectRoot, { - slug, - authorAgentId: attachedBy, - content: text, - ...(boundSession !== null ? { sessionId: boundSession.session.id } : {}), - ...(params.message !== undefined ? { changeSummary: params.message } : {}), - ...(priorVersionUid !== null ? { parentVersionUid: priorVersionUid } : {}), - // T13674: the declared base replaces the previous head as the - // single-writer/concurrent discriminator, so a stale base merges - // (append) or conflicts (non-append) instead of re-seeding over a - // concurrent writer's landed edit. - ...(baseSections !== undefined ? { baseSections } : {}), - }); + const projection = await projectDocUpdateThroughSections(projectRoot, projectionInput); + const committedText = projection.head.advanced + ? projection.mergedText + : await readVersionText(projectRoot, projection.head.head.headVersionUid); + if (committedText !== buf.toString('utf-8')) { + pushWarning({ + // @sync-invariant none:input-shape the warning only reports that another writer landed between the preview and the commit; both versions are kept and nothing is rejected + code: 'W_DOC_HEAD_DIVERGED', + message: + `slug '${slug}': another writer landed between the merge preview and its commit — the doc row holds the previewed text, ` + + `the conduit head is ${projection.head.head.headVersionUid}. Re-fetch (\`cleo docs fetch ${slug}\`) and re-apply if needed.`, + }); + } // T13674: the declared base was superseded AND at least one edited // section composed into the head — its text is the merge of the // caller's edit and a concurrent writer's. A section that only @@ -1045,9 +1124,15 @@ export async function updateDocBySlug( }); } } catch (projectionErr) { - process.emitWarning( - `docs section CRDT projection failed after update of slug '${slug}': ${projectionErr instanceof Error ? projectionErr.message : String(projectionErr)}`, - ); + const { readDocHeadVersionUid: readHead } = await import('./fetch-result.js'); + const head = await readHead(projectRoot, slug); + pushWarning({ + // @sync-invariant none:input-shape the warning only reports a failed projection commit after the row write; the row and the version history are intact and nothing is rejected + code: 'W_DOC_PROJECTION_FAILED', + message: + `slug '${slug}': the doc row was written from the previewed merge, but committing the section projection failed ` + + `(${projectionErr instanceof Error ? projectionErr.message : String(projectionErr)}); the conduit head stays at ${head ?? 'none'} — re-run the update to land it`, + }); } } @@ -1078,6 +1163,39 @@ export async function updateDocBySlug( }; } +/** + * How long a docs update waits for tracked background writes to settle before + * it previews a merge (T13881). A write still running after this is not + * rolled back by the preview: the preview is skipped instead. + */ +const PREVIEW_SETTLE_BUDGET_MS = 5_000; + +/** + * A conduit version's text, read from its content-addressed blob (T13842). + * `null` when the version or its bytes cannot be read; the caller then keeps + * the incoming bytes. + */ +async function readVersionText(projectRoot: string, versionUid: string): Promise { + const { getConduitAttachmentVersion } = await import('../store/conduit-doc-history.js'); + const version = await getConduitAttachmentVersion(projectRoot, versionUid); + if (!version?.storageKey) return null; + const { readFile } = await import('node:fs/promises'); + return readFile( + join(getCleoDirAbsolute(projectRoot), 'attachments', 'sha256', version.storageKey), + 'utf-8', + ).catch(() => null); +} + +/** The current conduit head's text for a slug (T13842), or `null` when it has none or it is unreadable. */ +async function readCurrentHeadText(projectRoot: string, slug: string): Promise { + const { getConduitDocHead, resolveConduitDocUid } = await import( + '../store/conduit-doc-history.js' + ); + const docUid = await resolveConduitDocUid(projectRoot, slug); + const head = docUid !== null ? await getConduitDocHead(projectRoot, docUid) : null; + return head !== null ? readVersionText(projectRoot, head.headVersionUid) : null; +} + /** The bytes as text when they are valid UTF-8 without NUL bytes; else null. */ function utf8TextOf(buf: Buffer): string | null { if (buf.includes(0)) return null;