fix(T13851): journal-stream remedy names only commands this build can run

.changeset/t13851-journal-remedy-older-client.md

@@ -0,0 +1,6 @@
+---
+id: t13851-journal-remedy-older-client
+tasks: [T13851, T13856]
+kind: fix
+summary: "while sync.push is unreleased, journal-stream and E_SYNC_DISABLED remedies name no command this build refuses: they say a pre-release build started the journal, no released CLEO can push yet, and pull, restore and verify keep working"
+---

packages/core/src/cloud/__tests__/nexus-vault.test.ts

@@ -79,6 +79,7 @@ import {
   legacyStrands,
   recordBareAccounts,
   setSyncFlag,
+  UNRELEASED_FLAGS,
 } from '../../store/sync/flags.js';
 import { cutGenesis, genesisCutOf, genesisPending } from '../../store/sync/genesis.js';
 import { readStreamCursor } from '../../store/sync/pull.js';
@@ -149,13 +150,15 @@ import {
   confirmableRetirements,
   enableSyncPush,
   formatTransitionPruned,
+  journalStreamRemedy,
   nexusVaultStatus,
   pullSyncStream,
   pushNexusVault,
   pushSyncStream,
   releaseNexusVaultLease,
   restoreNexusVault,
   serverRetirementsFor,
+  syncDisabledRemedy,
   verifyNexusVault,
 } from '../nexus-vault.js';
 import {
@@ -3992,6 +3995,29 @@ function projectDataKey(): Buffer {
   );
 }
 
