Add churn mode for incremental correctness testing

- Add -churn n CLI option to simulate file add/remove cycles
- Alternates between removing n random files and adding them back
- Shows issue count changes per run with detailed diff
- Add remove_batch to ReactiveFileCollection for batched removals (28x faster)
- Add reset_stats to Reactive for per-operation stat tracking
- Include churn time in timing report with proper accounting
- Print aggregate stats at end: mean/std for churn time and issue changes
- Add skip_file parameter to filter removed files from processing

Author: cristianoc

analysis/reactive/src/Reactive.ml

@@ -144,6 +144,21 @@ module Registry = struct
     Hashtbl.clear combinators;
     dirty_nodes := []
 
+  let reset_stats () =
+    Hashtbl.iter
+      (fun _ info ->
+        info.stats.deltas_received <- 0;
+        info.stats.entries_received <- 0;
+        info.stats.adds_received <- 0;
+        info.stats.removes_received <- 0;
+        info.stats.process_count <- 0;
+        info.stats.process_time_ns <- 0L;
+        info.stats.deltas_emitted <- 0;
+        info.stats.entries_emitted <- 0;
+        info.stats.adds_emitted <- 0;
+        info.stats.removes_emitted <- 0)
+      nodes
+
   (** Generate Mermaid diagram of the pipeline *)
   let to_mermaid () =
     let buf = Buffer.create 256 in
