Fix pipeline reset consistency across backend state, simulation replay, and dashboard metrics

backend/app/redis_client.py

@@ -161,6 +161,50 @@ def get_counters() -> dict:
     except Exception as exc:
         logger.debug(f"[Redis] get_counters failed: {exc}")
         return {}
+def reset_counters() -> bool:
+    """
+    Hard-reset all shared pipeline metric counters and the attack_types hash.
+    Must be called on pipeline reset — otherwise get_metrics() keeps serving
+    stale numbers from Redis even after Postgres has been zeroed, since
+    Redis is checked first and was never touched by the old reset code.
+    """
+    r = get_client()
+    if r is None:
+        return False
+    try:
+        pipe = r.pipeline(transaction=False)
+        for field in METRIC_KEYS:
+            pipe.set(_rkey(field), 0)
+        pipe.delete(ATTACK_TYPES_KEY)
+        pipe.execute()
+        logger.info("[Redis] Pipeline metric counters reset to 0")
+        return True
+    except Exception as exc:
+        logger.warning(f"[Redis] reset_counters failed: {exc}")
+        return False
+
+
+def reset_admin_stats() -> bool:
+    """
+    Hard-reset all admin alert stat counters to 0 (real SET, not SETNX).
+    init_admin_stats_from_db() uses SETNX on purpose (so a second pod
+    starting doesn't clobber live counters) — but that means it can never
+    be used to clear stale values after a reset. This is the explicit
+    clear path admin_store.get_stats() needs.
+    """
+    r = get_client()
+    if r is None:
+        return False
+    try:
+        pipe = r.pipeline(transaction=False)
+        for field in ADMIN_STAT_KEYS:
+            pipe.set(_admin_key(field), 0)
+        pipe.execute()
+        logger.info("[Redis] Admin alert stat counters reset to 0")
+        return True
+    except Exception as exc:
+        logger.warning(f"[Redis] reset_admin_stats failed: {exc}")
+        return False
 
 
 def is_available() -> bool:
@@ -254,3 +298,4 @@ def init_admin_stats_from_db(db_stats: dict) -> bool:
     except Exception as exc:
         logger.warning(f"[Redis] init_admin_stats_from_db failed: {exc}")
         return False
+

backend/app/routes/admin.py

