feat: add a small abstraction for db & rr replays: try to support load_location for rr

alehander92 · alehander92 · commit 966d626f4c32 · 2025-10-13T21:23:59.000+03:00
diff --git a/src/db-backend/src/db.rs b/src/db-backend/src/db.rs
@@ -1,23 +1,26 @@
 use num_bigint::BigInt;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
+use std::error::Error;
 use std::path::PathBuf;
 use std::vec::Vec;
 
-use crate::distinct_vec::DistinctVec;
-use crate::expr_loader::ExprLoader;
-use crate::lang::Lang;
-use crate::task::{Call, CallArg, Location, RRTicks};
-use crate::value::{Type, Value};
 use log::{error, info, warn};
 use runtime_tracing::{
     CallKey, EventLogKind, FullValueRecord, FunctionId, FunctionRecord, Line, PathId, Place, StepId, TypeId, TypeKind,
     TypeRecord, TypeSpecificInfo, ValueRecord, VariableId, NO_KEY,
 };
 
+use crate::distinct_vec::DistinctVec;
+use crate::expr_loader::ExprLoader;
+use crate::lang::Lang;
+use crate::replay::Replay;
+use crate::task::{Call, CallArg, Location, RRTicks, NO_INDEX};
+use crate::value::{Type, Value};
+
 const NEXT_INTERNAL_STEP_OVERS_LIMIT: usize = 1_000;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Db {
     pub workdir: PathBuf,
     pub functions: DistinctVec<FunctionId, FunctionRecord>,
@@ -769,3 +772,34 @@ pub enum EndOfProgram {
 //     pub step_id: StepId,
 //     pub place: Place,
 // }
+
+#[derive(Debug)]
+pub struct DbReplay {
+    pub db: Box<Db>,
+    pub step_id: StepId,
+}
+
+impl DbReplay {
+    pub fn new(db: Box<Db>) -> DbReplay {
+        DbReplay { db, step_id: StepId(0) }
+    }
+
+    pub fn step_id_jump(&mut self, step_id: StepId) {
+        if step_id.0 != NO_INDEX {
+            self.step_id = step_id;
+        }
+    }
+}
+
+impl Replay for DbReplay {
+    fn load_location(&mut self, expr_loader: &mut ExprLoader) -> Result<Location, Box<dyn Error>> {
+        info!("load_location: db replay");
+        let call_key = self.db.call_key_for_step(self.step_id);
+        Ok(self.db.load_location(self.step_id, call_key, expr_loader))
+    }
+
+    fn run_to_entry(&mut self) -> Result<(), Box<dyn Error>> {
+        self.step_id_jump(StepId(0));
+        Ok(())
+    }
+}
diff --git a/src/db-backend/src/handler.rs b/src/db-backend/src/handler.rs
@@ -10,11 +10,12 @@ use runtime_tracing::{CallKey, EventLogKind, Line, PathId, StepId, VariableId, N
 
 use crate::calltrace::Calltrace;
 use crate::dap::{self, DapClient, DapMessage};
-use crate::db::{Db, DbCall, DbRecordEvent, DbStep};
+use crate::db::{Db, DbCall, DbRecordEvent, DbReplay, DbStep};
 use crate::event_db::{EventDb, SingleTableId};
 use crate::expr_loader::ExprLoader;
 use crate::flow_preloader::FlowPreloader;
 use crate::program_search_tool::ProgramSearchTool;
+use crate::replay::Replay;
 use crate::rr_dispatcher::{CtRRArgs, RRDispatcher};
 // use crate::response::{};
 use crate::dap_types;
@@ -53,11 +54,13 @@ pub struct Handler {
     pub previous_step_id: StepId,
 
     pub trace_kind: TraceKind,
-    pub rr: RRDispatcher,
+    // pub rr: RRDispatcher,
+    pub replay: Box<dyn Replay>,
+
     pub breakpoint_list: Vec<HashMap<usize, BreakpointRecord>>,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub enum TraceKind {
     DB,
     RR,
@@ -100,10 +103,15 @@ impl Handler {
         let mut breakpoint_list: Vec<HashMap<usize, BreakpointRecord>> = Default::default();
         breakpoint_list.resize_with(db.paths.len(), HashMap::new);
         let step_lines_loader = StepLinesLoader::new(&db, &mut expr_loader);
+        let replay: Box<dyn Replay> = if trace_kind == TraceKind::DB {
+            Box::new(DbReplay::new(db.clone()))
+        } else {
+            Box::new(RRDispatcher::new(ct_rr_args))
+        };
         // let sender = sender::Sender::new();
         Handler {
             trace_kind,
-            db,
+            db: db.clone(),
             step_id: StepId(0),
             last_call_key: CallKey(0),
             indirect_send,
@@ -117,7 +125,8 @@ impl Handler {
             step_lines_loader,
             dap_client: DapClient::default(),
             previous_step_id: StepId(0),
-            rr: RRDispatcher::new(ct_rr_args),
+            // rr: RRDispatcher::new(ct_rr_args),
+            replay,
             resulting_dap_messages: vec![],
             raw_diff_index: None,
         }
@@ -215,29 +224,28 @@ impl Handler {
         self.send_dap(&response)
     }
 
-    fn complete_move(&mut self, is_main: bool) -> Result<(), Box<dyn Error>> {
-        let call_key = self.db.call_key_for_step(self.step_id);
-        let reset_flow = is_main || call_key != self.last_call_key;
-        self.last_call_key = call_key;
-        info!("complete move: step_id: {:?}", self.step_id);
-        let move_state = MoveState {
-            status: "".to_string(),
-            location: self.db.load_location(self.step_id, call_key, &mut self.expr_loader),
-            c_location: Location::default(),
-            main: is_main,
-            reset_flow,
-            stop_signal: RRGDBStopSignal::OtherStopSignal,
-            frame_info: FrameInfo::default(),
-        };
-
-        // info!("move_state {:?}", move_state);
+    // will be sent after completion of query
+    fn prepare_stopped_event(&mut self, is_main: bool) -> Result<(), Box<dyn Error>> {
         let reason = if is_main { "entry" } else { "step" };
         info!("generate stopped event");
         let raw_event = self.dap_client.stopped_event(reason)?;
         info!("raw stopped event: {:?}", raw_event);
         self.send_dap(&raw_event)?;
-        let raw_complete_move_event = self.dap_client.complete_move_event(&move_state)?;
+        Ok(())
+    }
+
+    fn prepare_complete_move_event(&mut self, move_state: &MoveState) -> Result<(), Box<dyn Error>> {
+        let raw_complete_move_event = self.dap_client.complete_move_event(move_state)?;
         self.send_dap(&raw_complete_move_event)?;
+        Ok(())
+    }
+
+    fn prepare_output_events(&mut self) -> Result<(), Box<dyn Error>> {
+        if self.trace_kind == TraceKind::RR {
+            warn!("prepare_output_events not implemented for rr");
+            return Ok(()); // TODO
+        }
+
         if self.step_id.0 > self.previous_step_id.0 {
             let mut raw_output_events: Vec<dap::DapMessage> = vec![];
             for event in self.db.events.iter() {
@@ -264,9 +272,14 @@ impl Handler {
                 self.send_dap(raw_output_event)?;
             }
         }
-        self.previous_step_id = self.step_id;
+        Ok(())
+    }
 
-        // self.send_notification(NotificationKind::Success, "Complete move!", true)?;
+    fn prepare_eventual_error_event(&mut self) -> Result<(), Box<dyn Error>> {
+        if self.trace_kind == TraceKind::RR {
+            warn!("prepare_eventual_error_event not implemented for rr");
+            return Ok(()); // TODO
+        }
 
         let exact = false; // or for now try as flow // true just for this exact step
         let step_events = self.db.load_step_events(self.step_id, exact);
@@ -283,15 +296,44 @@ impl Handler {
         Ok(())
     }
 
+    fn complete_move(&mut self, is_main: bool) -> Result<(), Box<dyn Error>> {
+        info!("complete_move");
+
+        // self.db.load_location(self.step_id, call_key, &mut self.expr_loader),
+        let location = self.replay.load_location(&mut self.expr_loader)?;
+        // let call_key = location.call_key; // self.db.call_key_for_step(self.step_id);
+        // TODO: change if we need to support non-int keys
+        let call_key = CallKey(location.key.parse::<i64>()?);
+        let reset_flow = is_main || call_key != self.last_call_key;
+        self.last_call_key = call_key;
+        info!("  location: {location:?}");
+
+        let move_state = MoveState {
+            status: "".to_string(),
+            location,
+            c_location: Location::default(),
+            main: is_main,
+            reset_flow,
+            stop_signal: RRGDBStopSignal::OtherStopSignal,
+            frame_info: FrameInfo::default(),
+        };
+
+        self.prepare_stopped_event(is_main)?;
+        self.prepare_complete_move_event(&move_state)?;
+        self.prepare_output_events()?;
+
+        self.previous_step_id = self.step_id;
+
+        // self.send_notification(NotificationKind::Success, "Complete move!", true)?;
+
+        self.prepare_eventual_error_event()?;
+
+        Ok(())
+    }
+
     pub fn run_to_entry(&mut self, _req: dap::Request) -> Result<(), Box<dyn Error>> {
-        match self.trace_kind {
-            TraceKind::DB => {
-                self.step_id_jump(StepId(0));
-            }
-            TraceKind::RR => {
-                self.rr.run_to_entry()?;
-            }
-        }
+        self.replay.run_to_entry()?;
+        self.step_id = StepId(0); // TODO: use only db replay step_id or another workaround?
         self.complete_move(true)?;
         Ok(())
     }
diff --git a/src/db-backend/src/lib.rs b/src/db-backend/src/lib.rs
@@ -40,6 +40,7 @@ pub mod lang;
 pub mod paths;
 pub mod program_search_tool;
 pub mod query;
+pub mod replay;
 pub mod rr_dispatcher;
 pub mod step_lines_loader;
 pub mod task;
diff --git a/src/db-backend/src/main.rs b/src/db-backend/src/main.rs
@@ -39,6 +39,7 @@ mod lang;
 mod paths;
 mod program_search_tool;
 mod query;
+mod replay;
 mod rr_dispatcher;
 mod step_lines_loader;
 mod task;
diff --git a/src/db-backend/src/query.rs b/src/db-backend/src/query.rs
@@ -4,4 +4,5 @@ use serde::{Deserialize, Serialize};
 #[serde(tag = "kind")]
 pub enum CtRRQuery {
     RunToEntry,
+    LoadLocation,
 }
diff --git a/src/db-backend/src/replay.rs b/src/db-backend/src/replay.rs
@@ -0,0 +1,9 @@
+use std::error::Error;
+
+use crate::expr_loader::ExprLoader;
+use crate::task::Location;
+
+pub trait Replay: std::fmt::Debug {
+    fn load_location(&mut self, expr_loader: &mut ExprLoader) -> Result<Location, Box<dyn Error>>;
+    fn run_to_entry(&mut self) -> Result<(), Box<dyn Error>>;
+}
diff --git a/src/db-backend/src/rr_dispatcher.rs b/src/db-backend/src/rr_dispatcher.rs
@@ -9,8 +9,11 @@ use std::time::Duration;
 
 use log::info;
 
+use crate::expr_loader::ExprLoader;
 use crate::paths::ct_rr_worker_socket_path;
 use crate::query::CtRRQuery;
+use crate::replay::Replay;
+use crate::task::Location;
 
 #[derive(Debug)]
 pub struct RRDispatcher {
@@ -90,8 +93,8 @@ impl CtRRWorker {
         Ok(())
     }
 
-    // for now: don't return a typed value here, only ok or an error
-    pub fn run_query(&mut self, query: CtRRQuery) -> Result<(), Box<dyn Error>> {
+    // for now: don't return a typed value here, only Ok(raw value) or an error
+    pub fn run_query(&mut self, query: CtRRQuery) -> Result<String, Box<dyn Error>> {
         let raw_json = serde_json::to_string(&query)?;
 
         info!("send to worker {raw_json}\n");
@@ -110,8 +113,8 @@ impl CtRRWorker {
 
         info!("res {res}");
 
-        if res == "ok" {
-            Ok(())
+        if !res.starts_with("error:") {
+            Ok(res)
         } else {
             Err(format!("run_query ct rr worker error: {}", res).into())
         }
@@ -127,11 +130,6 @@ impl RRDispatcher {
         }
     }
 
-    pub fn run_to_entry(&mut self) -> Result<(), Box<dyn Error>> {
-        self.ensure_active_stable()?;
-        self.stable.run_query(CtRRQuery::RunToEntry)
-    }
-
     pub fn ensure_active_stable(&mut self) -> Result<(), Box<dyn Error>> {
         // start stable process if not active, store fields, setup ipc? store in stable
         if !self.stable.active {
@@ -145,3 +143,18 @@ impl RRDispatcher {
         Ok(())
     }
 }
+
+impl Replay for RRDispatcher {
+    fn load_location(&mut self, _expr_loader: &mut ExprLoader) -> Result<Location, Box<dyn Error>> {
+        self.ensure_active_stable()?;
+        Ok(serde_json::from_str::<Location>(
+            &self.stable.run_query(CtRRQuery::LoadLocation)?,
+        )?)
+    }
+
+    fn run_to_entry(&mut self) -> Result<(), Box<dyn Error>> {
+        self.ensure_active_stable()?;
+        let _ok = self.stable.run_query(CtRRQuery::RunToEntry)?;
+        Ok(())
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -4,4 +4,5 @@ use serde::{Deserialize, Serialize};`
`4`	`4`	`#[serde(tag = "kind")]`
`5`	`5`	`pub enum CtRRQuery {`
`6`	`6`	`RunToEntry,`
	`7`	`+ LoadLocation,`
`7`	`8`	`}`