+describe('journalStreamRemedy names only commands this build can run (T13851)', () => {
+  it('a build without sync.push names no command it cannot run; a build with it says to join', () => {
+    // While sync.push is unreleased no released CLEO ships it, so self-update would loop (T13856).
+    const unreleased = journalStreamRemedy(false);
+    expect(unreleased).not.toContain('cleo self-update');
+    expect(unreleased).not.toContain('cleo sync enable push');
+    expect(unreleased).toContain('no released CLEO can push to it yet');
+    expect(unreleased).toContain('pre-release or allowUnreleased');
+    expect(unreleased).toContain('`cleo cloud pull`, `cleo cloud restore` and `cleo cloud verify`');
+    const released = journalStreamRemedy(true);
+    expect(released).toContain('cleo sync enable push');
+    expect(released).not.toContain('cleo self-update');
+  });
+
+  it('the E_SYNC_DISABLED fix names cleo sync enable push only where this build ships sync.push (T13856)', () => {
+    const unreleased = syncDisabledRemedy(false);
+    expect(unreleased).not.toContain('cleo sync enable push');
+    expect(unreleased).not.toContain('cleo self-update');
+    expect(unreleased).toContain('no released CLEO can push to it yet');
+    expect(syncDisabledRemedy(true)).toContain('cleo sync enable push');
+  });
+});
+
 describe('cloud vault on a stream the change journal writes (segment/v3, checkpoint/v3; T13034)', () => {
   const PIN_HASH = 'c'.repeat(64);
   const insertTask = (m: Machine, id: string) =>
@@ -4137,6 +4163,12 @@ describe('cloud vault on a stream the change journal writes (segment/v3, checkpo
     expect(behind.fix).toContain('cleo cloud pull');
     expect(behind.fix).toContain('change journal (`sync.push`)');
     expect(behind.fix).not.toContain('--force');
+    // This build has `sync.push` unreleased, so `cleo sync enable push` refuses here: the remedy
+    // names the upgrade, never the enable step it would refuse (T13851).
+    expect(UNRELEASED_FLAGS.has('sync.push')).toBe(true);
+    expect(behind.fix).not.toContain('cleo self-update');
+    expect(behind.fix).not.toContain('cleo sync enable push');
+    expect(behind.fix).toContain('no released CLEO can push to it yet');
     const refused = await failure(on(a, () => pushNexusVault(vopts(a, { force: true }))));
     expect(refused.code).toBe('E_NEXUS_VAULT_STREAM_UPGRADED');
     expect(refused.message).toContain('checkpoint/v3');
@@ -5996,7 +6028,10 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => {
     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');
+    // This build has sync.push unreleased, where `cleo sync enable push` refuses (T13856).
+    expect(UNRELEASED_FLAGS.has('sync.push')).toBe(true);
+    expect(refused.fix).toBe(syncDisabledRemedy(false));
+    expect(refused.fix).not.toContain('cleo sync enable push');
   });
 
   it('S5-1: a store that synced a vault snapshot from before the journal genesis refuses to pull across it (T13306)', async () => {

packages/core/src/cloud/nexus-vault.ts

@@ -578,9 +578,36 @@ async function replaySegments(
 /**
  * What works on a stream the change journal writes (its head snapshot is
  * checkpoint/v3), for every message that would otherwise say "push" (T13034).
+ * A build where `sync.push` is unreleased cannot join the journal, since `cleo sync enable push`
+ * refuses there (T13851). While it is unreleased no released CLEO ships it either, so the remedy
+ * never sends the user to `cleo self-update`: a pre-release or allowUnreleased build started the
+ * journal, and pull, restore and verify keep working (T13856).
+ *
+ * @param pushReleased - Whether this build ships `sync.push`; defaults to the release state.
+ * @returns The remedy clause, naming only commands this build can run.
  */
-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';
+export function journalStreamRemedy(pushReleased = !UNRELEASED_FLAGS.has('sync.push')): string {
+  if (pushReleased) {
+    return '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';
+  }
+  return `on this stream local changes travel through the change journal (\`sync.push\`), which checkpoints the stream (T12999) and which this CLEO cannot run: ${SYNC_PUSH_UNRELEASED_CLAUSE}; \`cleo cloud pull\`, \`cleo cloud restore\` and \`cleo cloud verify\` keep working`;
+}
+
+/** Why no CLEO can push while `sync.push` is unreleased (T13856): no release ships it yet. */
+const SYNC_PUSH_UNRELEASED_CLAUSE =
+  '`sync.push` is not released yet, so a pre-release or allowUnreleased CLEO build started this journal and no released CLEO can push to it yet (T13856)';
+
+/**
+ * The fix for E_SYNC_DISABLED: turn the journal on where this build can, and otherwise say that no
+ * released CLEO can yet (T13856), never naming `cleo sync enable push` where it refuses.
+ *
+ * @param pushReleased - Whether this build ships `sync.push`; defaults to the release state.
+ * @returns The fix text.
+ */
+export function syncDisabledRemedy(pushReleased = !UNRELEASED_FLAGS.has('sync.push')): string {
+  if (pushReleased) return 'turn the change journal on first: `cleo sync enable push`';
+  return `the change journal is off here and this CLEO cannot turn it on: ${SYNC_PUSH_UNRELEASED_CLAUSE}; \`cleo cloud pull\`, \`cleo cloud restore\` and \`cleo cloud verify\` keep working`;
+}
 
 /** Whether a snapshot is checkpoint/v3: the change journal writes its stream (T13034). */
 function isJournalSnapshot(cp: Checkpoint | null | undefined): boolean {
@@ -632,7 +659,7 @@ function aheadOfThisCleo(
 function journalStreamRefusal(t: VaultTarget, parent: Checkpoint): NexusAccountError {
   return streamUpgradedError(
     `${t.streamId} takes only checkpoint/v3 snapshots: its head snapshot ${parent.checkpointId} is one (the change journal writes this stream), and the server refuses the checkpoint/v2 snapshots this CLEO's vault writes after it`,
-    `nothing was written; ${JOURNAL_STREAM_REMEDY}`,
+    `nothing was written; ${journalStreamRemedy()}`,
   );
 }
 
@@ -1325,7 +1352,7 @@ async function appendCheckpoint(a: {
       if (err instanceof NexusError && err.code === 'E_STREAM_VERSION') {
         throw streamUpgradedError(
           `the server refused the snapshot: ${t.streamId} takes only checkpoint/v3 snapshots (${err.message})`,
-          JOURNAL_STREAM_REMEDY,
+          journalStreamRemedy(),
         );
       }
       const countRefusal =
@@ -1872,7 +1899,7 @@ async function pushNexusVaultImpl(
   if (cut !== undefined) {
     throw streamUpgradedError(
       `this store started ${t.streamId}'s change journal (genesis cut at capture ${cut}); its changes are pushed as journal segments, not vault snapshots`,
-      `nothing was written; ${JOURNAL_STREAM_REMEDY}`,
+      `nothing was written; ${journalStreamRemedy()}`,
     );
   }
   const force = opts.force === true;
@@ -1915,7 +1942,7 @@ async function pushNexusVaultImpl(
       'E_NEXUS_VAULT_BEHIND',
       `another device pushed snapshot ${head.headCheckpointId} after this machine's last sync`,
       isJournalSnapshot(parent)
-        ? `run \`cleo cloud pull\` first (it refuses to overwrite local changes); ${JOURNAL_STREAM_REMEDY}`
+        ? `run \`cleo cloud pull\` first (it refuses to overwrite local changes); ${journalStreamRemedy()}`
         : 'run `cleo cloud pull` first (it refuses to overwrite local changes), or `--force` to push this store as a labelled fork',
     );
   }
@@ -3324,7 +3351,7 @@ async function cloudSyncImpl(
     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`',
+      syncDisabledRemedy(),
     );
   }
   return { apiUrl: conn.apiUrl, streams, warnings: [...conn.warnings] };
@@ -3873,7 +3900,7 @@ async function restoreNexusVaultImpl(opts: NexusVaultRestoreOptions): Promise<Cl
           ? `this store changed since its last cloud sync (${changed.slice(0, 8).join(', ')}); restoring would overwrite those changes`
           : 'this store holds data that was never synced with this cloud snapshot; restoring would replace it',
         isJournalSnapshot(targetCp)
-          ? `to keep them: ${JOURNAL_STREAM_REMEDY}; or pass --force to restore anyway (a safety backup is taken first)`
+          ? `to keep them: ${journalStreamRemedy()}; or pass --force to restore anyway (a safety backup is taken first)`
           : 'run `cleo cloud push` to keep them (as a fork with --force if another device pushed since), or pass --force to restore anyway (a safety backup is taken first)',
       );
     }
@@ -4088,7 +4115,7 @@ async function restoreNexusVaultImpl(opts: NexusVaultRestoreOptions): Promise<Cl
     if (keptUnsynced.length > 0) {
       warnings.push({
         code: 'W_NEXUS_VAULT_KEPT_LOCAL',
-        message: `kept ${keptUnsynced.length} local file(s) this store never synced with the cloud, in place of the snapshot's copies: ${keptUnsynced.slice(0, 8).join(', ')}${keptUnsynced.length > 8 ? ', …' : ''}; ${isJournalSnapshot(restored.checkpoint) ? `to keep them: ${JOURNAL_STREAM_REMEDY}; or` : 'push to keep them, or'} pull with --force to take the snapshot's`,
+        message: `kept ${keptUnsynced.length} local file(s) this store never synced with the cloud, in place of the snapshot's copies: ${keptUnsynced.slice(0, 8).join(', ')}${keptUnsynced.length > 8 ? ', …' : ''}; ${isJournalSnapshot(restored.checkpoint) ? `to keep them: ${journalStreamRemedy()}; or` : 'push to keep them, or'} pull with --force to take the snapshot's`,
       });
     }
     saveSynced(conn, t, restored.checkpoint, {
@@ -4164,7 +4191,7 @@ async function nexusVaultStatusImpl(
   if (headCp && isJournalSnapshot(headCp) && pendingChanges.length > 0) {
     warnings.push({
       code: 'W_NEXUS_VAULT_JOURNAL_STREAM',
-      message: `${pendingChanges.length} table(s) changed since the last sync, and the newest snapshot ${headCp.checkpointId} is checkpoint/v3 (the change journal writes this stream): ${JOURNAL_STREAM_REMEDY}`,
+      message: `${pendingChanges.length} table(s) changed since the last sync, and the newest snapshot ${headCp.checkpointId} is checkpoint/v3 (the change journal writes this stream): ${journalStreamRemedy()}`,
     });
   }
   return {
@@ -4445,11 +4472,11 @@ async function verifyNexusVaultImpl(
             ? '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}`
+                ? `this machine changed since the newest snapshot, which is checkpoint/v3 (the change journal writes this stream): ${journalStreamRemedy()}`
                 : 'run `cleo cloud push` to back up the local changes'
               : verdict === 'diverged'
                 ? journalHead
-                  ? `both sides changed, and the newest snapshot is checkpoint/v3 (the change journal writes this stream): ${JOURNAL_STREAM_REMEDY}; or \`cleo cloud pull --force\` to take the cloud (a safety backup is taken first)`
+                  ? `both sides changed, and the newest snapshot is checkpoint/v3 (the change journal writes this stream): ${journalStreamRemedy()}; or \`cleo cloud pull --force\` to take the cloud (a safety backup is taken first)`
                   : 'both sides changed: run `cleo cloud push --force` to keep this machine (a labelled fork), or `cleo cloud pull --force` to take the cloud (a safety backup is taken first)'
                 : verdict === 'empty'
                   ? 'run `cleo cloud push` to make the first snapshot'