Add TTL tracking, danger zone controls, and cluster-wide progress enhancements

kanwarujjaval · kanwarujjaval · commit affef02f3561 · 2026-03-26T15:27:22.000+05:30
- Introduced TTL tracking for Redis locks with detailed expiration handling in `/viz` route.
- Added "Danger Zone" actions for clearing MongoDB and Redis migration state, with confirmation prompts.
- Enhanced `/stats` route to compute cluster-wide progress and ETA based on multi-pod data.
- Enabled upsert functionality in MongoDB writes to better handle duplicates.
- Improved run management by cleaning old run data before starting fresh.
- Refined progress display, throughput calculations, and cluster-wide metrics visibility in multi-pod mode.
diff --git a/src/http/control-route.ts b/src/http/control-route.ts
@@ -6,6 +6,7 @@ import type { MongoReader } from '../source/mongo-reader.ts';
 import type { ClickHouseWriter } from '../target/clickhouse-writer.ts';
 import type { GlobalProgress } from '../state/global-progress.ts';
 import type { CollectionLock } from '../state/collection-lock.ts';
+import type { RedisHotState } from '../state/redis-hot-state.ts';
 import { transformBatch } from '../transform/normalize.ts';
 import { SkipCounter } from '../transform/skip-reasons.ts';
 import { deserializeCursor } from '../types/cursor.ts';
@@ -28,6 +29,7 @@ export interface ControlDeps {
   chWriter: ClickHouseWriter;
   globalProgress?: GlobalProgress;
   collectionLock?: CollectionLock;
+  redisState?: RedisHotState;
 }
 
 interface GcRequestBody {
@@ -303,4 +305,56 @@ export function registerControlRoutes(app: FastifyInstance, deps: ControlDeps):
     orchestrator.stopAfterBatch();
     return reply.status(200).send({ ok: true, message: 'Drain initiated — finishing current batch then releasing locks' });
   });
+
+  // ── DANGER ZONE ────────────────────────────────────────────────────
+
+  // POST /control/danger/clear-mongodb - drop all migration state from MongoDB
+  app.post('/control/danger/clear-mongodb', async (_request, reply) => {
+    const { manifestStore: ms } = deps;
+    try {
+      const runs = await ms.listRuns({ limit: 1000 });
+      let totalDeleted = 0;
+      for (const run of runs.runs) {
+        totalDeleted += await ms.deleteRunData(run.run_id);
+      }
+      // Also delete the run records themselves
+      const db = (ms as any).client.db((ms as any).dbName);
+      const runResult = await db.collection('mig_runs').deleteMany({});
+      totalDeleted += runResult.deletedCount ?? 0;
+      return reply.status(200).send({
+        ok: true,
+        message: `Cleared MongoDB migration state: ${totalDeleted} records deleted`,
+        deletedRecords: totalDeleted,
+      });
+    } catch (err) {
+      return reply.status(500).send({ ok: false, error: err instanceof Error ? err.message : String(err) });
+    }
+  });
+
+  // POST /control/danger/clear-redis - flush all migration keys from Redis
+  app.post('/control/danger/clear-redis', async (_request, reply) => {
+    const { redisState: rs } = deps;
+    if (!rs) {
+      return reply.status(400).send({ ok: false, error: 'Redis not available' });
+    }
+    try {
+      const redis = rs.getRedisClient();
+      const keys = await (async () => {
+        const found: string[] = [];
+        const stream = redis.scanStream({ match: 'mig:*', count: 500 });
+        for await (const batch of stream) found.push(...(batch as string[]));
+        return found;
+      })();
+      if (keys.length > 0) {
+        await redis.unlink(...keys);
+      }
+      return reply.status(200).send({
+        ok: true,
+        message: `Cleared Redis migration state: ${keys.length} keys deleted`,
+        deletedKeys: keys.length,
+      });
+    } catch (err) {
+      return reply.status(500).send({ ok: false, error: err instanceof Error ? err.message : String(err) });
+    }
+  });
 }
