Add LEDGER_START_PAUSED so a migration can be deployed stopped

.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

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"
 ---

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,
         },

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)),

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') {

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 (`<runId>-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) {

src/runtime/ledger-engine.ts

@@ -221,7 +221,18 @@ export async function runLedgerEngine(config: Config, logger: Logger): Promise<v
   });
   app.get('/report', async () => 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<string, unknown> | null; error: string | null } =

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<boolean> {
+    const doc = await this.rc().findOne({ _id: runId });
+    return doc?.start_gate_open === true;
+  }
+
+  async openStartGate(runId: string, openedBy: string): Promise<void> {
+    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<number | null> {
     const doc = await this.rc().findOne({ _id: runId });
     return doc?.cd_upper_bound_ms ?? null;