fix(cubestore): Immediately restart stale streaming jobs to avoid processing gaps for sparse streams

Author: paveltiunov

rust/cubestore/cubestore/src/scheduler/mod.rs

@@ -1,6 +1,6 @@
 use crate::cluster::{pick_worker_by_ids, Cluster};
 use crate::config::ConfigObj;
-use crate::metastore::job::{Job, JobType};
+use crate::metastore::job::{Job, JobStatus, JobType};
 use crate::metastore::partition::partition_file_name;
 use crate::metastore::table::Table;
 use crate::metastore::{
@@ -490,6 +490,21 @@ impl SchedulerImpl {
                 }
             }
         }
+        if let MetaStoreEvent::UpdateJob(_, new_job) = &event {
+            match new_job.get_row().job_type() {
+                JobType::TableImportCSV(location) if Table::is_stream_location(location) => {
+                    match new_job.get_row().status() {
+                        JobStatus::Error(e) if e.contains("Stale stream timeout") => {
+                            log::info!("Removing stale stream job: {:?}", new_job);
+                            self.meta_store.delete_job(new_job.get_id()).await?;
+                            self.reconcile_table_imports().await?;
+                        }
+                        _ => {}
+                    }
+                }
+                _ => {}
+            }
+        }
         if let MetaStoreEvent::DeleteJob(job) = event {
             match job.get_row().job_type() {
                 JobType::RepartitionChunk => match job.get_row().row_reference() {

rust/cubestore/cubestore/src/streaming/mod.rs

@@ -118,7 +118,8 @@ impl StreamingService for StreamingServiceImpl {
             Duration::from_secs(self.config_obj.stale_stream_timeout()),
             stream.next(),
         )
-        .await?
+        .await
+        .map_err(|e| CubeError::user(format!("Stale stream timeout: {}", e)))?
         {
             let rows = new_rows?;
             debug!("Received {} rows for {}", rows.len(), location);