refactor(sync): cloud sync classifies skipped legs by a typed refusal kind (T13315)

.changeset/t13315-typed-skip.md

@@ -0,0 +1,19 @@
+---
+id: t13315-typed-skip
+tasks: [T13315]
+kind: refactor
+summary: "`cleo cloud sync` sorts skipped legs by a typed refusal kind instead of matching refusal message text"
+---
+
+`cleo cloud sync` used to decide whether a push or pull leg was skipped or refused by matching the refusal's message
+against regexes. Rewording a refusal would silently turn a skip (`sync.push` off, no genesis yet) into a reported refusal.
+
+- **Push results** (`PushStreamReport`) now carry `refusedKind`, which is one of:
+  - `schema-missing`
+  - `push-off`
+  - `no-genesis`
+  - `genesis-pending`
+  - `store-behind`
+- **Pull results** (`PullStreamReport`) now carry `refusedKind`, which is `pull-off` or `segment`.
+- **Classification** is done by `classifySyncLegs`, using only these kinds. The message regexes are removed. A stream is
+  `disabled` when both legs are off. The pull position is reported unless the pull leg was skipped.

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

@@ -122,6 +122,7 @@ import { hasUnsyncedNexusBackup, runNexusFirstRun } from '../nexus-first-run.js'
 import { linkProjectToNexus } from '../nexus-link.js';
 import { listNexusNamedProjects, resolveNexusProjectRef } from '../nexus-project-names.js';
 import {
+  classifySyncLegs,
   cloudSync,
   enableSyncPush,
   nexusVaultStatus,
@@ -5950,3 +5951,40 @@ describe('sync enable push: the genesis checkpoint (T12343 S4-1b)', () => {
     expect(fake.stream(STREAM).checkpoints).toHaveLength(0);
   });
 });
+
+describe('cloud sync leg classification (T13315)', () => {
+  const none = { refused: null, refusedKind: null } as const;
+
+  it('classifies by refusal kind, so a reworded skip message is still a skip', () => {
+    const r = classifySyncLegs(
+      { refused: 'push is switched off for this store', refusedKind: 'push-off' },
+      { refused: 'pulling is disabled', refusedKind: 'pull-off' },
+    );
+    expect(r).toEqual({
+      skipped: ['push is switched off for this store', 'pulling is disabled'],
+      refused: [],
+      disabled: true,
+      pullSkipped: true,
+    });
+  });
+
+  it('push before genesis is skipped; a store behind the server and a refused segment are refusals', () => {
+    expect(
+      classifySyncLegs({ refused: 'no cut yet', refusedKind: 'no-genesis' }, none),
+    ).toMatchObject({ skipped: ['no cut yet'], refused: [], disabled: false, pullSkipped: false });
+    expect(
+      classifySyncLegs({ refused: 'not stored', refusedKind: 'genesis-pending' }, none).skipped,
+    ).toEqual(['not stored']);
+    expect(
+      classifySyncLegs(
+        { refused: 'behind', refusedKind: 'store-behind' },
+        { refused: 'bad segment', refusedKind: 'segment' },
+      ),
+    ).toMatchObject({ skipped: [], refused: ['behind', 'bad segment'], disabled: false });
+    // A message that reads like a skip but carries no skip kind is a refusal.
+    expect(
+      classifySyncLegs({ refused: 'sync.push is off', refusedKind: 'schema-missing' }, none)
+        .refused,
+    ).toEqual(['sync.push is off']);
+  });
+});

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

@@ -118,7 +118,7 @@ import {
   readGenesisCut,
 } from '../store/sync/genesis.js';
 import { type PullStreamReport, pullStream, readStreamCursor } from '../store/sync/pull.js';