@@ -489,15 +489,15 @@ async def confirm_reset(request: ResetConfirmRequest, admin=Depends(get_current_
         db.execute_query("UPDATE hpe_pipeline_metrics SET total_requests=0, total_threats=0, total_allowed=0, total_monitored=0, total_blocked=0, total_critical=0, total_latency_ms=0, attack_types='{}' WHERE id=1")
         db.execute_query("UPDATE hpe_simulation_state SET sim_index=0 WHERE id=1")
 
-        # 2. Reset in-memory caches
-        from app import threat_engine
+        # 2. Reset in-memory caches (local deltas) AND the Redis live-view
+        #    counters — get_metrics()/get_stats() prefer Redis when it's
+        #    reachable, so skipping this step leaves the dashboard showing
+        #    stale numbers even though Postgres was just zeroed.
+        from app import threat_engine, redis_client
         import app.routes.simulate as simulate_route
-        threat_engine._metrics = {
-            "total_requests": 0, "total_threats": 0, "total_allowed": 0,
-            "total_monitored": 0, "total_blocked": 0, "total_critical": 0,
-            "total_latency_ms": 0.0, "attack_types": {},
-        }
-        threat_engine._pending_updates = 0
+        threat_engine.reset_local_deltas()
+        redis_client.reset_counters()
+        redis_client.reset_admin_stats()
         simulate_route._sim_index = 0
         simulate_route._sim_batch_count = 0
 
@@ -518,8 +518,22 @@ async def confirm_reset(request: ResetConfirmRequest, admin=Depends(get_current_
                 elastic_client._es.indices.delete(index='hpe-audit-logs', ignore_unavailable=True)
                 elastic_client._es.indices.delete(index='hpe-threats', ignore_unavailable=True)
                 time.sleep(1)
+                # Recreate with the proper mappings instead of leaving it to
+                # dynamic mapping on the next indexed doc.
+                elastic_client.connect_elasticsearch()
             except Exception as e:
                 logger.error(f"ES index deletion error: {e}")
+        # 5. Reset simulation state
+        try:
+            # If you have a route or helper for this, call it
+            from app.routes.simulate import reset_simulation_state
+            reset_simulation_state()
+            logger.info("[RESET] Simulation state reset")
+        except Exception as e:
+            logger.warning(f"Could not reset simulation state: {e}")
+        # 6. Broadcast reset event to dashboards
+        from app.ws_manager import manager as ws_manager
+        await ws_manager.broadcast({"type": "pipeline_reset"})
 
         logger.warning(f"[ADMIN] Pipeline reset EXECUTED by {admin_username}")
         return {"success": True, "message": "Pipeline reset complete. Audit log and Kafka topics preserved."}

backend/app/routes/simulate.py

@@ -51,6 +51,13 @@ def _save_sim_index():
         except Exception as e:
             logger.error(f"Failed to save sim_index: {e}")
 
+def reset_simulation_state():
+    """Reset simulation index to 0 (called on pipeline reset)."""
+    global _sim_index, _sim_batch_count
+    _sim_index = 0
+    _sim_batch_count = 0
+    logger.info("[Simulate] Simulation state reset to index 0")
+
 def _load_test_events():
     global _test_events
     path = Path(TEST_EVENTS_PATH)

backend/app/threat_engine.py

@@ -55,6 +55,20 @@ def load_metrics_from_db():
     """No longer caches in memory. DB is the source of truth."""
     pass
 
+def reset_local_deltas():
+    """Reset all in-memory metric deltas to zero (called on pipeline reset)."""
+    global _pending_updates
+    with _metrics_lock:
+        for k in list(_local_deltas.keys()):
+            if k == "attack_types":
+                _local_deltas[k] = {}
+            elif k == "total_latency_ms":
+                _local_deltas[k] = 0.0
+            else:
+                _local_deltas[k] = 0
+        _pending_updates = 0
+    logger.info("[Metrics] Local in-memory deltas reset to 0")
+
 def flush_metrics_to_db():
     """Flush pending local deltas to Postgres using atomic increments."""
     global _pending_updates

frontend/src/App.jsx

@@ -61,6 +61,7 @@ export default function App() {
   // Queues
   const eventQueueRef = useRef([]);
   const processingRef = useRef(false);
+  const resetInProgressRef = useRef(false);
 
   // Expose switchTab globally
   useEffect(() => {
@@ -153,21 +154,21 @@ export default function App() {
 
       // 2. Update Pipeline Stage Latencies
       const stages = prediction.pipeline_stages || [];
-      
+
       // Always update latencies for display
       if (stages && stages.length > 0) {
         setStageLatencies(stages.map(s => parseFloat(s.latency_ms?.toFixed(1) || 0)));
       }
-      
+
       // Only run the old animation system for live portal events
       if (shouldAnimate) {
         // Set threat flow state FIRST before animation starts
         setIsThreatFlow(isThreat);
-        
+
         // Clean stages and active connector
         setActiveStage(-1);
         setActiveConnector(-1);
-        
+
         // Small delay to ensure state is set
         await sleep(50);
 
@@ -208,7 +209,7 @@ export default function App() {
           }
           await sleep(150);
         }
-        
+
         // Reset animation states after completion
         await sleep(800);
         setActiveStage(-1);
@@ -226,34 +227,37 @@ export default function App() {
       });
 
       // 4. Update HUD and dashboard stats
-      totalProcessedRef.current += 1;
-      latencySumRef.current += prediction.total_latency_ms || 0;
-
-      if (isThreat) {
-        threatsInterceptedRef.current += 1;
+      // Guard: if a reset arrived while this event was being processed, discard its counters
+      if (!resetInProgressRef.current) {
+        totalProcessedRef.current += 1;
+        latencySumRef.current += prediction.total_latency_ms || 0;
+
+        if (isThreat) {
+          threatsInterceptedRef.current += 1;
+
+          const aType = prediction.event_summary?.anomaly_type || 'Unknown';
+          attackTypesRef.current = {
+            ...attackTypesRef.current,
+            [aType]: (attackTypesRef.current[aType] || 0) + 1,
+          };
+        }
 
-        const aType = prediction.event_summary?.anomaly_type || 'Unknown';
-        attackTypesRef.current = {
-          ...attackTypesRef.current,
-          [aType]: (attackTypesRef.current[aType] || 0) + 1,
-        };
-      }
+        const threatAction = prediction.threat_action || 'ALLOW';
+        if (threatAction === 'ALLOW' || threatAction === 'MONITOR') {
+          allowedProcessedRef.current += 1;
+        } else if (threatAction === 'BLOCK' || threatAction === 'CRITICAL_ALERT') {
+          blockedProcessedRef.current += 1;
+        }
 
-      const threatAction = prediction.threat_action || 'ALLOW';
-      if (threatAction === 'ALLOW' || threatAction === 'MONITOR') {
-        allowedProcessedRef.current += 1;
-      } else if (threatAction === 'BLOCK' || threatAction === 'CRITICAL_ALERT') {
-        blockedProcessedRef.current += 1;
+        setEventsProcessed(totalProcessedRef.current);
+        setThreatsIntercepted(threatsInterceptedRef.current);
+        setTotalProcessed(totalProcessedRef.current);
+        setAllowedProcessed(allowedProcessedRef.current);
+        setBlockedProcessed(blockedProcessedRef.current);
+        setAvgLatency(totalProcessedRef.current > 0 ? latencySumRef.current / totalProcessedRef.current : 0);
+        setAttackTypes(attackTypesRef.current);
       }
 
-      setEventsProcessed(totalProcessedRef.current);
-      setThreatsIntercepted(threatsInterceptedRef.current);
-      setTotalProcessed(totalProcessedRef.current);
-      setAllowedProcessed(allowedProcessedRef.current);
-      setBlockedProcessed(blockedProcessedRef.current);
-      setAvgLatency(totalProcessedRef.current > 0 ? latencySumRef.current / totalProcessedRef.current : 0);
-      setAttackTypes(attackTypesRef.current);
-
       if (prediction.xgb_score !== undefined) {
         setLatestModelScores({
           xgb: prediction.xgb_score,
@@ -321,17 +325,44 @@ export default function App() {
             localSimTimer = null;
           }
         };
-
         ws.onmessage = (event) => {
           try {
             const message = JSON.parse(event.data);
             if (message.type === 'pipeline_result') {
               eventQueueRef.current.push(message.data);
             } else if (message.type === 'vpn_login_alert') {
               showVpnAlertBanner(message.data);
+            } else if (message.type === 'pipeline_reset') {
+              // Set guard so any in-flight event discards its counter updates
+              resetInProgressRef.current = true;
+
+              // Clear local accumulators
+              totalProcessedRef.current = 0;
+              threatsInterceptedRef.current = 0;
+              allowedProcessedRef.current = 0;
+              blockedProcessedRef.current = 0;
+              latencySumRef.current = 0;
+              attackTypesRef.current = {};
+              eventQueueRef.current = [];
+
+              // Reset the globe's counts
+              setEventsProcessed(0);
+              setThreatsIntercepted(0);
+              setTotalProcessed(0);
+              setAllowedProcessed(0);
+              setBlockedProcessed(0);
+              setAvgLatency(0);
+              setAttackTypes({});
+              setArcs([]);
+              setEventsLog([]);
+              setStageLatencies(Array(10).fill(0));
+
+              // Clear guard after a short delay so next real events are counted
+              setTimeout(() => { resetInProgressRef.current = false; }, 500);
             }
-          } catch (e) {
-            console.error('[HPE] Failed to parse simulation message:', e);
+          }
+          catch (e) {
+            console.error("[HPE] Failed to parse simulation message:", e);
           }
         };