diff --git a/.changeset/t13867-journal-unread-stale.md b/.changeset/t13867-journal-unread-stale.md new file mode 100644 index 0000000000..e0019a81ac --- /dev/null +++ b/.changeset/t13867-journal-unread-stale.md @@ -0,0 +1,24 @@ +--- +id: t13867-journal-unread-stale +tasks: [T13867] +kind: fix +summary: a store that reads only snapshots is no longer reported up to date while the change journal has moved past the newest snapshot +--- + +A store that does not read the change journal (it restored a snapshot and +never ran `cleo sync enable push`) holds only the newest snapshot. When other +devices then wrote through the journal, `cleo cloud pull` still said +`up-to-date` and `cleo cloud verify` said `match`, while every journal change +was missing. Now that pull refuses with `E_NEXUS_VAULT_BEHIND`, verify says +`behind`, and a restore warns (`W_NEXUS_VAULT_JOURNAL_UNREAD`), each naming +how many journal segments are missing and the way forward: `cleo sync enable +push`, or a vault push from a device that writes the journal. + +On a journal stream, `cleo cloud verify` no longer reports `ahead` only +because a table that snapshots alone carry (the file inventory, other +databases, the device and project registries) differs on this machine. Those +tables are still listed, with a `W_NEXUS_VAULT_SNAPSHOT_ONLY` warning. + +Every request now sends `x-cleo-stream-reader: 3`, so a Cleo Nexus server can +refuse a journal stream to a client that reads only snapshots. Released +v2026.10.19 does not send it. diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts index 5390e4430b..1c23509e4f 100644 --- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-vault.test.ts @@ -80,7 +80,13 @@ import { recordBareAccounts, setSyncFlag, } from '../../store/sync/flags.js'; -import { cutGenesis, genesisCutOf, genesisPending } from '../../store/sync/genesis.js'; +import { + cutGenesis, + GENESIS_CUT_KEY_PREFIX, + genesisCutOf, + genesisPending, + readGenesisCut, +} from '../../store/sync/genesis.js'; import { readStreamCursor } from '../../store/sync/pull.js'; import { activeReplica, @@ -109,7 +115,7 @@ import { uuidv7, verifyEd25519, } from '../crypto.js'; -import { type FetchLike, Http, NexusError } from '../http.js'; +import { type FetchLike, Http, NEXUS_STREAM_READER_HEADER, NexusError } from '../http.js'; import { cursorFromCheckpoint, Journal, @@ -423,6 +429,8 @@ class FakeNexus { beforeSegment: { deviceId: string; run: () => Promise } | null = null; /** Codes the next checkpoint creates are refused with, in order. */ refuseCheckpoint: string[] = []; + /** As Cleo Nexus with T13876: a v3 stream's head and checkpoint reads need the stream-reader header. */ + streamReaderRequired = false; /** Runs once before the next checkpoint create by `deviceId` (simulates a concurrent author). */ beforeCheckpoint: { deviceId: string; run: () => Promise } | null = null; /** Runs once after the server stored a checkpoint, before the response reaches the client. */ @@ -502,6 +510,19 @@ class FakeNexus { const token = (new Headers(init?.headers).get('authorization') ?? '').replace(/^Bearer /, ''); const device = [...this.devices.values()].find((d) => d.token === token); if (!device) throw new ApiFail(401, 'E_UNAUTHENTICATED'); + if (this.streamReaderRequired && method === 'GET') { + const sm = url.pathname.match(/^\/v1\/streams\/([^/]+)(\/checkpoints)?$/); + const st = sm ? this.streams.get(decodeURIComponent(sm[1] ?? '')) : undefined; + const headCp = st?.checkpoints.find((c) => c.checkpointId === st.headCheckpointId); + if ( + headCp && + manifestVersion(headCp.manifest) === 3 && + new Headers(init?.headers).get(NEXUS_STREAM_READER_HEADER.name) !== + NEXUS_STREAM_READER_HEADER.value + ) { + throw new ApiFail(409, 'E_STREAM_VERSION'); + } + } const body = typeof init?.body === 'string' ? (JSON.parse(init.body) as Record) : {}; if (method !== 'GET') this.writes.push(`${method} ${url.pathname}`); @@ -6163,6 +6184,67 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { return { a, aDb, genesis, b, bDb }; } + it('T13867: a store that reads snapshots only is behind once the journal moves past the head: pull refuses, verify says behind, never up-to-date or match', async () => { + fake.streamReaderRequired = true; + const { a, aDb } = await journalDevices(false); + // B restores the genesis and never joins the journal: it reads snapshots only. + const { b, bDb } = await restoredJoiner(); + // The restored bundle carries no genesis cut of A's: B does not pass for a journal reader. + expect(await on(b, () => readGenesisCut(bDb, STREAM))).toBeUndefined(); + expect((await on(b, () => restoreNexusVault(vopts(b, { mode: 'pull' })))).status).toBe( + 'up-to-date', + ); + expect((await on(b, () => verifyNexusVault(vopts(b)))).verdict).toBe('match'); + // A writes through the journal: a segment past the head checkpoint. + await on(a, async () => (await storeOf(aDb)).exec(insertTask('T3', 'from A'))); + await sync(a); + const pulled = await failure(on(b, () => restoreNexusVault(vopts(b, { mode: 'pull' })))); + expect(pulled.code).toBe('E_NEXUS_VAULT_BEHIND'); + expect(pulled.message).toContain('reads snapshots only'); + expect(pulled.fix).toContain('cleo sync enable push'); + const verified = await on(b, () => verifyNexusVault(vopts(b))); + expect(verified.verdict).toBe('behind'); + expect(verified.remedy).toContain('cleo sync enable push'); + // A genesis cut alone (as a restored bundle could carry A's) does not make B a journal reader. + await on(b, async () => { + const db = await storeOf(bDb); + db.exec( + 'CREATE TABLE IF NOT EXISTS _sync_meta (key TEXT PRIMARY KEY, value TEXT, updated_at TEXT)', + ); + db.prepare( + 'INSERT INTO _sync_meta (key, value, updated_at) VALUES (?, ?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value', + ).run(`${GENESIS_CUT_KEY_PREFIX}${STREAM}`, '0', '2026-10-17'); + }); + expect(await on(b, () => readGenesisCut(bDb, STREAM))).toBe(0); + const cutOnly = await failure(on(b, () => restoreNexusVault(vopts(b, { mode: 'pull' })))); + expect(cutOnly.code).toBe('E_NEXUS_VAULT_BEHIND'); + expect((await on(b, () => verifyNexusVault(vopts(b)))).verdict).toBe('behind'); + // A device that writes the journal is not held behind by it. + expect((await on(a, () => verifyNexusVault(vopts(a)))).verdict).toBe('match'); + // This client sent the stream-reader header throughout; a client without it (released + // v2026.10.19) is refused a v3 stream's reads by a T13876 server, never shown it as current. + const old = await fake.fetch(`${API}/v1/streams/${encodeURIComponent(STREAM)}/checkpoints`, { + headers: { authorization: `Bearer ${b.token}` }, + }); + expect(old.status).toBe(409); + expect(JSON.stringify(await old.json())).toContain('E_STREAM_VERSION'); + }); + + it('T13867: on a journal head, a table only snapshots carry is listed but does not decide the verdict', async () => { + const { a, aDb } = await journalDevices(false); + await on(a, async () => + (await storeOf(aDb)).exec( + "INSERT INTO schedules (schedule_id, cron_expr, title) VALUES ('s-t13867', '0 * * * *', 'machine-local schedule')", + ), + ); + const v = await on(a, () => verifyNexusVault(vopts(a))); + expect(v.tables.find((r) => r.table === 'schedules')?.match).toBe(false); + expect(v.verdict).toBe('match'); + expect(v.warnings.find((w) => w.code === 'W_NEXUS_VAULT_SNAPSHOT_ONLY')?.message).toContain( + 'schedules', + ); + }); + it('AC1: with sync.push on, cloud push checkpoints at the sealed journal hwm with no delta segment; the server v3 rule accepts it', async () => { const { a, aDb, genesis } = await journalDevices(true); // A write not sealed yet: the push seals and pushes it first. diff --git a/packages/core/src/cloud/http.ts b/packages/core/src/cloud/http.ts index c3abc6f53b..6e5a7b40e6 100644 --- a/packages/core/src/cloud/http.ts +++ b/packages/core/src/cloud/http.ts @@ -59,6 +59,14 @@ export function isSecureUrl(url: string): boolean { } } +/** + * The header saying this client reads journal (checkpoint/v3) streams (T13867). + * Cleo Nexus refuses a v3 stream's checkpoint and head reads to a request + * without it (T13876): such a client reads only snapshots and would report a + * stale store as current. Released v2026.10.19 does not send it. + */ +export const NEXUS_STREAM_READER_HEADER = { name: 'x-cleo-stream-reader', value: '3' } as const; + /** * Minimal envelope-aware HTTP client. It retries only failures that are safe to retry. Every API * write is idempotent or lineage-checked, so a retried write either no-ops or is refused explicitly. @@ -116,6 +124,7 @@ export class Http { try { const headers: Record = { authorization: `Bearer ${this.o.token}` }; if (this.o.deviceId) headers['x-cleo-device-id'] = this.o.deviceId; + headers[NEXUS_STREAM_READER_HEADER.name] = NEXUS_STREAM_READER_HEADER.value; if (body !== undefined) headers['content-type'] = 'application/json'; res = await this.fetch(`${this.o.baseUrl}${path}`, { method, diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts index a529f49a66..08d03206cb 100644 --- a/packages/core/src/cloud/nexus-vault.ts +++ b/packages/core/src/cloud/nexus-vault.ts @@ -582,6 +582,37 @@ async function replaySegments( const JOURNAL_STREAM_REMEDY = 'on this stream local changes travel through the change journal (`sync.push`), and with it on a vault push checkpoints the journal (T12999); a store that has not joined it joins with `cleo sync enable push`, which keeps its own rows and merges them into the stream (T13466); `cleo cloud pull`, `cleo cloud restore` and `cleo cloud verify` still work'; +/** What a store that reads only snapshots does about journal changes past the head (T13867). */ +const JOURNAL_UNREAD_REMEDY = + 'join the change journal with `cleo sync enable push` to receive them, or run `cleo cloud push` on a device that writes the journal (its push checkpoints the journal, T12999), then pull again'; + +/** + * The change-journal segments past a checkpoint/v3 head that this store does + * not hold (T13867): it reads snapshots only, so the head is all it has, and + * every segment past the head's `coversSeq` is missing from it. A store is on + * the journal when it holds a genesis cut of the stream AND `sync.push` or + * `sync.pull` is on: a cut alone, as a restored bundle could carry, does not + * make it one. 0 for any other head, or a store on the journal. + */ +async function journalUnread( + t: VaultTarget, + headSeq: number, + headCp: Checkpoint | null | undefined, +): Promise { + if (!headCp || !isJournalSnapshot(headCp) || headSeq <= headCp.coversSeq) return 0; + if (fs.existsSync(t.dbPath)) { + const { openNativeDatabase } = await import('../store/sqlite-native.js'); + const db = openNativeDatabase(t.dbPath, { readonly: true, enableWal: false }); + try { + const on = isSyncFlagOn(db, 'sync.push') || isSyncFlagOn(db, 'sync.pull'); + if (on && genesisCutOf(db, t.streamId) !== undefined) return 0; + } finally { + db.close(); + } + } + return headSeq - headCp.coversSeq; +} + /** Whether a snapshot is checkpoint/v3: the change journal writes its stream (T13034). */ function isJournalSnapshot(cp: Checkpoint | null | undefined): boolean { return cp !== null && cp !== undefined && manifestVersion(cp.manifest) === 3; @@ -3788,6 +3819,22 @@ async function restoreNexusVaultImpl(opts: NexusVaultRestoreOptions): Promise c.checkpointId === target) ?? null; + const unread = + target === head.headCheckpointId + ? await journalUnread( + t, + head.headSeq, + trusted.find((c) => c.checkpointId === target), + ) + : 0; + if (unread > 0) { + // @sync-invariant none:local-only a read-only refusal: nothing is written + throw vaultError( + 'E_NEXUS_VAULT_BEHIND', + `${t.streamId}: the change journal holds ${unread} segment(s) past snapshot ${target}, and this store reads snapshots only, so it is not up to date`, + JOURNAL_UNREAD_REMEDY, + ); + } warnings.push(...conn.state.drainWarnings()); return { apiUrl: conn.apiUrl, @@ -4109,6 +4156,16 @@ async function restoreNexusVaultImpl(opts: NexusVaultRestoreOptions): Promise 0) { + warnings.push({ + code: 'W_NEXUS_VAULT_JOURNAL_UNREAD', + message: `the change journal holds ${unread} segment(s) past snapshot ${restored.checkpoint.checkpointId}, which this store does not have: ${JOURNAL_UNREAD_REMEDY}`, + }); + } warnings.push(...conn.state.drainWarnings()); return { apiUrl: conn.apiUrl, @@ -4365,16 +4422,33 @@ async function verifyNexusVaultImpl( return { ...r, cloudRows, match: (r.localRows ?? 0) === cloudRows }; }); } + // On a journal head, tables only snapshots carry (files, other databases, the + // row-identity-exempt tables) hold machine-local state: listed, never the verdict (T13867). + const journalHead = isJournalSnapshot(headCp); + const snapshotOnly = new Set(Object.keys(ROW_IDENTITY_EXEMPT[tableScopeOf(t)])); + const decides = (r: CloudVaultTableDiff): boolean => + !journalHead || !(isSnapshotAuthoritativeKey(r.table) || snapshotOnly.has(r.table)); + const asideDiffer = tables.filter((r) => !r.match && !decides(r)).map((r) => r.table); + if (asideDiffer.length > 0) { + warnings.push({ + code: 'W_NEXUS_VAULT_SNAPSHOT_ONLY', + message: `${asideDiffer.length} table(s) only snapshots carry differ from the newest snapshot (machine-local, or not checkpointed yet), so they do not decide the verdict: ${asideDiffer.slice(0, 8).join(', ')}`, + }); + } + const unread = await journalUnread(t, head.headSeq, headCp); let verdict: CloudVerifyResult['verdict']; if (headUntrusted) verdict = 'untrusted'; else if (!headCp) verdict = 'empty'; - else if (tables.every((x) => x.match)) verdict = 'match'; + else if (unread > 0) verdict = 'behind'; + else if (tables.filter(decides).every((x) => x.match)) verdict = 'match'; else if (journalSide !== null) { const { localChanged, cloudAdvanced } = journalSide; verdict = localChanged && cloudAdvanced ? 'diverged' : cloudAdvanced ? 'behind' : 'ahead'; } else { const localChanged = last - ? compareWithSynced(local, last.manifest, git, fileChanges).some((d) => !d.match) + ? compareWithSynced(local, last.manifest, git, fileChanges).some( + (d) => !d.match && decides(d), + ) : true; const cloudAdvanced = synced?.lastCheckpointId !== headCp.checkpointId; verdict = localChanged && cloudAdvanced ? 'diverged' : cloudAdvanced ? 'behind' : 'ahead'; @@ -4433,7 +4507,6 @@ async function verifyNexusVaultImpl( // journal writes takes no push (T13034). const headFormat = formatWarning(t, headCp); if (headFormat) warnings.push(headFormat); - const journalHead = isJournalSnapshot(headCp); const remedy = deepFailure !== null ? `the cloud copy failed the deep check (${deepFailure}): do not pull it; see which device pushed the newest snapshot with \`cleo cloud vault\` and \`cleo cloud activity\`, then push a fresh snapshot from a machine whose store passes \`cleo cloud verify\`` @@ -4442,7 +4515,9 @@ async function verifyNexusVaultImpl( : headCp && vaultManifestFormat(headCp.manifest, hashKeyOf(t.dataKey)) === 'unknown' ? `the newest snapshot records a vault manifest format this CLEO does not recognise (this CLEO hashes format ${VAULT_MANIFEST_FORMAT_VERSION}): if a newer CLEO pushed it, upgrade CLEO on this machine, then verify again` : verdict === 'behind' - ? 'run `cleo cloud pull` to bring this machine to the newest snapshot' + ? unread > 0 + ? `the change journal holds ${unread} segment(s) past the newest snapshot, which this store reads snapshots only and does not have: ${JOURNAL_UNREAD_REMEDY}` + : 'run `cleo cloud pull` to bring this machine to the newest snapshot' : verdict === 'ahead' ? journalHead ? `this machine changed since the newest snapshot, which is checkpoint/v3 (the change journal writes this stream): ${JOURNAL_STREAM_REMEDY}`