From 300130b72583375af74925291572337b68a46ee9 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 19:11:31 -0700 Subject: [PATCH 1/2] feat(cloud): cleo cloud sync - seal, push, pull and apply each attached stream (T12996) cloudSync (cloud/nexus-vault.ts) runs pushSyncStream then pullSyncStream for each attached stream (the project's and the account's global store, or the one --scope names) and returns one CloudSyncResult: per stream a status (synced, disabled, paused, not-attached, refused) and sealed, sent, received, staged, applied, held and conflict counts with the staged position against the server's head. E_SYNC_DISABLED names `cleo sync enable push` when no attached stream has a journal flag on; pull now refuses on sync.pull before it looks for a position. An interrupted run resumes without sending or applying anything twice. `cleo cloud sync [--scope]` is the CLI verb. Co-Authored-By: Claude Opus 5.5 --- .changeset/t12996-cloud-sync.md | 21 +++ packages/cleo/src/cli/commands/cloud.ts | 21 +++ .../cli/lib/__tests__/nexus-vault-cli.test.ts | 35 +++++ packages/cleo/src/cli/lib/nexus-vault-cli.ts | 27 ++++ packages/contracts/src/index.ts | 2 + packages/contracts/src/nexus-account.ts | 2 + packages/contracts/src/nexus-vault.ts | 37 +++++ .../src/cloud/__tests__/nexus-vault.test.ts | 81 ++++++++++ packages/core/src/cloud/nexus-vault.ts | 142 +++++++++++++++++- 9 files changed, 365 insertions(+), 3 deletions(-) create mode 100644 .changeset/t12996-cloud-sync.md diff --git a/.changeset/t12996-cloud-sync.md b/.changeset/t12996-cloud-sync.md new file mode 100644 index 0000000000..0d2ee6083e --- /dev/null +++ b/.changeset/t12996-cloud-sync.md @@ -0,0 +1,21 @@ +--- +id: t12996-cloud-sync +tasks: [T12996] +kind: feat +summary: "`cleo cloud sync` - seal, push, pull and apply each attached stream in one call, one result per stream" +--- + +`cleo cloud sync [--scope project|global]` runs the change journal end to end for each attached stream: the project's and the +account's global store, or just the one `--scope` names (T12996). + +- **Per stream:** it seals pending writes, pushes them as segments (`pushSyncStream`), pulls the stream's new segments, and applies + them (`pullSyncStream`). +- **One LAFS envelope** (`CloudSyncResult`) reports, per stream: + - `status`: `synced`, `disabled`, `paused` (device clock ahead), `not-attached` or `refused`; + - sealed, built, sent, duplicates, received, staged, redelivered, applied, held and conflict counts; + - the staged position against the server's head. +- **`E_SYNC_DISABLED`** is returned, naming `cleo sync enable push`, when no attached stream has `sync.push` or `sync.pull` on. A store + with `sync.pull` off now refuses its pull before it looks for a pull position. +- **An interrupted run resumes on the next:** a segment is resent with the same bytes, and a transaction is never staged or applied + twice. +- `sync.push` and `sync.pull` stay unreleased: the CLI never opts in, so stores can't turn them on yet. diff --git a/packages/cleo/src/cli/commands/cloud.ts b/packages/cleo/src/cli/commands/cloud.ts index d1bb8f1908..b7c7164848 100644 --- a/packages/cleo/src/cli/commands/cloud.ts +++ b/packages/cleo/src/cli/commands/cloud.ts @@ -41,6 +41,7 @@ import { runCloudPull, runCloudPush, runCloudRestore, + runCloudSync, runCloudVault, runCloudVerify, } from '../lib/nexus-vault-cli.js'; @@ -128,6 +129,25 @@ const projectsSubCommand = defineCommand({ }, }); +const syncSubCommand = defineCommand({ + meta: { + name: 'sync', + description: + "Sync the change journal with Cleo Nexus: seal pending writes, push them as segments, pull the stream's new segments and apply them, for each attached stream (the project's and the account's global store; --scope picks one). One envelope with per-stream sealed, sent, received, applied, held and conflict counts and the staged position against the server's head. Refused with E_SYNC_DISABLED until `cleo sync enable push`. An interrupted run resumes on the next: nothing is sent or applied twice.", + }, + args: { + scope: { + type: 'string', + description: "One stream: 'project' or 'global' (default: every attached stream).", + }, + 'api-url': NEXUS_API_URL_ARG, + json: JSON_ARG, + }, + async run({ args }) { + await runCloudSync(args as Record); + }, +}); + const pushSubCommand = defineCommand({ meta: { name: 'push', @@ -311,6 +331,7 @@ export const cloudCommand = defineCommand({ projects: projectsSubCommand, activity: activitySubCommand, conflicts: conflictsSubCommand, + sync: syncSubCommand, push: pushSubCommand, pull: pullSubCommand, restore: restoreSubCommand, diff --git a/packages/cleo/src/cli/lib/__tests__/nexus-vault-cli.test.ts b/packages/cleo/src/cli/lib/__tests__/nexus-vault-cli.test.ts index a196417dc1..c8c0cc24c3 100644 --- a/packages/cleo/src/cli/lib/__tests__/nexus-vault-cli.test.ts +++ b/packages/cleo/src/cli/lib/__tests__/nexus-vault-cli.test.ts @@ -13,6 +13,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; const pushNexusVault = vi.fn(); const enableSyncPush = vi.fn(); +const cloudSync = vi.fn(); const restoreNexusVault = vi.fn(); const verifyNexusVault = vi.fn(); const nexusVaultStatus = vi.fn(); @@ -23,6 +24,7 @@ const resolveNexusProjectRef = vi.fn(); const assertNexusRestoreTarget = vi.fn(); vi.mock('@cleocode/core/cloud/nexus-vault.js', () => ({ + cloudSync, enableSyncPush, pushNexusVault, restoreNexusVault, @@ -46,6 +48,7 @@ const { runCloudPull, runCloudPush, runCloudRestore, + runCloudSync, runCloudVault, runCloudVerify, runSyncEnablePush, @@ -177,6 +180,38 @@ describe('flags reach the core calls', () => { expect(written()).toContain('cp-1'); }); + it('cloud sync passes the API URL, and a scope only when one is given (T12996)', async () => { + cloudSync.mockReset().mockResolvedValue({ + apiUrl: API, + streams: [ + { + scope: 'project', + streamId: 'project:p', + status: 'synced', + refused: null, + sealed: 1, + built: 1, + sent: 1, + duplicates: 0, + received: 2, + staged: 2, + redelivered: 0, + applied: 2, + held: 0, + conflicts: 0, + after: 7, + head: 7, + }, + ], + warnings: [], + }); + await runCloudSync({ 'api-url': API }); + expect(opts(cloudSync)).toEqual({ apiUrl: API }); + await runCloudSync({ scope: 'global' }); + expect(cloudSync.mock.calls[1]?.[0]).toEqual({ apiUrl: undefined, scope: 'global' }); + expect(written()).toContain('project:p'); + }); + it('restore passes checkpoint, project, into and force', async () => { await runCloudRestore({ scope: 'project', diff --git a/packages/cleo/src/cli/lib/nexus-vault-cli.ts b/packages/cleo/src/cli/lib/nexus-vault-cli.ts index b2c25cb40f..ba638f3b85 100644 --- a/packages/cleo/src/cli/lib/nexus-vault-cli.ts +++ b/packages/cleo/src/cli/lib/nexus-vault-cli.ts @@ -18,6 +18,7 @@ import type { CloudPushResult, CloudRestoreResult, CloudSyncPushEnableResult, + CloudSyncResult, CloudVaultScope, CloudVaultStatusResult, CloudVerifyResult, @@ -106,6 +107,32 @@ export async function runSyncEnablePush(args: Args): Promise { ); } +/** + * `cleo cloud sync [--scope]`: seal, push, pull and apply each attached + * stream (T12996). Without `--scope`, every attached stream. + * + * @param args - Parsed args. + */ +export async function runCloudSync(args: Args): Promise { + const scope = stringArg(args, 'scope') === undefined ? undefined : scopeArg(args, 'cloud.sync'); + await runCloudRead( + 'cloud.sync', + async () => + (await vaultModule()).cloudSync({ + apiUrl: nexusApiUrlArg(args), + ...(scope !== undefined ? { scope } : {}), + }), + (r) => + r.streams + .map((st) => + st.status === 'synced' + ? `${st.streamId}: sent ${st.sent} segment(s), received ${st.received}, applied ${st.applied}${st.held > 0 ? `, ${st.held} held` : ''}${st.conflicts > 0 ? `, ${st.conflicts} in conflict` : ''} (at ${st.after} of ${st.head}).` + : `${st.streamId ?? st.scope}: ${st.status}${st.refused ? ` (${st.refused})` : ''}.`, + ) + .join('\n'), + ); +} + /** * `cleo cloud pull [--scope] [--force]`. * diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 2518c8c505..77502e08f4 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -3137,6 +3137,8 @@ export { type CloudPushResult, type CloudRestoreResult, type CloudSyncPushEnableResult, + type CloudSyncResult, + type CloudSyncStreamResult, type CloudVaultLease, type CloudVaultScope, type CloudVaultSnapshot, diff --git a/packages/contracts/src/nexus-account.ts b/packages/contracts/src/nexus-account.ts index f5ef38f2e8..fbc3b384fa 100644 --- a/packages/contracts/src/nexus-account.ts +++ b/packages/contracts/src/nexus-account.ts @@ -106,6 +106,8 @@ export const NEXUS_ACCOUNT_ERROR_CODES = [ 'E_NEXUS_SYNC_REFUSED', /** Another device already started this stream's change journal: this store joins it by pulling. */ 'E_NEXUS_SYNC_STREAM_JOURNALED', + /** `cleo cloud sync` found no stream with `sync.push` or `sync.pull` on (`cleo sync enable`). */ + 'E_SYNC_DISABLED', // Projects by name (`cleo cloud restore `, T13102). /** No project of the account has that name, label or id. */ 'E_NEXUS_PROJECT_NOT_FOUND', diff --git a/packages/contracts/src/nexus-vault.ts b/packages/contracts/src/nexus-vault.ts index cf2ab80b25..d357991ef6 100644 --- a/packages/contracts/src/nexus-vault.ts +++ b/packages/contracts/src/nexus-vault.ts @@ -204,6 +204,43 @@ export interface CloudSyncPushEnableResult { warnings: CloudWarning[]; } +/** One stream of `cleo cloud sync` (T12996). */ +export interface CloudSyncStreamResult { + scope: CloudVaultScope; + streamId: string | null; + /** + * `synced`: pushed and pulled what was enabled; `disabled`: neither `sync.push` nor `sync.pull` is on; + * `paused`: push paused for a device clock ahead of the server's; `not-attached`: this machine has no + * store on that stream; `refused`: a precondition or the server refused (see `refused`). + */ + status: 'synced' | 'disabled' | 'paused' | 'not-attached' | 'refused'; + /** Why this stream did not sync fully, or null. */ + refused: string | null; + /** Transactions sealed, segments persisted, segments the server stored (`duplicates` of them retries). */ + sealed: number; + built: number; + sent: number; + duplicates: number; + /** Segments received, transactions staged, re-deliveries skipped. */ + received: number; + staged: number; + redelivered: number; + /** Transactions applied; still held (pending, clock skew, or a newer schema); in conflict. */ + applied: number; + held: number; + conflicts: number; + /** The stream position this store has staged up to, and the server's head (null when not pulled). */ + after: number | null; + head: number | null; +} + +/** `cleo cloud sync` (T12996): seal, push, pull and apply each attached stream. */ +export interface CloudSyncResult { + apiUrl: string; + streams: CloudSyncStreamResult[]; + warnings: CloudWarning[]; +} + /** `cleo cloud pull` / `cleo cloud restore`. */ export interface CloudRestoreResult { apiUrl: string; diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts index 9de695d88f..78ff68c5c8 100644 --- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-vault.test.ts @@ -118,6 +118,7 @@ import { hasUnsyncedNexusBackup, runNexusFirstRun } from '../nexus-first-run.js' import { linkProjectToNexus } from '../nexus-link.js'; import { listNexusNamedProjects, resolveNexusProjectRef } from '../nexus-project-names.js'; import { + cloudSync, enableSyncPush, nexusVaultStatus, pullSyncStream, @@ -5156,6 +5157,86 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { expect(again).toMatchObject({ segments: 0, staged: 0 }); }); + it('C-1: cloud sync seals, pushes, pulls and applies the stream in one call; a rerun does nothing', async () => { + const { m, dbPath } = await journalMachine(); + await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true }))); + await on(m, async () => { + const db = await storeOf(dbPath); + setSyncFlag(db, 'sync.pull', true, { allowUnreleased: true }); + db.exec( + "INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('C1', 'title C1', 'task', 'pending', 'medium', 'uid-C1', 'fp-C1')", + ); + }); + const r = await on(m, () => cloudSync(vopts(m, { scope: 'project', allowUnreleased: true }))); + expect(r.streams).toHaveLength(1); + expect(r.streams[0]).toMatchObject({ + scope: 'project', + streamId: STREAM, + status: 'synced', + refused: null, + sealed: 1, + sent: 1, + received: 1, + staged: 1, + applied: 1, + held: 0, + conflicts: 0, + }); + expect(r.streams[0]?.after).toBe(r.streams[0]?.head); + const again = await on(m, () => + cloudSync(vopts(m, { scope: 'project', allowUnreleased: true })), + ); + expect(again.streams[0]).toMatchObject({ status: 'synced', sealed: 0, sent: 0, received: 0 }); + }); + + it('C-1: an interrupted sync resumes on the next run with nothing sent or applied twice', async () => { + const { m, dbPath } = await journalMachine(); + await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true }))); + await on(m, async () => { + const db = await storeOf(dbPath); + setSyncFlag(db, 'sync.pull', true, { allowUnreleased: true }); + db.exec( + "INSERT INTO tasks_tasks (id, title, type, status, priority, uid, birth_fp) VALUES ('C2', 'title C2', 'task', 'pending', 'medium', 'uid-C2', 'fp-C2')", + ); + }); + fake.beforeSegment = { + deviceId: DEVICE_A, + run: async () => { + throw new ApiFail(403, 'E_FORBIDDEN'); + }, + }; + await expect( + on(m, () => cloudSync(vopts(m, { scope: 'project', allowUnreleased: true }))), + ).rejects.toThrow(); + const before = fake.stream(STREAM).segments.length; + const r = await on(m, () => cloudSync(vopts(m, { scope: 'project', allowUnreleased: true }))); + expect(r.streams[0]).toMatchObject({ + status: 'synced', + built: 0, + sent: 1, + received: 1, + applied: 1, + }); + expect(fake.stream(STREAM).segments).toHaveLength(before + 1); + await on(m, async () => { + const db = await storeOf(dbPath); + expect((db.prepare('SELECT count(*) AS n FROM _sync_inbox').get() as { n: number }).n).toBe( + 1, + ); + expect( + (db.prepare("SELECT count(*) AS n FROM tasks_tasks WHERE id = 'C2'").get() as { n: number }) + .n, + ).toBe(1); + }); + }); + + it('C-1: with the journal flags off, cloud sync refuses with E_SYNC_DISABLED naming the remedy', async () => { + const { m } = await journalMachine(); + const refused = await failure(on(m, () => cloudSync(vopts(m, { scope: 'project' })))); + expect(refused.code).toBe('E_SYNC_DISABLED'); + expect(refused.fix).toContain('cleo sync enable push'); + }); + it('S4-2: a device clock ahead of the server pauses push', async () => { const { m, dbPath } = await journalMachine(); await on(m, () => enableSyncPush(vopts(m, { allowUnreleased: true }))); diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts index 52f156a309..b9e88a0eb1 100644 --- a/packages/core/src/cloud/nexus-vault.ts +++ b/packages/core/src/cloud/nexus-vault.ts @@ -62,6 +62,8 @@ import type { CloudPushResult, CloudRestoreResult, CloudSyncPushEnableResult, + CloudSyncResult, + CloudSyncStreamResult, CloudVaultLease, CloudVaultScope, CloudVaultSnapshot, @@ -103,7 +105,7 @@ import { sha256File, } from '../store/portable-bundle-scan.js'; import { FIRST_OPEN_LOCK_SUFFIX } from '../store/sqlite.js'; -import { UNRELEASED_FLAGS } from '../store/sync/flags.js'; +import { isSyncFlagOn, UNRELEASED_FLAGS } from '../store/sync/flags.js'; import { completeGenesis, cutGenesisWithSnapshot, @@ -1852,7 +1854,8 @@ async function pullSyncStreamImpl(opts: NexusVaultCommandOptions = {}): Promise< // (the genesis it cut, or the snapshot it restored), seeded from its signed // replica map. let initialCursor = readStreamCursor(db, t.streamId); - if (initialCursor === null) { + // With sync.pull off, pullStream refuses before it needs a position. + if (initialCursor === null && isSyncFlagOn(db, 'sync.pull')) { const head = await streamHead(conn, t.streamId); const checkpoints = head.headCheckpointId ? await listCheckpoints(conn, t.streamId) : []; const { state: synced } = syncedState(conn, t, checkpoints, head.headCheckpointId); @@ -1871,7 +1874,7 @@ async function pullSyncStreamImpl(opts: NexusVaultCommandOptions = {}): Promise< scope: tableScopeOf(t), stream: t.streamId, replica: replicaId, - initialCursor, + initialCursor: initialCursor ?? initialPullCursor(), pull: async (cursor) => { const page = await journal.pull( { @@ -1909,6 +1912,120 @@ async function pullSyncStreamImpl(opts: NexusVaultCommandOptions = {}): Promise< }); } +/** A `cleo cloud sync` stream result with nothing done yet. */ +const syncStreamResult = ( + scope: CloudVaultScope, + fields: Partial, +): CloudSyncStreamResult => ({ + scope, + streamId: null, + status: 'synced', + refused: null, + sealed: 0, + built: 0, + sent: 0, + duplicates: 0, + received: 0, + staged: 0, + redelivered: 0, + applied: 0, + held: 0, + conflicts: 0, + after: null, + head: null, + ...fields, +}); + +/** Codes meaning this machine has no store on the stream of a scope. */ +const NOT_ATTACHED_CODES: ReadonlySet = new Set([ + 'E_NEXUS_VAULT_NOT_LINKED', + 'E_NEXUS_NOT_A_PROJECT', +]); + +/** Seal, push, pull and apply one scope's stream (T12996). */ +async function syncOneStream( + opts: NexusVaultCommandOptions & { allowUnreleased?: boolean }, + scope: CloudVaultScope, +): Promise { + const failed = (err: unknown): CloudSyncStreamResult | null => { + if (!(err instanceof NexusAccountError)) return null; + return NOT_ATTACHED_CODES.has(err.code) + ? syncStreamResult(scope, { status: 'not-attached', refused: err.message }) + : syncStreamResult(scope, { status: 'refused', refused: `${err.code}: ${err.message}` }); + }; + let push: PushStreamReport; + try { + push = await pushSyncStreamImpl({ ...opts, scope }); + } catch (err) { + const r = failed(err); + if (r) return r; + throw err; + } + let pull: PullStreamReport; + try { + pull = await pullSyncStreamImpl({ ...opts, scope }); + } catch (err) { + const r = failed(err); + if (r) + return { + ...r, + streamId: push.stream, + sealed: push.sealed, + built: push.built, + sent: push.pushed, + }; + throw err; + } + const pushOff = push.refused === 'sync.push is off'; + const pullOff = pull.refused === 'sync.pull is off'; + const refused = [push.refused, pull.refused].filter( + (r): r is string => r !== null && r !== 'sync.push is off' && r !== 'sync.pull is off', + ); + const apply = pull.apply; + return syncStreamResult(scope, { + streamId: push.stream, + status: + pushOff && pullOff + ? 'disabled' + : refused.length > 0 + ? 'refused' + : push.clockAhead + ? 'paused' + : 'synced', + refused: refused.length > 0 ? refused.join('; ') : null, + sealed: push.sealed, + built: push.built, + sent: push.pushed, + duplicates: push.duplicates, + received: pull.segments, + staged: pull.staged, + redelivered: pull.redelivered, + applied: apply?.applied ?? 0, + held: (apply?.pending ?? 0) + (apply?.heldSkew ?? 0) + (apply?.refusedSchema ?? 0), + conflicts: apply?.conflict ?? 0, + after: pull.refused === null ? pull.after : null, + head: pull.refused === null ? pull.head : null, + }); +} + +async function cloudSyncImpl( + opts: NexusVaultCommandOptions & { allowUnreleased?: boolean } = {}, +): Promise { + const conn = await connectNexusVault(opts); + const scopes: CloudVaultScope[] = opts.scope ? [opts.scope] : ['project', 'global']; + const streams: CloudSyncStreamResult[] = []; + for (const scope of scopes) streams.push(await syncOneStream(opts, scope)); + const worked = streams.filter((r) => r.status !== 'disabled' && r.status !== 'not-attached'); + if (worked.length === 0 && streams.some((r) => r.status === 'disabled')) { + throw vaultError( + 'E_SYNC_DISABLED', + 'no attached stream has sync.push or sync.pull on', + 'turn the change journal on first: `cleo sync enable push`', + ); + } + return { apiUrl: conn.apiUrl, streams, warnings: [...conn.warnings] }; +} + /** The keyed hash of an empty table (a table the parent lists that this store no longer has). */ function buildEmptyHash(t: VaultTarget, table: string): string { return emptyVaultTableHash(hashKeyOf(t.dataKey), table); @@ -2778,6 +2895,25 @@ export function pullSyncStream(opts: NexusVaultCommandOptions = {}): Promise pullSyncStreamImpl(opts)); } +/** + * `cleo cloud sync`: seal, push, pull and apply the change journal of each + * attached stream (the project's `project:`, the account's `home:`; + * T12996). One result per stream: sealed, persisted, sent, received, staged, + * applied, held and in conflict, and how far the store has staged against + * the server's head. An interrupted run resumes on the next: a segment is + * resent with the same bytes, and a transaction is never staged twice. + * + * @param opts - `scope` (one stream; default every attached one), overrides; + * `allowUnreleased` for tests and staging only. + * @returns Every stream's result. + * @throws {NexusAccountError} `E_SYNC_DISABLED` when no attached stream has a journal flag on. + */ +export function cloudSync( + opts: NexusVaultCommandOptions & { allowUnreleased?: boolean } = {}, +): Promise { + return mapped(() => cloudSyncImpl(opts)); +} + /** * `cleo cloud pull` / `cleo cloud restore`: verify and activate a snapshot. * From 244b21114a153031d3c9bbb7a8e1049321433cc6 Mon Sep 17 00:00:00 2001 From: Keaton Hoskins Date: Tue, 6 Oct 2026 20:07:44 -0700 Subject: [PATCH 2/2] fix(cloud): cloud sync isolates streams, shares one session, skips idle legs (T12996 review) review-hotfix LOWs on #1962: - LOW-1: any error on a stream is reported on that stream (status 'failed', with its code), never thrown, so the next stream still syncs. - LOW-2: one connection and one account-key unlock serve every stream of a run (push and pull take a shared session) instead of five connects and four keychain reads. - LOW-3: a flag off, or push before this store has its own genesis, is reported in `skipped`, not as `refused`, so a store that only pulls shows synced. Co-Authored-By: Claude Opus 5.5 --- .changeset/t12996-cloud-sync.md | 5 +- packages/contracts/src/nexus-vault.ts | 10 +- .../src/cloud/__tests__/nexus-vault.test.ts | 8 +- packages/core/src/cloud/nexus-vault.ts | 128 +++++++++++++----- 4 files changed, 110 insertions(+), 41 deletions(-) diff --git a/.changeset/t12996-cloud-sync.md b/.changeset/t12996-cloud-sync.md index 0d2ee6083e..31c23205ff 100644 --- a/.changeset/t12996-cloud-sync.md +++ b/.changeset/t12996-cloud-sync.md @@ -11,11 +11,14 @@ account's global store, or just the one `--scope` names (T12996). - **Per stream:** it seals pending writes, pushes them as segments (`pushSyncStream`), pulls the stream's new segments, and applies them (`pullSyncStream`). - **One LAFS envelope** (`CloudSyncResult`) reports, per stream: - - `status`: `synced`, `disabled`, `paused` (device clock ahead), `not-attached` or `refused`; + - `status`: `synced`, `disabled`, `paused` (device clock ahead), `not-attached`, `refused`, or `failed` (an error stopped this + stream, reported, never thrown, so the next stream still syncs); + - `skipped`: legs with nothing to do (a flag off, or push before this store has its own genesis), which never count as refused; - sealed, built, sent, duplicates, received, staged, redelivered, applied, held and conflict counts; - the staged position against the server's head. - **`E_SYNC_DISABLED`** is returned, naming `cleo sync enable push`, when no attached stream has `sync.push` or `sync.pull` on. A store with `sync.pull` off now refuses its pull before it looks for a pull position. - **An interrupted run resumes on the next:** a segment is resent with the same bytes, and a transaction is never staged or applied twice. +- **One connection and one account-key unlock serve every stream of a run.** - `sync.push` and `sync.pull` stay unreleased: the CLI never opts in, so stores can't turn them on yet. diff --git a/packages/contracts/src/nexus-vault.ts b/packages/contracts/src/nexus-vault.ts index d357991ef6..ae78c7dfc6 100644 --- a/packages/contracts/src/nexus-vault.ts +++ b/packages/contracts/src/nexus-vault.ts @@ -211,11 +211,17 @@ export interface CloudSyncStreamResult { /** * `synced`: pushed and pulled what was enabled; `disabled`: neither `sync.push` nor `sync.pull` is on; * `paused`: push paused for a device clock ahead of the server's; `not-attached`: this machine has no - * store on that stream; `refused`: a precondition or the server refused (see `refused`). + * store on that stream; `refused`: a precondition or the server refused (see `refused`); `failed`: an + * error stopped this stream (see `refused`); the other streams still synced. */ - status: 'synced' | 'disabled' | 'paused' | 'not-attached' | 'refused'; + status: 'synced' | 'disabled' | 'paused' | 'not-attached' | 'refused' | 'failed'; /** Why this stream did not sync fully, or null. */ refused: string | null; + /** + * Legs with nothing to do, not failures: a flag off, or push before this store has its own genesis + * (a store that only pulls). + */ + skipped: string[]; /** Transactions sealed, segments persisted, segments the server stored (`duplicates` of them retries). */ sealed: number; built: number; diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts index deb03bfde7..7cae0f777a 100644 --- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts +++ b/packages/core/src/cloud/__tests__/nexus-vault.test.ts @@ -5205,9 +5205,11 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => { throw new ApiFail(403, 'E_FORBIDDEN'); }, }; - await expect( - on(m, () => cloudSync(vopts(m, { scope: 'project', allowUnreleased: true }))), - ).rejects.toThrow(); + // The failure is this stream's, reported, never thrown: every other stream still syncs. + const failed = await on(m, () => cloudSync(vopts(m, { allowUnreleased: true }))); + expect(failed.streams.map((x) => x.scope)).toEqual(['project', 'global']); + expect(failed.streams[0]).toMatchObject({ status: 'failed', sent: 0 }); + expect(failed.streams[0]?.refused).toMatch(/E_FORBIDDEN/); const before = fake.stream(STREAM).segments.length; const r = await on(m, () => cloudSync(vopts(m, { scope: 'project', allowUnreleased: true }))); expect(r.streams[0]).toMatchObject({ diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts index 944a328fb1..32754733e5 100644 --- a/packages/core/src/cloud/nexus-vault.ts +++ b/packages/core/src/cloud/nexus-vault.ts @@ -158,6 +158,7 @@ import { nexusApiErrorToAccountError } from './nexus-enrol.js'; import { readNexusProjectLink } from './nexus-link.js'; import { connectNexusVault, + type NexusAccountKey, type NexusVaultConnection, type NexusVaultOptions, nexusHomeDataKey, @@ -1749,15 +1750,35 @@ async function enableSyncPushImpl( } } +/** One connection, the account key and one scope's store, shared by a stream's push and pull (T12996). */ +interface StreamSession { + readonly conn: NexusVaultConnection; + readonly key: NexusAccountKey; + readonly t: VaultTarget; +} + +/** Connect, unlock the account key and resolve the scope's store, once. */ +async function openStreamSession(opts: NexusVaultCommandOptions): Promise { + const conn = await connectNexusVault(opts); + const key = await unlockNexusAccountKey(conn); + const t = await resolveTarget(conn, key.masterKey, opts, 'push'); + return { conn, key, t }; +} + async function pushSyncStreamImpl( opts: NexusVaultCommandOptions & { /** Push although `sync.push` is unreleased (tests and staging only). Never set from user input. */ allowUnreleased?: boolean; } = {}, ): Promise { - const conn = await connectNexusVault(opts); - const key = await unlockNexusAccountKey(conn); - const t = await resolveTarget(conn, key.masterKey, opts, 'push'); + return pushWithSession(await openStreamSession(opts), opts); +} + +async function pushWithSession( + session: StreamSession, + opts: { readonly allowUnreleased?: boolean }, +): Promise { + const { conn, t } = session; const replicaId = t.replicaId; if (!replicaId) { throw vaultError( @@ -1823,9 +1844,11 @@ async function pushSyncStreamImpl( } async function pullSyncStreamImpl(opts: NexusVaultCommandOptions = {}): Promise { - const conn = await connectNexusVault(opts); - const key = await unlockNexusAccountKey(conn); - const t = await resolveTarget(conn, key.masterKey, opts, 'push'); + return pullWithSession(await openStreamSession(opts)); +} + +async function pullWithSession(session: StreamSession): Promise { + const { conn, key, t } = session; const replicaId = t.replicaId; if (!replicaId) { throw vaultError( @@ -1952,6 +1975,7 @@ const syncStreamResult = ( streamId: null, status: 'synced', refused: null, + skipped: [], sealed: 0, built: 0, sent: 0, @@ -1973,48 +1997,79 @@ const NOT_ATTACHED_CODES: ReadonlySet = new Set([ 'E_NEXUS_NOT_A_PROJECT', ]); -/** Seal, push, pull and apply one scope's stream (T12996). */ +/** Push or pull refusals that mean "nothing to do here", not a failure (LOW-3 on #1962). */ +const SKIP_REASONS: readonly RegExp[] = [ + /^sync\.push is off$/, + /^sync\.pull is off$/, + /has no genesis cut on this store/, + /genesis checkpoint is not stored yet/, +]; + +/** The code and message of a failure, for a stream result. */ +function failureText(err: unknown): string { + if (err instanceof NexusAccountError || err instanceof NexusError) + return `${err.code}: ${err.message}`; + return err instanceof Error ? err.message : String(err); +} + +/** + * Seal, push, pull and apply one scope's stream (T12996). The run's one + * connection and account key serve both legs. A failure is reported on this stream + * (`failed`), never thrown, so the next stream still syncs. + */ async function syncOneStream( + shared: { readonly conn: NexusVaultConnection; readonly key: NexusAccountKey }, opts: NexusVaultCommandOptions & { allowUnreleased?: boolean }, scope: CloudVaultScope, ): Promise { - const failed = (err: unknown): CloudSyncStreamResult | null => { - if (!(err instanceof NexusAccountError)) return null; - return NOT_ATTACHED_CODES.has(err.code) - ? syncStreamResult(scope, { status: 'not-attached', refused: err.message }) - : syncStreamResult(scope, { status: 'refused', refused: `${err.code}: ${err.message}` }); - }; + let session: StreamSession; + try { + session = { + ...shared, + t: await resolveTarget(shared.conn, shared.key.masterKey, { ...opts, scope }, 'push'), + }; + } catch (err) { + if (err instanceof NexusAccountError && NOT_ATTACHED_CODES.has(err.code)) { + return syncStreamResult(scope, { status: 'not-attached', refused: err.message }); + } + return syncStreamResult(scope, { status: 'failed', refused: failureText(err) }); + } + const streamId = session.t.streamId; + if (!session.t.replicaId) { + return syncStreamResult(scope, { + streamId, + status: 'not-attached', + refused: 'this copy of the project is not attached from this device', + }); + } let push: PushStreamReport; try { - push = await pushSyncStreamImpl({ ...opts, scope }); + push = await pushWithSession(session, opts); } catch (err) { - const r = failed(err); - if (r) return r; - throw err; + return syncStreamResult(scope, { streamId, status: 'failed', refused: failureText(err) }); } let pull: PullStreamReport; try { - pull = await pullSyncStreamImpl({ ...opts, scope }); + pull = await pullWithSession(session); } catch (err) { - const r = failed(err); - if (r) - return { - ...r, - streamId: push.stream, - sealed: push.sealed, - built: push.built, - sent: push.pushed, - }; - throw err; + return syncStreamResult(scope, { + streamId, + status: 'failed', + refused: failureText(err), + sealed: push.sealed, + built: push.built, + sent: push.pushed, + duplicates: push.duplicates, + }); } + const isSkip = (r: string | null): boolean => r !== null && SKIP_REASONS.some((x) => x.test(r)); + const skipped = [push.refused, pull.refused].filter((r): r is string => isSkip(r)); + const refused = [push.refused, pull.refused].filter((r): r is string => r !== null && !isSkip(r)); const pushOff = push.refused === 'sync.push is off'; const pullOff = pull.refused === 'sync.pull is off'; - const refused = [push.refused, pull.refused].filter( - (r): r is string => r !== null && r !== 'sync.push is off' && r !== 'sync.pull is off', - ); const apply = pull.apply; return syncStreamResult(scope, { - streamId: push.stream, + streamId, status: pushOff && pullOff ? 'disabled' @@ -2024,6 +2079,7 @@ async function syncOneStream( ? 'paused' : 'synced', refused: refused.length > 0 ? refused.join('; ') : null, + skipped, sealed: push.sealed, built: push.built, sent: push.pushed, @@ -2034,18 +2090,20 @@ async function syncOneStream( applied: apply?.applied ?? 0, held: (apply?.pending ?? 0) + (apply?.heldSkew ?? 0) + (apply?.refusedSchema ?? 0), conflicts: apply?.conflict ?? 0, - after: pull.refused === null ? pull.after : null, - head: pull.refused === null ? pull.head : null, + after: pull.refused === null || !isSkip(pull.refused) ? pull.after : null, + head: pull.refused === null || !isSkip(pull.refused) ? pull.head : null, }); } async function cloudSyncImpl( opts: NexusVaultCommandOptions & { allowUnreleased?: boolean } = {}, ): Promise { + // One connection and one account-key unlock for every stream (LOW-2 on #1962). const conn = await connectNexusVault(opts); + const key = await unlockNexusAccountKey(conn); const scopes: CloudVaultScope[] = opts.scope ? [opts.scope] : ['project', 'global']; const streams: CloudSyncStreamResult[] = []; - for (const scope of scopes) streams.push(await syncOneStream(opts, scope)); + for (const scope of scopes) streams.push(await syncOneStream({ conn, key }, opts, scope)); const worked = streams.filter((r) => r.status !== 'disabled' && r.status !== 'not-attached'); if (worked.length === 0 && streams.some((r) => r.status === 'disabled')) { throw vaultError(