feat: prover redis cleanup job (#1852)

* redis queue cleanup job

* format

* format

Author: sergeytimoshin

forester/tests/e2e_v1_test.rs

@@ -234,7 +234,7 @@ async fn test_e2e_v1() {
         work_report_sender2,
     ));
 
-    const EXPECTED_EPOCHS: u64 = 3; // We expect to process 2 epochs (0 and 1)
+    const EXPECTED_EPOCHS: u64 = 2; // We expect to process 2 epochs (0 and 1)
 
     let mut processed_epochs = HashSet::new();
     let mut total_processed = 0;

forester/tests/e2e_v2_test.rs

@@ -64,7 +64,7 @@ use crate::test_utils::{get_registration_phase_start_slot, init, wait_for_slot};
 mod test_utils;
 
 const MINT_TO_NUM: u64 = 5;
-const DEFAULT_TIMEOUT_SECONDS: u64 = 60 * 15;
+const DEFAULT_TIMEOUT_SECONDS: u64 = 60 * 5;
 const COMPUTE_BUDGET_LIMIT: u32 = 1_000_000;
 
 #[derive(Debug, Clone, Copy, PartialEq)]
@@ -741,7 +741,7 @@ async fn execute_test_transactions<R: Rpc + Indexer + MerkleTreeExt, I: Indexer>
     address_v1_counter: &mut u64,
     address_v2_counter: &mut u64,
 ) {
-    let mut iterations = 10;
+    let mut iterations = 4;
     if is_v2_state_test_enabled() {
         let batch_size =
             get_state_v2_batch_size(rpc, &env.v2_state_trees[0].merkle_tree).await as usize;

prover/server/main.go

@@ -970,6 +970,22 @@ func startCleanupRoutines(redisQueue *server.RedisQueue) {
 		logging.Logger().Info().Msg("Startup cleanup of old proof requests completed")
 	}
 
+	if err := redisQueue.CleanupStuckProcessingJobs(); err != nil {
+		logging.Logger().Error().
+			Err(err).
+			Msg("Failed to cleanup stuck processing jobs on startup")
+	} else {
+		logging.Logger().Info().Msg("Startup cleanup of stuck processing jobs completed")
+	}
+
+	if err := redisQueue.CleanupOldFailedJobs(); err != nil {
+		logging.Logger().Error().
+			Err(err).
+			Msg("Failed to cleanup old failed jobs on startup")
+	} else {
+		logging.Logger().Info().Msg("Startup cleanup of old failed jobs completed")
+	}
+
 	if err := redisQueue.CleanupOldResults(); err != nil {
 		logging.Logger().Error().
 			Err(err).
@@ -978,6 +994,31 @@ func startCleanupRoutines(redisQueue *server.RedisQueue) {
 		logging.Logger().Info().Msg("Startup cleanup of old results completed")
 	}
 
+	if err := redisQueue.CleanupOldResultKeys(); err != nil {
+		logging.Logger().Error().
+			Err(err).
+			Msg("Failed to cleanup old result keys on startup")
+	} else {
+		logging.Logger().Info().Msg("Startup cleanup of old result keys completed")
+	}
+
+	go func() {
+		processingTicker := time.NewTicker(10 * time.Second)
+		defer processingTicker.Stop()
+
+		logging.Logger().Info().Msg("Started stuck processing jobs cleanup routine (every 10 seconds)")
+
+		for range processingTicker.C {
+			if err := redisQueue.CleanupStuckProcessingJobs(); err != nil {
+				logging.Logger().Error().
+					Err(err).
+					Msg("Failed to cleanup stuck processing jobs")
+			} else {
+				logging.Logger().Debug().Msg("Stuck processing jobs cleanup completed")
+			}
+		}
+	}()
+
 	// Start cleanup for old proof requests (every 10 minutes)
 	go func() {
 		requestTicker := time.NewTicker(10 * time.Minute)
@@ -996,12 +1037,30 @@ func startCleanupRoutines(redisQueue *server.RedisQueue) {
 		}
 	}()
 
-	// Start less frequent cleanup for old results (every 1 hour)
+	// Start cleanup for old failed jobs (every 30 minutes)
+	go func() {
+		failedTicker := time.NewTicker(30 * time.Minute)
+		defer failedTicker.Stop()
+
+		logging.Logger().Info().Msg("Started old failed jobs cleanup routine (every 30 minutes)")
+
+		for range failedTicker.C {
+			if err := redisQueue.CleanupOldFailedJobs(); err != nil {
+				logging.Logger().Error().
+					Err(err).
+					Msg("Failed to cleanup old failed jobs")
+			} else {
+				logging.Logger().Debug().Msg("Old failed jobs cleanup completed")
+			}
+		}
+	}()
+
+	// Start cleanup for old results (every 30 minutes)
 	go func() {
-		resultTicker := time.NewTicker(1 * time.Hour)
+		resultTicker := time.NewTicker(30 * time.Minute)
 		defer resultTicker.Stop()
 
-		logging.Logger().Info().Msg("Started old results cleanup routine (every 1 hour)")
+		logging.Logger().Info().Msg("Started old results cleanup routine (every 30 minutes)")
 
 		for range resultTicker.C {
 			if err := redisQueue.CleanupOldResults(); err != nil {
@@ -1011,6 +1070,14 @@ func startCleanupRoutines(redisQueue *server.RedisQueue) {
 			} else {
 				logging.Logger().Debug().Msg("Old results cleanup completed")
 			}
+
+			if err := redisQueue.CleanupOldResultKeys(); err != nil {
+				logging.Logger().Error().
+					Err(err).
+					Msg("Failed to cleanup old result keys")
+			} else {
+				logging.Logger().Debug().Msg("Old result keys cleanup completed")
+			}
 		}
 	}()
 }