-import { type PushStreamReport, pushStream } from '../store/sync/push.js';
+import { type PushRefusal, type PushStreamReport, pushStream } from '../store/sync/push.js';
 import { replayPinOf } from '../store/sync/replay-pin.js';
 import {
   activeReplica,
@@ -2061,13 +2061,52 @@ const NOT_ATTACHED_CODES: ReadonlySet<string> = new Set([
   'E_NEXUS_NOT_A_PROJECT',
 ]);
 
-/** 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/,
-];
+/** Push refusals that mean "nothing to do here", not a failure (LOW-3 on #1962). */
+const PUSH_SKIPS: ReadonlySet<PushRefusal> = new Set(['push-off', 'no-genesis', 'genesis-pending']);
+
+/** How one stream's push and pull legs ended, classified by refusal kind (T13315). */
+export interface SyncLegOutcome {
+  /** Refusals that mean "nothing to do here": a flag off, or push before this store's genesis. */
+  readonly skipped: string[];
+  /** Every other refusal. */
+  readonly refused: string[];
+  /** Both flags off: the stream is disabled. */
+  readonly disabled: boolean;
+  /** The pull leg was skipped, so its position says nothing. */
+  readonly pullSkipped: boolean;
+}
+
+/**
+ * Classify a stream's push and pull results (T13315). Classification uses
+ * each report's typed `refusedKind`, never its message, so rewording a
+ * refusal never turns a skip into a refusal.
+ *
+ * @param push - The push leg's `refused` and `refusedKind`.
+ * @param pull - The pull leg's `refused` and `refusedKind`.
+ * @returns Skipped and refused messages, and whether the stream is disabled.
+ */
+export function classifySyncLegs(
+  push: Pick<PushStreamReport, 'refused' | 'refusedKind'>,
+  pull: Pick<PullStreamReport, 'refused' | 'refusedKind'>,
+): SyncLegOutcome {
+  const pushSkipped = push.refusedKind !== null && PUSH_SKIPS.has(push.refusedKind);
+  const pullSkipped = pull.refusedKind === 'pull-off';
+  const skipped: string[] = [];
+  const refused: string[] = [];
+  for (const [msg, skip] of [
+    [push.refused, pushSkipped],
+    [pull.refused, pullSkipped],
+  ] as const) {
+    if (msg === null) continue;
+    (skip ? skipped : refused).push(msg);
+  }
+  return {
+    skipped,
+    refused,
+    disabled: push.refusedKind === 'push-off' && pullSkipped,
+    pullSkipped,
+  };
+}
 
 /** The code and message of a failure, for a stream result. */
 function failureText(err: unknown): string {
@@ -2126,22 +2165,18 @@ async function syncOneStream(
       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 legs = classifySyncLegs(push, pull);
+  const { skipped, refused } = legs;
   const apply = pull.apply;
   return syncStreamResult(scope, {
     streamId,
-    status:
-      pushOff && pullOff
-        ? 'disabled'
-        : refused.length > 0
-          ? 'refused'
-          : push.clockAhead
-            ? 'paused'
-            : 'synced',
+    status: legs.disabled
+      ? 'disabled'
+      : refused.length > 0
+        ? 'refused'
+        : push.clockAhead
+          ? 'paused'
+          : 'synced',
     refused: refused.length > 0 ? refused.join('; ') : null,
     skipped,
     sealed: push.sealed,
@@ -2154,8 +2189,8 @@ 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 || !isSkip(pull.refused) ? pull.after : null,
-    head: pull.refused === null || !isSkip(pull.refused) ? pull.head : null,
+    after: legs.pullSkipped ? null : pull.after,
+    head: legs.pullSkipped ? null : pull.head,
   });
 }
 

packages/core/src/store/sync/pull.ts

@@ -112,6 +112,12 @@ export interface PullStreamReport {
    * segment was staged and applied, and the cursor stops just before it.
    */
   readonly refused: string | null;
+  /**
+   * The kind of `refused` (callers branch on this; the message may be
+   * reworded): `pull-off`, or `segment` for a refused segment. Null when
+   * nothing was refused.
+   */
+  readonly refusedKind: 'pull-off' | 'segment' | null;
   /** Segments received, vault deltas passed over, transactions staged, re-deliveries skipped. */
   readonly segments: number;
   readonly vaultDeltas: number;
@@ -267,6 +273,7 @@ export async function pullStream(
     return {
       stream: o.stream,
       refused: 'sync.pull is off',
+      refusedKind: 'pull-off',
       segments: 0,
       vaultDeltas: 0,
       staged: 0,
@@ -355,6 +362,7 @@ export async function pullStream(
   return {
     stream: o.stream,
     refused: refusedSegment,
+    refusedKind: refusedSegment === null ? null : 'segment',
     segments,
     vaultDeltas,
     staged,

packages/core/src/store/sync/push.ts

@@ -88,11 +88,29 @@ export interface PushStreamOptions {
   readonly maxDriftMs?: number;
 }
 
+/**
+ * Why {@link pushStream} sent nothing, as a kind callers branch on (the
+ * message in `refused` is for people and may be reworded):
+ * - `schema-missing`: the sync schema is not installed;
+ * - `push-off`: `sync.push` is off;
+ * - `no-genesis`: this store has no genesis cut on the stream;
+ * - `genesis-pending`: the genesis checkpoint is not stored yet;
+ * - `store-behind`: the server holds a later segment of this replica.
+ */
+export type PushRefusal =
+  | 'schema-missing'
+  | 'push-off'
+  | 'no-genesis'
+  | 'genesis-pending'
+  | 'store-behind';
+
 /** What {@link pushStream} did. */
 export interface PushStreamReport {
   readonly stream: string;
   /** Why nothing was sent, or null. */
   readonly refused: string | null;
+  /** The kind of `refused`, or null when nothing was refused. */
+  readonly refusedKind: PushRefusal | null;
   /** Push paused because this device's clock is ahead of the server's. */
   readonly clockAhead: boolean;
   /** Transactions sealed, segments packed and persisted, segments the server stored (`duplicates` of them retries). */
@@ -107,6 +125,7 @@ export interface PushStreamReport {
 const empty = (stream: string, fields: Partial<PushStreamReport>): PushStreamReport => ({
   stream,
   refused: null,
+  refusedKind: null,
   clockAhead: false,
   sealed: 0,
   built: 0,
@@ -145,16 +164,22 @@ export async function pushStream(
   const env = o.env ?? process.env;
   const now = o.now ?? Date.now;
   if (!hasTable(db, '_sync_segment'))
-    return empty(o.stream, { refused: 'sync schema not installed' });
-  if (!isSyncFlagOn(db, 'sync.push', env)) return empty(o.stream, { refused: 'sync.push is off' });
+    return empty(o.stream, {
+      refused: 'sync schema not installed',
+      refusedKind: 'schema-missing',
+    });
+  if (!isSyncFlagOn(db, 'sync.push', env))
+    return empty(o.stream, { refused: 'sync.push is off', refusedKind: 'push-off' });
   if (genesisCutOf(db, o.stream) === undefined) {
     return empty(o.stream, {
       refused: `${o.stream} has no genesis cut on this store: run \`cleo sync enable push\``,
+      refusedKind: 'no-genesis',
     });
   }
   if (genesisPending(db, o.stream)) {
     return empty(o.stream, {
       refused: `${o.stream}'s genesis checkpoint is not stored yet: run \`cleo sync enable push\` again`,
+      refusedKind: 'genesis-pending',
     });
   }
   const atIso = new Date(now()).toISOString();
@@ -171,6 +196,7 @@ export async function pushStream(
   if (o.serverLastReplicaSeq !== null && (local === null || o.serverLastReplicaSeq > local)) {
     return empty(o.stream, {
       refused: `the server holds replicaSeq ${o.serverLastReplicaSeq} of replica ${o.replica} on ${o.stream}, but this store persisted only ${local ?? 'none'}: the store is behind (restored or copied) and must rebind to a new replica (T12753)`,
+      refusedKind: 'store-behind',
     });
   }