diff --git a/.env.example b/.env.example index 005bd06..55459ec 100644 --- a/.env.example +++ b/.env.example @@ -20,6 +20,9 @@ MANIFEST_DB=countly_drill # where progress state lives: mig_ranges + m LEDGER_RUN_ID=migration-1 # stable resume key — keep it the same across restarts #POD_ID=pod-1 # unique per instance when running multiple pods #EXIT_ON_COMPLETE=true # exit 0 when all chunks are done (one-shot orchestration) +#LEDGER_START_PAUSED=true # deploy now, start later: pods come up, serve the + # dashboard and wait (nothing read, mapped or indexed) + # until Start is pressed once for the whole run # ─── Sizing (see the UI's Configuration card for guidance) ─── #LEDGER_CHUNK_DOCS_TARGET=2000000 diff --git a/k8s/migration.yaml b/k8s/migration.yaml index 33ce35a..11dc186 100644 --- a/k8s/migration.yaml +++ b/k8s/migration.yaml @@ -32,6 +32,9 @@ data: CLICKHOUSE_TABLE: "drill_events" # Stable resume key — keep identical across ALL pods and restarts. LEDGER_RUN_ID: "migration-1" + # Deploy now, start later: pods serve the dashboard and wait until Start + # is pressed once (the gate lives in the ledger, so it covers every pod). + #LEDGER_START_PAUSED: "true" # On a replica set, offload the primary (exact reads — source is frozen): #MONGO_READ_PREFERENCE: "secondaryPreferred" --- diff --git a/src/config/loader.ts b/src/config/loader.ts index ad627b0..6b436d9 100644 --- a/src/config/loader.ts +++ b/src/config/loader.ts @@ -25,6 +25,7 @@ function envToRawConfig(env: NodeJS.ProcessEnv) { maxChunkDays: env.LEDGER_MAX_CHUNK_DAYS, cdUpperBoundMs: env.LEDGER_CD_UPPER_BOUND, captureTransformErrors: env.LEDGER_CAPTURE_TRANSFORM_ERRORS, + startPaused: env.LEDGER_START_PAUSED, dryRun: env.DRY_RUN, dryRunSamplePct: env.DRY_RUN_SAMPLE_PCT, }, diff --git a/src/config/schema.ts b/src/config/schema.ts index 6f6b3f1..9605a2f 100644 --- a/src/config/schema.ts +++ b/src/config/schema.ts @@ -76,6 +76,12 @@ export const configSchema = z.object({ } return ms; }), + // Deploy now, start later: hold every pod BEFORE mapping until + // an operator opens the run's start gate (dashboard Start button, + // or POST /control/resume). The gate lives in the ledger, so one + // click starts the whole fleet, pods that join later start + // immediately, and a pod that restarts after Start stays started. + startPaused: booleanFromEnv.default(false), // Dry run: sampled rehearsal against a Null-engine clone. dryRun: booleanFromEnv.default(false), dryRunSamplePct: numberFromEnv.default(2).pipe(z.number().min(0.1).max(5)), diff --git a/src/http/ledger-viz-route.ts b/src/http/ledger-viz-route.ts index 9fbefcb..8cfd681 100644 --- a/src/http/ledger-viz-route.ts +++ b/src/http/ledger-viz-route.ts @@ -819,7 +819,10 @@ async function tick() { var hint = document.getElementById('pause-hint'); if (isPaused) { hint.style.display = ''; - hint.textContent = '\u23f8 ENGINE PAUSED' + + hint.textContent = stats.pauseReason === 'not-started' + ? '\u23f8 NOT STARTED \u2014 deployed and waiting. Nothing has been read, mapped or indexed yet; ' + + 'run preflight, build indexes and rehearse first, then click Start to begin the run (all pods).' + : '\u23f8 ENGINE PAUSED' + (stats.pauseReason === 'breaker-transient' ? ' (backend outage \u2014 auto-resume armed)' : stats.pauseReason === 'breaker-data' ? ' (systematic data problem \u2014 needs you)' : ' (by operator)') + ' \u2014 Retry / Replay / Waive only QUEUE work; click Resume to process it.'; @@ -828,7 +831,7 @@ async function tick() { if (prBtn) { if (isPaused) { prBtn.dataset.action = 'resume'; - prBtn.innerHTML = '\u25b6 Resume'; + prBtn.innerHTML = stats.pauseReason === 'not-started' ? '\u25b6 Start' : '\u25b6 Resume'; prBtn.classList.add('primary'); prBtn.disabled = false; } else if (stats.status === 'running') { diff --git a/src/runtime/chunk-orchestrator.ts b/src/runtime/chunk-orchestrator.ts index 319efae..2dec4da 100644 --- a/src/runtime/chunk-orchestrator.ts +++ b/src/runtime/chunk-orchestrator.ts @@ -138,7 +138,7 @@ export class ChunkOrchestrator { private consecutiveFailed = 0; private sourceShrankChunks = 0; private streakHadPermanent = false; - private pauseReason: 'operator' | 'breaker-transient' | 'breaker-data' | null = null; + private pauseReason: 'operator' | 'not-started' | 'breaker-transient' | 'breaker-data' | null = null; private probeOkStreak = 0; private autoResuming = false; private resumeProbeTimer: NodeJS.Timeout | null = null; @@ -172,7 +172,7 @@ export class ChunkOrchestrator { // ------------------------------------------------------------------------- stopAfterChunk(): void { this.stopping = true; } - pause(reason: 'operator' | 'breaker-transient' | 'breaker-data' = 'operator'): void { + pause(reason: 'operator' | 'not-started' | 'breaker-transient' | 'breaker-data' = 'operator'): void { this.paused = true; this.pauseReason = reason; if (this.status === 'running') this.status = 'paused'; @@ -213,6 +213,35 @@ export class ChunkOrchestrator { this.finishedAt = 0; const { config } = this.d; + // ── START GATE ──────────────────────────────────────────────────────── + // Deploy now, start later. Held BEFORE mapping deliberately: mapping + // builds the {cd:1,_id:1} index on the source and cuts the chunk grid, + // neither of which a not-yet-started run should be doing. The HTTP + // surface is already listening, so preflight, index builds and the dry + // run are all available from the dashboard while the pod waits here. + if (config.ledger.startPaused) { + // Effective run id: a dry run has its own gate (`-dry`), so + // starting a rehearsal cannot silently pre-authorise the real run. + const gateRunId = this.runId; + if (!(await this.d.ledger.isStartGateOpen(gateRunId).catch(() => false))) { + this.pause('not-started'); + this.logger.warn( + { runId: gateRunId }, + 'LEDGER_START_PAUSED: holding before mapping — press Start on the dashboard (POST /control/resume) to begin', + ); + while (this.paused && !this.stopping) { + if (await this.d.ledger.isStartGateOpen(gateRunId).catch(() => false)) { + this.logger.info({ runId: gateRunId }, 'Start gate opened — beginning the run'); + this.resume(); + break; + } + await sleep(3_000); + } + if (this.stopping) { this.status = 'stopped'; return; } + this.startedAt = Date.now(); // the run starts when it was started + } + } + // Transient-outage self-healing: only acts while paused with reason // 'breaker-transient' (backend outage tripped the failure breaker) — // every other pause stays owned by the operator. @@ -263,6 +292,10 @@ export class ChunkOrchestrator { let mapPass = 0; for (;;) { + if (this.stopping) break; + // Honour a pause here too: without this, a pause issued during a long + // mapping or top-up pass is not observed until claiming begins. + while (this.paused && !this.stopping) await sleep(1_000); if (this.stopping) break; const storedBound = await this.d.ledger.getStoredBound(this.runId).catch(() => null); if (storedBound !== null) { diff --git a/src/runtime/ledger-engine.ts b/src/runtime/ledger-engine.ts index 1c2ffab..07c7d57 100644 --- a/src/runtime/ledger-engine.ts +++ b/src/runtime/ledger-engine.ts @@ -221,7 +221,18 @@ export async function runLedgerEngine(config: Config, logger: Logger): Promise orchestrator.getReport()); app.post('/control/pause', async () => { orchestrator.pause(); return { status: orchestrator.getStatus() }; }); - app.post('/control/resume', async () => { orchestrator.resume(); return { status: orchestrator.getStatus() }; }); + // Resume doubles as Start: opening the gate is what releases every pod + // held by LEDGER_START_PAUSED, not just the one serving this request. + app.post('/control/resume', async () => { + if (config.ledger.startPaused) { + const gateRunId = config.ledger.dryRun ? `${config.ledger.runId}-dry` : config.ledger.runId; + await ledger.openStartGate(gateRunId, config.worker.podId).catch((err) => { + logger.error({ err }, 'Failed to open the start gate — other pods stay held'); + }); + } + orchestrator.resume(); + return { status: orchestrator.getStatus() }; + }); // Replay runs in the background: a mass DLQ (systematic failure on a // 10B-doc run) can hold millions of entries — not one HTTP request's work. const replayState: { status: string; result: Record | null; error: string | null } = diff --git a/src/state/ledger-store.ts b/src/state/ledger-store.ts index 87d519a..d275a4b 100644 --- a/src/state/ledger-store.ts +++ b/src/state/ledger-store.ts @@ -475,7 +475,10 @@ export class LedgerStore { * wins as the source of truth when present; this store only fills in * when env is unset, and pods re-read it at every map pass. */ - private rc(): Collection<{ _id: string; cd_upper_bound_ms: number; set_at: Date; set_by: string }> { + private rc(): Collection<{ + _id: string; cd_upper_bound_ms: number; set_at: Date; set_by: string; + start_gate_open?: boolean; start_gate_opened_at?: Date; start_gate_opened_by?: string; + }> { if (!this.coll) throw new Error('LedgerStore not connected'); return this.client.db(this.dbName).collection('mig_run_config'); } @@ -512,6 +515,29 @@ export class LedgerStore { }; } + /** + * Start gate (mig_run_config): with LEDGER_START_PAUSED set, pods hold + * before mapping until this is opened once for the run. It lives with the + * run rather than in one pod's memory, so a single Start covers every pod + * (including ones that join afterwards) and survives restarts. + */ + async isStartGateOpen(runId: string): Promise { + const doc = await this.rc().findOne({ _id: runId }); + return doc?.start_gate_open === true; + } + + async openStartGate(runId: string, openedBy: string): Promise { + await this.rc().updateOne( + { _id: runId }, + [{ $set: { + start_gate_open: true, + start_gate_opened_at: { $ifNull: ['$start_gate_opened_at', '$$NOW'] }, + start_gate_opened_by: { $ifNull: ['$start_gate_opened_by', openedBy] }, + } }] as never, + { upsert: true }, + ); + } + async getStoredBound(runId: string): Promise { const doc = await this.rc().findOne({ _id: runId }); return doc?.cd_upper_bound_ms ?? null;