Updated readme and fixed missing completion time from status

Author: daniel-percona

config.yaml

@@ -118,8 +118,8 @@ validation:
 
   # --------------------------------------------------------------------------
   # full_validation: Controls the scope of CDC data validation.
-  #   - true:  (Default) Validates all operations (Inserts, Upserts, Updates, and Deletes).
-  #   - false: Validates ONLY Deletes. This is a massive performance optimization 
+  #   - true:  Validates all operations (Inserts, Upserts, Updates, and Deletes).
+  #   - false: (Default) Validates ONLY Deletes. This is a massive performance optimization 
   #            since Inserts and Updates are guaranteed by the source change stream 
   #            payload, whereas Deletes lose their full document payload.
   #

internal/cdc/cdc.go

@@ -427,14 +427,13 @@ func (m *CDCManager) hydrateAndFixOperations(ctx context.Context, ns string, bat
 	}
 
 	foundDocs := make(map[string]bson.M)
-	if err == nil {
-		defer cursor.Close(ctx)
-		for cursor.Next(ctx) {
-			var doc bson.M
-			if err := cursor.Decode(&doc); err == nil {
-				idStr := formatID(doc["_id"])
-				foundDocs[idStr] = doc
-			}
+
+	defer cursor.Close(ctx)
+	for cursor.Next(ctx) {
+		var doc bson.M
+		if err := cursor.Decode(&doc); err == nil {
+			idStr := formatID(doc["_id"])
+			foundDocs[idStr] = doc
 		}
 	}
 

internal/config/config.go

@@ -224,7 +224,7 @@ func LoadConfig() {
 
 	// Validation Defaults
 	viper.SetDefault("validation.enabled", true)
-	viper.SetDefault("validation.full_validation", true)
+	viper.SetDefault("validation.full_validation", false)
 	viper.SetDefault("validation.batch_size", 100)
 	viper.SetDefault("validation.retry_interval_ms", 500)
 	viper.SetDefault("validation.max_validation_workers", 4)

internal/flow/monitor.go

@@ -112,6 +112,7 @@ func (m *Manager) WaitIfPaused() {
 func (m *Manager) HandlePause(w http.ResponseWriter, r *http.Request) {
 	m.emergencyPaused.Store(true)
 	if m.statusMgr != nil {
+		m.statusMgr.SetEmergencyPaused(true)
 		m.statusMgr.SetState("paused", "Paused manually by user")
 		m.statusMgr.Persist(context.Background())
 	}
@@ -123,7 +124,7 @@ func (m *Manager) HandlePause(w http.ResponseWriter, r *http.Request) {
 func (m *Manager) HandleResume(w http.ResponseWriter, r *http.Request) {
 	m.emergencyPaused.Store(false)
 	if m.statusMgr != nil {
-		// Restore state gracefully based on current migration phase
+		m.statusMgr.SetEmergencyPaused(false)
 		if m.statusMgr.IsMigrationFinalized() {
 			m.statusMgr.SetState("completed", "Migration Finalized")
 		} else if m.statusMgr.IsCloneCompleted() {

internal/status/status.go

@@ -121,6 +121,7 @@ type Manager struct {
 	estimatedTotalDocs  atomic.Int64
 	clonedBytes         atomic.Int64
 	clonedDocs          atomic.Int64
+	initialSyncLag      atomic.Value
 
 	// Timestamps (Stored as Unix Seconds)
 	lastSourceEventTime  atomic.Int64
@@ -146,6 +147,7 @@ type Manager struct {
 	fcWriterQueue     atomic.Int64
 	fcNativeLagged    atomic.Bool
 	fcNativeSustainer atomic.Int64
+	fcEmergencyPaused atomic.Bool
 
 	// Validation Stats (In-Memory)
 	valTotalChecked atomic.Int64
@@ -155,6 +157,20 @@ type Manager struct {
 	valQueueSize    atomic.Int64
 }
 
+func parseTime(v interface{}) time.Time {
+	if v == nil {
+		return time.Time{}
+	}
+	if t, ok := v.(time.Time); ok {
+		return t
+	}
+	// Handle BSON DateTime (which is an int64 under the hood representing milliseconds)
+	if dt, ok := v.(bson.DateTime); ok {
+		return time.Unix(int64(dt)/1000, (int64(dt)%1000)*1000000).UTC()
+	}
+	return time.Time{}
+}
+
 func NewManager(client *mongo.Client, isSource bool, version string) *Manager {
 	dbName := config.Cfg.Migration.MetadataDB
 	collName := config.Cfg.Migration.StatusCollection
@@ -169,7 +185,8 @@ func NewManager(client *mongo.Client, isSource bool, version string) *Manager {
 		startTime:  time.Now().UTC(),
 	}
 	m.fcReason.Store("")
-	m.idxCurrentNs.Store("") // Safely initialize empty string
+	m.idxCurrentNs.Store("")
+	m.initialSyncLag.Store(0.0)
 	return m
 }
 
@@ -286,8 +303,15 @@ func (m *Manager) SetInitialSyncEnd(t time.Time) {
 func (m *Manager) IsInitialSyncCompleted() bool {
 	return m.initialSyncCompleted.Load()
 }
+
 func (m *Manager) SetInitialSyncCompleted(lagSeconds float64) {
 	m.initialSyncCompleted.Store(true)
+	m.initialSyncLag.Store(lagSeconds)
+}
+
+func (m *Manager) SetEmergencyPaused(paused bool) {
+	m.fcEmergencyPaused.Store(paused)
+	go m.Persist(context.Background())
 }
 
 func (m *Manager) IsCloneCompleted() bool {
@@ -387,16 +411,27 @@ func (m *Manager) LoadAndMerge(ctx context.Context) error {
 			m.idxFinished.Store(val)
 		}
 	}
-
 	m.mu.Lock()
-	if val, ok := doc["initialSyncStart"].(time.Time); ok {
-		m.initialSyncStart = val
+
+	if val, ok := doc["initialSyncStart"]; ok {
+		m.initialSyncStart = parseTime(val)
 	}
-	if val, ok := doc["initialSyncEnd"].(time.Time); ok {
-		m.initialSyncEnd = val
+	if val, ok := doc["initialSyncEnd"]; ok {
+		m.initialSyncEnd = parseTime(val)
 	}
 	m.mu.Unlock()
 
+	if val, ok := doc["initialSyncLag"]; ok {
+		switch v := val.(type) {
+		case float64:
+			m.initialSyncLag.Store(v)
+		case int32:
+			m.initialSyncLag.Store(float64(v))
+		case int64:
+			m.initialSyncLag.Store(float64(v))
+		}
+	}
+
 	if fc, ok := doc["flowControl"].(bson.M); ok {
 		if val, ok := fc["isPaused"].(bool); ok {
 			m.fcIsPaused.Store(val)
@@ -416,6 +451,9 @@ func (m *Manager) LoadAndMerge(ctx context.Context) error {
 		if val, ok := fc["assignedRateLimit"]; ok {
 			m.fcNativeSustainer.Store(parseInt64(val))
 		}
+		if val, ok := fc["emergencyPaused"].(bool); ok {
+			m.fcEmergencyPaused.Store(val)
+		}
 	}
 
 	return nil
@@ -437,9 +475,12 @@ func (m *Manager) Persist(ctx context.Context) {
 		"completed":            m.idxFinished.Load(),
 	}
 
+	lag, _ := m.initialSyncLag.Load().(float64)
+
 	fcStatus := bson.M{
 		"enabled":             config.Cfg.FlowControl.Enabled,
 		"isPaused":            m.fcIsPaused.Load(),
+		"emergencyPaused":     m.fcEmergencyPaused.Load(),
 		"pauseReason":         m.fcReason.Load().(string),
 		"currentQueuedOps":    m.fcQueuedOps.Load(),
 		"currentWriterQueue":  m.fcWriterQueue.Load(),
@@ -470,6 +511,7 @@ func (m *Manager) Persist(ctx context.Context) {
 			"initialSyncCompleted":  m.initialSyncCompleted.Load(),
 			"initialSyncStart":      start,
 			"initialSyncEnd":        end,
+			"initialSyncLag":        lag,
 			"migrationFinalized":    m.migrationFinalized.Load(),
 			"flowControl":           fcStatus,
 			"indexing":              idxStatus,
@@ -548,6 +590,7 @@ func (m *Manager) buildStatusOutput(valStore ValidatorStore) StatusOutput {
 	lastSourceUnix := m.lastSourceEventTime.Load()
 	lastAppliedUnix := m.lastAppliedEventTime.Load()
 	lastBatchUnix := m.lastBatchAppliedAt.Load()
+	lag, _ := m.initialSyncLag.Load().(float64)
 
 	idleTime := 0.0
 	cdcLag := 0.0
@@ -670,20 +713,23 @@ func (m *Manager) buildStatusOutput(valStore ValidatorStore) StatusOutput {
 			StartedAt:               startStr,
 			EndedAt:                 endStr,
 			Duration:                duration,
-			CompletionLagSeconds:    0,
+			CompletionLagSeconds:    lag,
 		},
 	}
-
-	if config.Cfg.FlowControl.Enabled {
+	isEmergencyPaused := m.fcEmergencyPaused.Load()
+	if config.Cfg.FlowControl.Enabled || isEmergencyPaused {
 		output.FlowControl = &FCStatus{
-			Enabled:             true,
-			IsPaused:            m.fcIsPaused.Load(),
+			Enabled:             config.Cfg.FlowControl.Enabled,
+			IsPaused:            m.fcIsPaused.Load() || isEmergencyPaused,
 			PauseReason:         m.fcReason.Load().(string),
 			CurrentQueuedOps:    int(m.fcQueuedOps.Load()),
 			CurrentWriterQueue:  int(m.fcWriterQueue.Load()),
 			NativeLagged:        m.fcNativeLagged.Load(),
 			NativeSustainerRate: int(m.fcNativeSustainer.Load()),
 		}
+		if isEmergencyPaused {
+			output.FlowControl.PauseReason = "Paused manually by user"
+		}
 	}
 
 	return output

readme.md

@@ -642,7 +642,7 @@ This command is used when you are ready to cut over to your new environment. It
 
 ```bash
 --- docStreamer Status (Live) ---
-PID: 775378 (Querying http://localhost:8080/status)
+PID: 1321077 (Querying http://localhost:8080/status)
 {
     "version": "dev",
     "ok": true,
@@ -664,41 +664,41 @@ PID: 775378 (Querying http://localhost:8080/status)
         "isReplicationLagged": false,
         "assignedRateLimit": 0
     },
-    "timeSinceLastEventSeconds": 1861.430341972,
+    "timeSinceLastEventSeconds": 606.175643039,
     "cdcLagSeconds": 0,
-    "totalEventsApplied": 6182,
-    "adaptiveSerialBatches": 0,
-    "insertedDocs": 2830,
-    "updatedDocs": 2306,
-    "deletedDocs": 1046,
+    "totalEventsApplied": 43906,
+    "adaptiveSerialBatches": 7,
+    "insertedDocs": 34422,
+    "updatedDocs": 4710,
+    "deletedDocs": 4774,
     "validation": {
         "queuedBatches": 0,
-        "totalChecked": 10266,
-        "mismatchFound": 64,
-        "mismatchFixed": 64,
+        "totalChecked": 2519,
+        "mismatchFound": 24,
+        "mismatchFixed": 24,
         "pendingMismatches": 0,
         "hotKeysWaiting": 0,
         "syncPercent": 100,
-        "lastValidatedAt": "2026-02-26T21:38:03Z"
+        "lastValidatedAt": "2026-03-03T19:46:33Z"
     },
     "lastSourceEventTime": {
-        "ts": "1772140028",
-        "isoDate": "2026-02-26T21:07:08Z"
+        "ts": "1772567184",
+        "isoDate": "2026-03-03T19:46:24Z"
     },
     "lastAppliedEventTime": {
-        "ts": "1772140028",
-        "isoDate": "2026-02-26T21:07:08Z"
+        "ts": "1772567184",
+        "isoDate": "2026-03-03T19:46:24Z"
     },
-    "lastBatchAppliedAt": "2026-02-26T21:38:09Z",
+    "lastBatchAppliedAt": "2026-03-03T19:56:29Z",
     "initialSync": {
         "completed": true,
         "progressPercent": 100,
-        "clonedDocs": 3459543,
-        "estimatedTotalDocs": 3455411,
-        "clonedBytes": 7026191966,
-        "estimatedTotalBytes": 8853615388,
-        "clonedSizeHuman": "6.5 GB",
-        "estimatedCloneSizeHuman": "8.2 GB",
+        "clonedDocs": 4945277,
+        "estimatedTotalDocs": 4945205,
+        "clonedBytes": 7864311434,
+        "estimatedTotalBytes": 11430286453,
+        "clonedSizeHuman": "7.3 GB",
+        "estimatedCloneSizeHuman": "10.6 GB",
         "startedAt": "",
         "endedAt": "",
         "duration": "N/A",
@@ -821,6 +821,7 @@ Percona docStreamer generates three separate logs, each of the logs location and
 1. Application Log (`logs/docStreamer.log`): Tracks the overall application status and any errors encountered.
 2. Full Load Log (`logs/full_load.log`): Dedicated to the initial full synchronization process. This log, together with the status endpoint, helps you monitor the progress of the initial sync.
 3. CDC Log (`logs/cdc.log`): Dedicated to Change Data Capture (CDC) operations. These operations begin only after the full sync is complete, so this log will remain empty until that point. Use it, along with the status endpoint, to track CDC progress.
+4. Validation Log (`logs/validator.log`): Dedicated log for tracking the validation progress
 
 #### Logging Configuration
 
@@ -832,6 +833,7 @@ The application generates specialized logs to help you monitor different stages
 | `file_path` | `logs/docStreamer.log` | The primary application log containing system status and errors. |
 | `ops_log_path` | `logs/cdc.log` | Dedicated log for Change Data Capture (CDC) operations and batch details. |
 | `full_load_log_path` | `logs/full_load.log` | Dedicated log for tracking the progress and batches of the initial full sync. |
+| `validator_log_path` | `logs/validator.log` | Dedicated log for tracking the validation progress. |
 
 <details>
 <summary>Application log sample:</summary>
@@ -1149,6 +1151,7 @@ In addition to monitoring lock queues, the Adaptive Flow Control mechanism uses
 | `latency_threshold_ms` | `250` | The maximum acceptable latency for a `serverStatus` command. If exceeded, the system pauses to allow the target to recover. |
 | `active_client_threshold` | `50` | The maximum number of total concurrent active clients allowed on the target before throttling occurs. |
 | `min_wired_tiger_tickets` | `0` | The minimum number of available WiredTiger write tickets required. If it drops below this value, the system pauses. Set to `0` to disable. |
+| `target_max_queued_ops` | `50` | The safety limit for the Target's Global Lock Queue. If any node exceeds this many queued operations, docStreamer pauses. |
 
 #### Ad-Hoc Emergency Flow Control (Pause / Resume)
 
@@ -1185,6 +1188,7 @@ The data validation engine is highly configurable to balance performance impact
 | Setting | Default | Description |
 |--------:|--------:|-------------|
 | `enabled` | `true` | Master switch for the validation engine. If false, final document verification after CDC writes are skipped. CDC is guaranteed to sync the documents; this is an optional additional validation check. |
+| `full_validation` | `false` | Controls the scope of CDC data validation. If true, all ops are validated. If false (recommended), ONLY Deletes are validated (massive performance boost since inserts/updates are guaranteed by the stream payload). Note: This setting is hot-reloadable and can be changed without restarting the app. |
 | `batch_size` | `100` | Network vs. Memory Trade-off. Controls how many document IDs are bundled into a single database lookup. Larger batches reduce network round-trips but increase memory usage. |
 | `max_validation_workers` | `4` | Concurrency Control. The number of parallel worker threads fetching and comparing documents. Increase this if you have spare CPU/Network capacity and notice validation lagging behind CDC. |
 | `queue_size` | `2000` | Buffer Capacity. The size of the channel buffering CDC events before validation. If the CDC writer is faster than the validator and this buffer fills up, validation requests will be queued and this could cause slowing down the replication stream. |