Use batch processing for ReactiveFileCollection

Process all files as a single Batch delta instead of individual deltas.
This dramatically reduces the number of updates flowing through reactive
combinators during bulk loading.

Key changes:
- Add process_files_batch to ReactiveFileCollection (emits single Batch)
- Add process_file_silent helper (processes without emitting)
- Update ReactiveAnalysis.process_files to use batch processing

Performance improvement on deadcode-benchmark:
- Before: fixpoint recv=601,563 updates
- After:  fixpoint recv=13 updates (batches)

This means downstream combinators process all file data together
instead of handling 600K+ individual updates one at a time.

Author: cristianoc

analysis/reactive/src/Reactive.ml

@@ -487,9 +487,7 @@ let join (left : ('k1, 'v1) t) (right : ('k2, 'v2) t)
     | Batch entries ->
       (* Collect all affected right keys, then process *)
       let right_keys =
-        entries
-        |> List.map (fun (k, _) -> k)
-        |> List.sort_uniq compare
+        entries |> List.map (fun (k, _) -> k) |> List.sort_uniq compare
       in
       let all_downstream = right_keys |> List.concat_map process_right_key in
       (* Deduplicate *)
@@ -1104,8 +1102,7 @@ let fixpoint ~(init : ('k, unit) t) ~(edges : ('k, 'k list) t) () : ('k, unit) t
       (* Net changes: added if in added_set but not removed_set, etc. *)
       let net_added =
         Hashtbl.fold
-          (fun k () acc ->
-            if Hashtbl.mem removed_set k then acc else k :: acc)
+          (fun k () acc -> if Hashtbl.mem removed_set k then acc else k :: acc)
           added_set []
       in
       let net_removed =
@@ -1144,8 +1141,7 @@ let fixpoint ~(init : ('k, unit) t) ~(edges : ('k, 'k list) t) () : ('k, unit) t
       List.iter (fun k -> Hashtbl.replace removed_set k ()) !all_removed;
       let net_added =
         Hashtbl.fold
-          (fun k () acc ->
-            if Hashtbl.mem removed_set k then acc else k :: acc)
+          (fun k () acc -> if Hashtbl.mem removed_set k then acc else k :: acc)
           added_set []
       in
       let net_removed =

analysis/reactive/src/ReactiveFileCollection.ml

@@ -70,10 +70,31 @@ let process_if_changed t path =
     emit t (Reactive.Set (path, value));
     true (* changed *)
 
-(** Process multiple files *)
+(** Process multiple files (emits individual deltas) *)
 let process_files t paths =
   List.iter (fun path -> ignore (process_if_changed t path)) paths
 
+(** Process a file without emitting. Returns batch entry if changed. *)
+let process_file_silent t path =
+  let new_id = get_file_id path in
+  match Hashtbl.find_opt t.internal.cache path with
+  | Some (old_id, _) when not (file_changed ~old_id ~new_id) ->
+    None (* unchanged *)
+  | _ ->
+    let raw = t.internal.read_file path in
+    let value = t.internal.process path raw in
+    Hashtbl.replace t.internal.cache path (new_id, value);
+    Some (Reactive.set path value)
+
+(** Process multiple files and emit as a single batch.
+    More efficient than process_files when processing many files at once. *)
+let process_files_batch t paths =
+  let entries =
+    paths |> List.filter_map (fun path -> process_file_silent t path)
+  in
+  if entries <> [] then emit t (Reactive.Batch entries);
+  List.length entries
+
 (** Remove a file *)
 let remove t path =
   Hashtbl.remove t.internal.cache path;

analysis/reactive/src/ReactiveFileCollection.mli

@@ -40,7 +40,13 @@ val to_collection : ('raw, 'v) t -> (string, 'v) Reactive.t
 (** {1 Processing} *)
 
 val process_files : ('raw, 'v) t -> string list -> unit
-(** Process files, emitting deltas for changed files. *)
+(** Process files, emitting individual deltas for each changed file. *)
+
+val process_files_batch : ('raw, 'v) t -> string list -> int
+(** Process files, emitting a single [Batch] delta with all changes.
+    Returns the number of files that changed.
+    More efficient than [process_files] when processing many files at once,
+    as downstream combinators can process all changes together. *)
 
 val process_if_changed : ('raw, 'v) t -> string -> bool
 (** Process a file if changed. Returns true if file was processed. *)

analysis/reactive/test/ReactiveTest.ml

@@ -1759,13 +1759,7 @@ let test_batch_flatmap () =
     | Remove k -> received_entries := [(k, None)] @ !received_entries);
 
   (* Send a batch *)
-  emit
-    (Batch
-       [
-         Reactive.set "a" 1;
-         Reactive.set "b" 2;
-         Reactive.set "c" 3;
-       ]);
+  emit (Batch [Reactive.set "a" 1; Reactive.set "b" 2; Reactive.set "c" 3]);
 
   Printf.printf "Received batches: %d, entries: %d\n" !received_batches
     (List.length !received_entries);

analysis/reanalyze/src/ReactiveAnalysis.ml

@@ -74,27 +74,27 @@ let create ~config : t =
       process_cmt_infos ~config ~cmtFilePath:path cmt_infos)
 
 (** Process all files incrementally using ReactiveFileCollection.
-    First run processes all files. Subsequent runs only process changed files. *)
+    First run processes all files. Subsequent runs only process changed files.
+    Uses batch processing to emit all changes as a single Batch delta. *)
 let process_files ~(collection : t) ~config:_ cmtFilePaths : all_files_result =
   Timing.time_phase `FileLoading (fun () ->
-      let processed = ref 0 in
-      let from_cache = ref 0 in
-
-      (* Add/update all files in the collection *)
-      cmtFilePaths
-      |> List.iter (fun cmtFilePath ->
-             let was_in_collection =
-               ReactiveFileCollection.mem collection cmtFilePath
-             in
-             let changed =
-               ReactiveFileCollection.process_if_changed collection cmtFilePath
-             in
-             if changed then incr processed
-             else if was_in_collection then incr from_cache);
+      let total_files = List.length cmtFilePaths in
+      let cached_before =
+        cmtFilePaths
+        |> List.filter (fun p -> ReactiveFileCollection.mem collection p)
+        |> List.length
+      in
+
+      (* Process all files as a batch - emits single Batch delta *)
+      let processed =
+        ReactiveFileCollection.process_files_batch collection cmtFilePaths
+      in
+      let from_cache = total_files - processed in
 
       if !Cli.timing then
-        Printf.eprintf "Reactive: %d files processed, %d from cache\n%!"
-          !processed !from_cache;
+        Printf.eprintf
+          "Reactive: %d files processed, %d from cache (was cached: %d)\n%!"
+          processed from_cache cached_before;
 
       (* Collect results from the collection *)
       let dce_data_list = ref [] in