diff --git a/.changeset/t13315-typed-skip.md b/.changeset/t13315-typed-skip.md new file mode 100644 index 0000000000..5234810ec2 --- /dev/null +++ b/.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. diff --git a/packages/core/src/cloud/__tests__/nexus-vault.test.ts b/packages/core/src/cloud/__tests__/nexus-vault.test.ts index 9bdb6e1a8d..ffe108cb9c 100644 --- a/packages/core/src/cloud/__tests__/nexus-vault.test.ts +++ b/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']); + }); +}); diff --git a/packages/core/src/cloud/nexus-vault.ts b/packages/core/src/cloud/nexus-vault.ts index c58ec9ff5e..7fef03b3b9 100644 --- a/packages/core/src/cloud/nexus-vault.ts +++ b/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 = 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 = 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, + pull: Pick, +): 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, }); } diff --git a/packages/core/src/store/sync/pull.ts b/packages/core/src/store/sync/pull.ts index 391f0b3edc..4ac838ff2e 100644 --- a/packages/core/src/store/sync/pull.ts +++ b/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, diff --git a/packages/core/src/store/sync/push.ts b/packages/core/src/store/sync/push.ts index 3db8df7a9a..5073802521 100644 --- a/packages/core/src/store/sync/push.ts +++ b/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 => ({ 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', }); }