fix: saved backlogs are not reloaded at startup on non-preload mode

server/src/backlog/manager.rs

@@ -20,7 +20,7 @@ use super::{Backlog, BacklogOpt, BacklogState};
 
 const BACKLOGMGR_DOWNSTREAM_QUEUE_SIZE: usize = 64;
 
-mod storage;
+pub mod storage;
 
 #[derive(PartialEq, Eq, Hash, Clone)]
 pub struct ManagerReport {
@@ -50,6 +50,7 @@ pub struct BacklogManager {
     pub cancel_tx: broadcast::Sender<()>,
     pub report_tx: mpsc::Sender<ManagerReport>,
     pub directive: Directive,
+    pub test_env: bool,
     load_param: OpLoadParameter,
     report_interval: u64,
     backlogs: Arc<RwLock<Vec<Arc<Backlog>>>>,
@@ -94,6 +95,7 @@ impl BacklogManager {
             cancel_tx: manager_option.cancel_tx.clone(),
             report_tx: manager_option.report_tx.clone(),
             directive,
+            test_env: manager_option.test_env,
             load_param: load_param.to_owned(),
             report_interval: *report_interval,
             backlogs: Arc::new(RwLock::new(vec![])),
@@ -133,7 +135,7 @@ impl BacklogManager {
 
     async fn load_from_storage(&self) -> Result<()> {
         let mut backlogs = self.backlogs.write().await;
-        let res = storage::load(false, self.directive.id).await;
+        let res = storage::load(self.test_env, self.directive.id).await;
         match res {
             Err(e) => {
                 for cause in e.chain() {
@@ -199,6 +201,16 @@ impl BacklogManager {
     }
 
     pub async fn start(&self, ready_tx: oneshot::Sender<()>) -> Result<()> {
+        // lock the rx channel asap
+        // if this fails, it means another instance is already running and we should abort
+        let mut upstream_rx = self.upstream_rx.try_lock().map_err(|e| {
+            error!(
+                directive.id = self.directive.id,
+                "another instance is already running for this directive ID, exiting this one"
+            );
+            e
+        })?;
+
         let mut cancel_rx = self.cancel_tx.subscribe();
         let downstream_tx = self.downstream_tx.clone();
 
@@ -231,24 +243,8 @@ impl BacklogManager {
         // initial report
         _ = report_sender.send(mgr_report.clone()).await;
 
-        // if this fails, it means another instance is already running and we should abort
-        let mut upstream_rx = self.upstream_rx.try_lock().map_err(|e| {
-            error!(
-                directive.id = self.directive.id,
-                "another instance is already running for this directive ID, exiting this one"
-            );
-            e
-        })?;
-
         let mut delete_rx = self.delete_rx.lock().await;
 
-        debug!("about to send ready signal");
-        ready_tx
-            .send(())
-            .map_err(|e| anyhow!("cannot send ready signal: {:?}", e))?;
-
-        debug!("about to check interval timers");
-
         // if the option is set, activate idle_timer to exit the manager thread when there's no backlog after the specified minutes
         let (mut checker, mut idle_timeout) = if let Some(reg) = self.lazy_loader.as_ref() {
             debug!(
@@ -272,6 +268,12 @@ impl BacklogManager {
             )
         };
 
+        // notify only after cache entry is inserted
+        debug!("notifying spawner that backlog manager is ready");
+        ready_tx
+            .send(())
+            .map_err(|e| anyhow!("cannot send ready signal: {:?}", e))?;
+
         debug!(directive.id = self.directive.id, "listening for event");
 
         loop {
@@ -289,7 +291,7 @@ impl BacklogManager {
                         if  backlogs.len() > 0 {
                             let v = &*backlogs;
                             info!(self.directive.id, "saving {} backlogs to disk", v.len());
-                            if let Err(err) = storage::save(false, self.directive.id, v.to_vec()).await {
+                            if let Err(err) = storage::save(self.test_env, self.directive.id, v.to_vec()).await {
                                 error!(directive.id = self.directive.id, "error saving backlogs: {:?}", err);
                             } else {
                                 debug!(directive.id = self.directive.id, "{} backlogs saved", backlogs.len());
@@ -324,7 +326,6 @@ impl BacklogManager {
                             // the order between cache invalidation and dropping the lock here doesn't matter that much
                             // it will just move the location of potential event loss during the transition from this instance to the next.
                             // that's only applicable if there's incoming event while we're exiting this one.
-
                             v.cache.invalidate(&self.directive.id);
                             // events that are in-flight here could potentially create a new backlog manager,
                             // which may not be able to lock the upstream_rx yet before this next line is executed.

server/src/backlog/manager/storage.rs

@@ -10,6 +10,30 @@ use anyhow::Result;
 
 use crate::{backlog::Backlog, utils};
 
+pub fn list(test_env: bool) -> Result<Vec<u64>> {
+    let backlog_dir = utils::log_dir(test_env)?.join("backlogs");
+    let mut res = vec![];
+    let files = std::fs::read_dir(backlog_dir)?;
+    files
+        .filter_map(Result::ok)
+        .filter(|d| {
+            if let Some(e) = d.path().extension() {
+                e == "json"
+            } else {
+                false
+            }
+        })
+        .for_each(|f| {
+            if let Ok(v) = f.file_name().into_string() {
+                let s = v.replace(".json", "");
+                if let Ok(id) = s.parse::<u64>() {
+                    res.push(id);
+                }
+            }
+        });
+    Ok(res)
+}
+
 pub async fn load(test_env: bool, directive_id: u64) -> Result<Vec<Backlog>> {
     let backlog_dir = utils::log_dir(test_env)?.join("backlogs");
     let filename = backlog_dir.join(directive_id.to_string() + ".json");