@@ -1167,3 +1182,4 @@ let fixpoint ~name ~(init : ('k, unit) t) ~(edges : ('k, 'k list) t) () :
 let to_mermaid () = Registry.to_mermaid ()
 let print_stats () = Registry.print_stats ()
 let reset () = Registry.clear ()
+let reset_stats () = Registry.reset_stats ()

analysis/reactive/src/Reactive.mli

@@ -164,3 +164,6 @@ val print_stats : unit -> unit
 
 val reset : unit -> unit
 (** Clear all registered nodes (for tests) *)
+
+val reset_stats : unit -> unit
+(** Reset all node statistics to zero (keeps nodes intact) *)

analysis/reactive/src/ReactiveFileCollection.ml

@@ -82,6 +82,19 @@ let remove t path =
   Hashtbl.remove t.internal.cache path;
   emit t (Reactive.Remove path)
 
+(** Remove multiple files as a batch *)
+let remove_batch t paths =
+  let entries =
+    paths
+    |> List.filter_map (fun path ->
+           if Hashtbl.mem t.internal.cache path then (
+             Hashtbl.remove t.internal.cache path;
+             Some (path, None))
+           else None)
+  in
+  if entries <> [] then emit t (Reactive.Batch entries);
+  List.length entries
+
 (** Clear all cached data *)
 let clear t = Hashtbl.clear t.internal.cache
 

analysis/reactive/src/ReactiveFileCollection.mli

@@ -54,6 +54,10 @@ val process_if_changed : ('raw, 'v) t -> string -> bool
 val remove : ('raw, 'v) t -> string -> unit
 (** Remove a file from the collection. *)
 
+val remove_batch : ('raw, 'v) t -> string list -> int
+(** Remove multiple files as a batch. Returns the number of files removed.
+    More efficient than calling [remove] multiple times. *)
+
 (** {1 Cache Management} *)
 
 val invalidate : ('raw, 'v) t -> string -> unit

analysis/reanalyze/src/Cli.ml

@@ -34,5 +34,8 @@ let reactive = ref false
 (* number of analysis runs (for benchmarking reactive mode) *)
 let runs = ref 1
 
+(* number of files to churn (remove/re-add) between runs for incremental testing *)
+let churn = ref 0
+
 (* output mermaid diagram of reactive pipeline *)
 let mermaid = ref false

analysis/reanalyze/src/Log_.ml

@@ -197,6 +197,7 @@ module Stats = struct
   let issues = ref []
   let addIssue (issue : Issue.t) = issues := issue :: !issues
   let clear () = issues := []
+  let get_issue_count () = List.length !issues
 
   let getSortedIssues () =
     let counters2 = Hashtbl.create 1 in

analysis/reanalyze/src/Reanalyze.ml

@@ -204,8 +204,14 @@ let processFilesParallel ~config ~numDomains (cmtFilePaths : string list) :
 
 (** Process all cmt files and return results for DCE and Exception analysis.
     Conceptually: map process_cmt_file over all files. *)
-let processCmtFiles ~config ~cmtRoot ~reactive_collection : all_files_result =
-  let cmtFilePaths = collectCmtFilePaths ~cmtRoot in
+let processCmtFiles ~config ~cmtRoot ~reactive_collection ~skip_file :
+    all_files_result =
+  let cmtFilePaths =
+    let all = collectCmtFilePaths ~cmtRoot in
+    match skip_file with
+    | Some should_skip -> List.filter (fun p -> not (should_skip p)) all
+    | None -> all
+  in
   (* Reactive mode: use incremental processing that skips unchanged files *)
   match reactive_collection with
   | Some collection ->
@@ -245,10 +251,10 @@ let shuffle_list lst =
   Array.to_list arr
 
 let runAnalysis ~dce_config ~cmtRoot ~reactive_collection ~reactive_merge
-    ~reactive_liveness ~reactive_solver =
+    ~reactive_liveness ~reactive_solver ~skip_file =
   (* Map: process each file -> list of file_data *)
   let {dce_data_list; exception_results} =
-    processCmtFiles ~config:dce_config ~cmtRoot ~reactive_collection
+    processCmtFiles ~config:dce_config ~cmtRoot ~reactive_collection ~skip_file
   in
   (* Get exception results from reactive collection if available *)
   let exception_results =
@@ -522,20 +528,141 @@ let runAnalysisAndReport ~cmtRoot =
            ~config:dce_config)
     | _ -> None
   in
+  (* Collect CMT file paths once for churning *)
+  let cmtFilePaths =
+    if !Cli.churn > 0 then Some (collectCmtFilePaths ~cmtRoot) else None
+  in
+  (* Track previous issue count for diff reporting *)
+  let prev_issue_count = ref 0 in
+  (* Track currently removed files (to add them back on next run) *)
+  let removed_files = ref [] in
+  (* Set of removed files for filtering in processCmtFiles *)
+  let removed_set = Hashtbl.create 64 in
+  (* Aggregate stats for churn mode *)
+  let churn_times = ref [] in
+  let issues_added_list = ref [] in
+  let issues_removed_list = ref [] in
   for run = 1 to numRuns do
     Timing.reset ();
     (* Clear stats at start of each run to avoid accumulation *)
     if run > 1 then Log_.Stats.clear ();
+    (* Print run header first *)
     if numRuns > 1 && !Cli.timing then
       Printf.eprintf "\n=== Run %d/%d ===\n%!" run numRuns;
+    (* Churn: alternate between remove and add phases *)
+    (if !Cli.churn > 0 then
+       match (reactive_collection, cmtFilePaths) with
+       | Some collection, Some paths ->
+         Reactive.reset_stats ();
+         if run > 1 && !removed_files <> [] then (
+           (* Add back previously removed files *)
+           let to_add = !removed_files in
+           removed_files := [];
+           (* Clear removed set so these files get processed again *)
+           List.iter (fun p -> Hashtbl.remove removed_set p) to_add;
+           let t0 = Unix.gettimeofday () in
+           let processed =
+             ReactiveFileCollection.process_files_batch
+               (collection
+                 : ReactiveAnalysis.t
+                 :> (_, _) ReactiveFileCollection.t)
+               to_add
+           in
+           let elapsed = Unix.gettimeofday () -. t0 in
+           Timing.add_churn_time elapsed;
+           churn_times := elapsed :: !churn_times;
+           if !Cli.timing then (
+             Printf.eprintf "  Added back %d files (%.3fs)\n%!" processed
+               elapsed;
+             (match reactive_liveness with
+             | Some liveness -> ReactiveLiveness.print_stats ~t:liveness
+             | None -> ());
+             match reactive_solver with
+             | Some solver -> ReactiveSolver.print_stats ~t:solver
+             | None -> ()))
+         else if run > 1 then (
+           (* Remove new random files *)
+           let numChurn = min !Cli.churn (List.length paths) in
+           let shuffled = shuffle_list paths in
+           let to_remove = List.filteri (fun i _ -> i < numChurn) shuffled in
+           removed_files := to_remove;
+           (* Mark as removed so processCmtFiles skips them *)
+           List.iter (fun p -> Hashtbl.replace removed_set p ()) to_remove;
+           let t0 = Unix.gettimeofday () in
+           let removed =
+             ReactiveFileCollection.remove_batch
+               (collection
+                 : ReactiveAnalysis.t
+                 :> (_, _) ReactiveFileCollection.t)
+               to_remove
+           in
+           let elapsed = Unix.gettimeofday () -. t0 in
+           Timing.add_churn_time elapsed;
+           churn_times := elapsed :: !churn_times;
+           if !Cli.timing then (
+             Printf.eprintf "  Removed %d files (%.3fs)\n%!" removed elapsed;
+             (match reactive_liveness with
+             | Some liveness -> ReactiveLiveness.print_stats ~t:liveness
+             | None -> ());
+             match reactive_solver with
+             | Some solver -> ReactiveSolver.print_stats ~t:solver
+             | None -> ()))
+       | _ -> ());
+    (* Skip removed files in reactive mode *)
+    let skip_file =
+      if Hashtbl.length removed_set > 0 then
+        Some (fun path -> Hashtbl.mem removed_set path)
+      else None
+    in
     runAnalysis ~dce_config ~cmtRoot ~reactive_collection ~reactive_merge
-      ~reactive_liveness ~reactive_solver;
-    if run = numRuns then (
-      (* Only report on last run *)
+      ~reactive_liveness ~reactive_solver ~skip_file;
+    (* Report issue count with diff *)
+    let current_count = Log_.Stats.get_issue_count () in
+    if !Cli.churn > 0 then (
+      let diff = current_count - !prev_issue_count in
+      (* Track added/removed separately *)
+      if run > 1 then
+        if diff > 0 then
+          issues_added_list := float_of_int diff :: !issues_added_list
+        else if diff < 0 then
+          issues_removed_list := float_of_int (-diff) :: !issues_removed_list;
+      let diff_str =
+        if run = 1 then ""
+        else if diff >= 0 then Printf.sprintf " (+%d)" diff
+        else Printf.sprintf " (%d)" diff
+      in
       Log_.Stats.report ~config:dce_config;
-      Log_.Stats.clear ());
+      if !Cli.timing then
+        Printf.eprintf "  Total issues: %d%s\n%!" current_count diff_str;
+      prev_issue_count := current_count)
+    else if run = numRuns then
+      (* Only report on last run for non-churn mode *)
+      Log_.Stats.report ~config:dce_config;
+    Log_.Stats.clear ();
     Timing.report ()
   done;
+  (* Print aggregate churn stats *)
+  if !Cli.churn > 0 && !Cli.timing && List.length !churn_times > 0 then (
+    let calc_stats lst =
+      if lst = [] then (0.0, 0.0)
+      else
+        let n = float_of_int (List.length lst) in
+        let sum = List.fold_left ( +. ) 0.0 lst in
+        let mean = sum /. n in
+        let variance =
+          List.fold_left (fun acc x -> acc +. ((x -. mean) ** 2.0)) 0.0 lst /. n
+        in
+        (mean, sqrt variance)
+    in
+    let time_mean, time_std = calc_stats !churn_times in
+    let added_mean, added_std = calc_stats !issues_added_list in
+    let removed_mean, removed_std = calc_stats !issues_removed_list in
+    Printf.eprintf "\n=== Churn Summary ===\n";
+    Printf.eprintf "  Churn operations: %d\n" (List.length !churn_times);
+    Printf.eprintf "  Churn time: mean=%.3fs std=%.3fs\n" time_mean time_std;
+    Printf.eprintf "  Issues added: mean=%.0f std=%.0f\n" added_mean added_std;
+    Printf.eprintf "  Issues removed: mean=%.0f std=%.0f\n" removed_mean
+      removed_std);
   if !Cli.json then EmitJson.finish ()
 
 let cli () =
@@ -657,6 +784,10 @@ let cli () =
       ( "-runs",
         Int (fun n -> Cli.runs := n),
         "n Run analysis n times (for benchmarking cache effectiveness)" );
+      ( "-churn",
+        Int (fun n -> Cli.churn := n),
+        "n Remove and re-add n random files between runs (tests incremental \
+         correctness)" );
       ("-version", Unit versionAndExit, "Show version information and exit");
       ("--version", Unit versionAndExit, "Show version information and exit");
     ]