diff --git a/src/http/stats-route.ts b/src/http/stats-route.ts
@@ -133,10 +133,8 @@ export function registerStatsRoute(app: FastifyInstance, deps: StatsDeps): void
     const overallPct = totalEstimated > 0
       ? Math.min(100, Math.round((overallDocsRead / totalEstimated) * 100)) : 0;
 
-    // ETA based on elapsed time and progress percentage
-    const etaMs = overallPct > 0 && overallPct < 100 && elapsedMs > 0
-      ? Math.round(((Date.now() - startedAt.getTime()) / overallPct) * (100 - overallPct))
-      : null;
+    // ETA computed later using cluster-wide data
+    let etaMs: number | null = null;
 
     // Commands from Redis (best-effort)
     let commands: CommandFlags = {};
@@ -240,20 +238,58 @@ export function registerStatsRoute(app: FastifyInstance, deps: StatsDeps): void
       };
     });
 
+    // ── Compute cluster-wide aggregates for summary ─────────────────
+    const clusterDocsRead = clusterData
+      ? mergedCollectionProgress.reduce((s, c) => s + (c.docsRead ?? 0), 0)
+      : overallDocsRead;
+    const clusterEstimated = clusterData
+      ? mergedCollectionProgress.reduce((s, c) => s + (c.estimated ?? 0), 0) || totalEstimated
+      : totalEstimated;
+    const clusterPct = clusterEstimated > 0
+      ? Math.min(100, Math.round((clusterDocsRead / clusterEstimated) * 100))
+      : overallPct;
+    const clusterDone = clusterData
+      ? mergedCollectionProgress.filter(c => c.status === 'completed' || c.status === 'skipped').length
+      : progress.completedCollections + progress.skippedCollections;
+    const clusterFailed = clusterData
+      ? mergedCollectionProgress.filter(c => c.status === 'failed').length
+      : progress.failedCollections;
+    const clusterProcessing = clusterData
+      ? mergedCollectionProgress.filter(c => c.status === 'processing').length
+      : (progress.currentCollection ? 1 : 0);
+    const clusterTotal = clusterData
+      ? (progress.totalCollections || mergedCollectionProgress.length)
+      : progress.totalCollections;
+
+    // ETA based on cluster-wide progress
+    const serviceElapsedMs = Date.now() - startedAt.getTime();
+    etaMs = clusterPct > 0 && clusterPct < 100 && serviceElapsedMs > 0
+      ? Math.round((serviceElapsedMs / clusterPct) * (100 - clusterPct))
+      : null;
+
     const payload = {
-      // ── Quick-glance summary ──────────────────────────────────────────
+      // ── Quick-glance summary (cluster-wide when multi-pod) ──────────
       summary: {
-        overall: progressBar(overallPct),
-        overallPct,
+        overall: progressBar(clusterPct),
+        overallPct: clusterPct,
         currentCollection: progress.currentCollection
           ? `${progress.currentCollection}  ${progressBar(currentCollPct)}`
           : 'idle',
         currentCollectionPct: currentCollPct,
-        collections: `${progress.completedCollections}/${progress.totalCollections} done`
-          + (progress.failedCollections > 0 ? `, ${progress.failedCollections} failed` : '')
-          + (progress.skippedCollections > 0 ? `, ${progress.skippedCollections} skipped` : ''),
-        docsProgress: `${fmtNum(overallDocsRead)} / ~${fmtNum(totalEstimated)} docs`,
-        throughput: `${fmtNum(Math.round(batchStats?.docsPerSecond ?? 0))} docs/s`,
+        collections: `${clusterDone}/${clusterTotal} done`
+          + (clusterFailed > 0 ? `, ${clusterFailed} failed` : '')
+          + (clusterProcessing > 0 ? `, ${clusterProcessing} processing` : ''),
+        docsProgress: `${fmtNum(clusterDocsRead)} / ~${fmtNum(clusterEstimated)} docs`,
+        throughput: (() => {
+          // In multi-pod mode, compute cluster-wide throughput from total docs / uptime
+          const clusterDocs = clusterData
+            ? mergedCollectionProgress.reduce((s, c) => s + (c.docsRead ?? 0), 0)
+            : 0;
+          if (clusterDocs > 0 && uptimeSec > 0) {
+            return `${fmtNum(Math.round(clusterDocs / uptimeSec))} docs/s`;
+          }
+          return `${fmtNum(Math.round(batchStats?.docsPerSecond ?? 0))} docs/s`;
+        })(),
         elapsed: (() => {
           const isTerminal = runnerStatus === 'completed' || runnerStatus === 'failed' || runnerStatus === 'stopped';
           if (isTerminal && frozenElapsedMs === null) {
diff --git a/src/http/viz-route.ts b/src/http/viz-route.ts
@@ -295,7 +295,7 @@ const DASHBOARD_HTML = /* html */ `<!DOCTYPE html>
   <h3 style="margin-top:12px">Active Locks <span id="lock-count" style="font-weight:400;color:var(--text2)"></span></h3>
   <div class="coll-scroll" style="max-height:160px">
     <table class="coll-table">
-      <thead><tr><th>Collection</th><th>Pod</th><th>Acquired</th><th>Action</th></tr></thead>
+      <thead><tr><th>Collection</th><th>Pod</th><th>Acquired</th><th>Expires In</th><th>Action</th></tr></thead>
       <tbody id="lock-body"></tbody>
     </table>
   </div>
@@ -346,6 +346,17 @@ const DASHBOARD_HTML = /* html */ `<!DOCTYPE html>
   </div>
 </div>
 
+<!-- Danger Zone -->
+<div class="card full" style="margin-top:24px;border-color:var(--red)">
+  <h3 style="color:var(--red)">Danger Zone</h3>
+  <p style="font-size:11px;color:var(--text2);margin-bottom:12px">These actions are irreversible and will destroy migration state. Use with caution.</p>
+  <div style="display:flex;gap:12px;flex-wrap:wrap">
+    <button class="btn" id="btn-clear-mongo" style="padding:6px 16px;font-size:12px;border-color:var(--red);color:var(--red)">Clear MongoDB State</button>
+    <button class="btn" id="btn-clear-redis" style="padding:6px 16px;font-size:12px;border-color:var(--red);color:var(--red)">Clear Redis State</button>
+  </div>
+  <div id="danger-msg" style="margin-top:8px;font-size:11px"></div>
+</div>
+
 <script>
 (function() {
   'use strict';
@@ -386,6 +397,30 @@ const DASHBOARD_HTML = /* html */ `<!DOCTYPE html>
   $('btn-global-resume').addEventListener('click', function() { doControl('global/resume'); });
   $('btn-global-stop').addEventListener('click', function() { doControl('global/stop'); });
 
+  // Danger zone buttons
+  $('btn-clear-mongo').addEventListener('click', function() {
+    if (!confirm('WARNING: This will permanently delete ALL migration runs, batches, and events from MongoDB. This cannot be undone. Continue?')) return;
+    var msg = $('danger-msg');
+    fetch('/control/danger/clear-mongodb', { method: 'POST' })
+      .then(function(r) { return r.json(); })
+      .then(function(d) {
+        msg.textContent = d.ok ? 'MongoDB cleared: ' + d.deletedRecords + ' records deleted' : 'Error: ' + d.error;
+        msg.style.color = d.ok ? 'var(--green)' : 'var(--red)';
+      })
+      .catch(function(e) { msg.textContent = 'Error: ' + e.message; msg.style.color = 'var(--red)'; });
+  });
+  $('btn-clear-redis').addEventListener('click', function() {
+    if (!confirm('WARNING: This will permanently delete ALL migration keys from Redis (cursors, bitmaps, live state, locks, heartbeats). This cannot be undone. Continue?')) return;
+    var msg = $('danger-msg');
+    fetch('/control/danger/clear-redis', { method: 'POST' })
+      .then(function(r) { return r.json(); })
+      .then(function(d) {
+        msg.textContent = d.ok ? 'Redis cleared: ' + d.deletedKeys + ' keys deleted' : 'Error: ' + d.error;
+        msg.style.color = d.ok ? 'var(--green)' : 'var(--red)';
+      })
+      .catch(function(e) { msg.textContent = 'Error: ' + e.message; msg.style.color = 'var(--red)'; });
+  });
+
   function statusClass(s) {
     if (s === 'running' || s === 'stopping') return 'status-running';
     if (s === 'waiting_for_index') return 'status-waiting_for_index';
@@ -402,7 +437,6 @@ const DASHBOARD_HTML = /* html */ `<!DOCTYPE html>
   }
 
   function truncHash(name) {
-    if (name && name.length > 50) return name.slice(0, 20) + '...' + name.slice(-8);
     return name || '-';
   }
 
@@ -421,6 +455,7 @@ const DASHBOARD_HTML = /* html */ `<!DOCTYPE html>
       filtered = cp;
     }
 
+    filtered.sort(function(a, b) { return (b.estimated || 0) - (a.estimated || 0); });
     if (filtered.length > 200) filtered = filtered.slice(0, 200);
 
     var tbody = $('coll-body');
@@ -447,13 +482,9 @@ const DASHBOARD_HTML = /* html */ `<!DOCTYPE html>
 
       // Collection name
       var td1 = document.createElement('td');
-      td1.style.maxWidth = '200px';
-      td1.style.overflow = 'hidden';
-      td1.style.textOverflow = 'ellipsis';
-      td1.style.whiteSpace = 'nowrap';
       td1.style.fontSize = '11px';
-      td1.setAttribute('title', c.collection || '');
-      td1.textContent = truncHash(c.collection);
+      td1.style.wordBreak = 'break-all';
+      td1.textContent = c.collection || '-';
       tr.appendChild(td1);
 
       // Status tag
@@ -841,7 +872,7 @@ const DASHBOARD_HTML = /* html */ `<!DOCTYPE html>
         if (locks.length === 0) {
           var noLockRow = document.createElement('tr');
           var noLockCell = document.createElement('td');
-          noLockCell.setAttribute('colspan', '4');
+          noLockCell.setAttribute('colspan', '5');
           noLockCell.className = 'muted';
           noLockCell.style.textAlign = 'center';
           noLockCell.textContent = 'No active locks';
@@ -864,6 +895,20 @@ const DASHBOARD_HTML = /* html */ `<!DOCTYPE html>
             lkTd3.style.cssText = 'font-size:10px;color:var(--text2)';
             lkTd3.textContent = lk.acquiredAt ? new Date(lk.acquiredAt).toLocaleTimeString() : '-';
             lkRow.appendChild(lkTd3);
+            var lkTdTtl = document.createElement('td');
+            lkTdTtl.style.cssText = 'font-size:10px;color:var(--text2)';
+            var ttl = lk.ttlSec;
+            if (ttl > 0) {
+              var ttlMin = Math.floor(ttl / 60);
+              var ttlSec = ttl % 60;
+              lkTdTtl.textContent = ttlMin > 0 ? ttlMin + 'm ' + ttlSec + 's' : ttlSec + 's';
+              if (ttl < 60) lkTdTtl.style.color = 'var(--red)';
+              else if (ttl < 120) lkTdTtl.style.color = 'var(--yellow)';
+            } else {
+              lkTdTtl.textContent = 'no TTL';
+              lkTdTtl.style.color = 'var(--red)';
+            }
+            lkRow.appendChild(lkTdTtl);
             var lkTd4 = document.createElement('td');
             var relBtn = document.createElement('button');
             relBtn.className = 'btn';
diff --git a/src/main.ts b/src/main.ts
@@ -229,6 +229,7 @@ async function main(): Promise<void> {
     chWriter,
     globalProgress,
     collectionLock,
+    redisState,
   });
 
   registerRunRoutes(app, {
diff --git a/src/runtime/resolve-run.ts b/src/runtime/resolve-run.ts
@@ -111,6 +111,9 @@ export async function resolveRun(opts: ResolveRunOpts): Promise<ResolvedRun> {
         const upperBoundId = serializeCursor(upperBound);
         const activeRun = await manifestStore.getActiveRun(sourceNs, targetTable);
         if (activeRun) {
+            // Clean old run data before starting fresh
+            const deleted = await manifestStore.deleteRunData(activeRun.run_id);
+            logger.info({ oldRunId: activeRun.run_id, deletedRecords: deleted }, 'Cleaned old run data for fresh start');
             await manifestStore.updateRunStatus(activeRun.run_id, 'completed');
         }
         await createNewRun(runId, upperBoundId);
@@ -123,6 +126,9 @@ export async function resolveRun(opts: ResolveRunOpts): Promise<ResolvedRun> {
         if (!activeRun) throw new Error('No existing run to clone from');
         const runId = randomUUID();
         const upperBoundId = activeRun.upper_bound_cursor;
+        // Clean old run data before starting fresh
+        const deleted = await manifestStore.deleteRunData(activeRun.run_id);
+        logger.info({ oldRunId: activeRun.run_id, deletedRecords: deleted }, 'Cleaned old run data for fresh start');
         await manifestStore.updateRunStatus(activeRun.run_id, 'completed');
         await createNewRun(runId, upperBoundId);
         logger.info({ runId, upperBoundId, sourceNs, targetTable }, 'Created new migration run (clone-run mode)');
diff --git a/src/state/async-batch-writer.ts b/src/state/async-batch-writer.ts
@@ -149,7 +149,7 @@ export class AsyncBatchWriter {
         }
 
         try {
-            // Bulk insert batch records
+            // Bulk upsert batch records (handles duplicates via update)
             await this.manifestStore.bulkInsertBatches(batchDocs);
 
             // Advance cursor to latest position
@@ -160,6 +160,7 @@ export class AsyncBatchWriter {
                 "Async flush completed",
             );
         } catch (err) {
+            const errMsg = err instanceof Error ? err.message : String(err);
             // Re-queue failed items (capped to prevent unbounded growth)
             const maxDepth = this.config.maxQueueDepth ?? 1000;
             const totalAfterRequeue = this.queue.length + items.length;
@@ -174,7 +175,7 @@ export class AsyncBatchWriter {
                 );
             }
             this.logger.warn(
-                { error: err instanceof Error ? err.message : String(err), count: items.length, queueDepth: this.queue.length },
+                { error: errMsg, count: items.length, queueDepth: this.queue.length },
                 "Async flush to MongoDB failed, re-queued",
             );
         } finally {
diff --git a/src/state/collection-lock.ts b/src/state/collection-lock.ts
@@ -11,6 +11,7 @@ export interface LockInfo {
     collectionName: string;
     podId: string;
     acquiredAt: string;
+    ttlSec: number;
 }
 
 export interface CollectionLockConfig {
@@ -224,19 +225,28 @@ export class CollectionLock {
         }
         if (keys.length === 0) return [];
 
-        const values = await this.redis.mget(...keys);
+        // Fetch values and TTLs in parallel via pipeline
+        const pipeline = this.redis.pipeline();
+        for (const key of keys) {
+            pipeline.get(key);
+            pipeline.ttl(key);
+        }
+        const results = await pipeline.exec();
+
         const prefix = `${this.config.keyPrefix}:lock:`;
         const locks: LockInfo[] = [];
 
         for (let i = 0; i < keys.length; i++) {
-            const val = values[i];
+            const val = results?.[i * 2]?.[1] as string | null;
+            const ttl = results?.[i * 2 + 1]?.[1] as number ?? -1;
             if (!val) continue;
             try {
                 const data = JSON.parse(val) as { podId: string; acquiredAt: string };
                 locks.push({
                     collectionName: keys[i].slice(prefix.length),
                     podId: data.podId,
                     acquiredAt: data.acquiredAt,
+                    ttlSec: ttl,
                 });
             } catch {
                 // skip malformed entries
diff --git a/src/state/manifest-store.ts b/src/state/manifest-store.ts