analysis/reanalyze/src/Timing.ml

@@ -3,6 +3,8 @@
 let enabled = ref false
 
 type phase_times = {
+  (* Churn (file add/remove) *)
+  mutable churn: float;
   (* CMT processing sub-phases *)
   mutable file_loading: float;
   mutable result_collection: float;
@@ -15,6 +17,7 @@ type phase_times = {
 
 let times =
   {
+    churn = 0.0;
     file_loading = 0.0;
     result_collection = 0.0;
     merging = 0.0;
@@ -26,12 +29,15 @@ let times =
 let timing_mutex = Mutex.create ()
 
 let reset () =
+  times.churn <- 0.0;
   times.file_loading <- 0.0;
   times.result_collection <- 0.0;
   times.merging <- 0.0;
   times.solving <- 0.0;
   times.reporting <- 0.0
 
+let add_churn_time t = times.churn <- times.churn +. t
+
 let now () = Unix.gettimeofday ()
 
 let time_phase phase_name f =
@@ -56,8 +62,11 @@ let report () =
   if !enabled then (
     let cmt_total = times.file_loading in
     let analysis_total = times.merging +. times.solving in
-    let total = cmt_total +. analysis_total +. times.reporting in
+    let total = times.churn +. cmt_total +. analysis_total +. times.reporting in
     Printf.eprintf "\n=== Timing ===\n";
+    if times.churn > 0.0 then
+      Printf.eprintf "  Churn:              %.3fs (%.1f%%)\n" times.churn
+        (100.0 *. times.churn /. total);
     Printf.eprintf "  CMT processing:     %.3fs (%.1f%%)\n" cmt_total
       (100.0 *. cmt_total /. total);
     (* Only show parallel-specific timing when used *)

analysis/src/DceCommand.ml

@@ -2,6 +2,7 @@ let command () =
   Reanalyze.RunConfig.dce ();
   let dce_config = Reanalyze.DceConfig.current () in
   Reanalyze.runAnalysis ~dce_config ~cmtRoot:None ~reactive_collection:None
-    ~reactive_merge:None ~reactive_liveness:None ~reactive_solver:None;
+    ~reactive_merge:None ~reactive_liveness:None ~reactive_solver:None
+    ~skip_file:None;
   let issues = !Reanalyze.Log_.Stats.issues in
   Printf.printf "issues:%d\n" (List.length issues)