fix: avoid any updates after table is closed

analytic_engine/src/compaction/scheduler.rs

@@ -656,7 +656,16 @@ impl ScheduleWorker {
                     self.max_unflushed_duration,
                 );
 
-                let mut serial_exec = table_data.serial_exec.lock().await;
+                let mut serial_exec = if let Some(v) = table_data.acquire_serial_exec_ctx().await {
+                    v
+                } else {
+                    warn!(
+                        "Table is closed, ignore this periodical flush, table:{}",
+                        table_data.name
+                    );
+                    continue;
+                };
+
                 let flush_scheduler = serial_exec.flush_scheduler();
                 // Instance flush the table asynchronously.
                 if let Err(e) = flusher

analytic_engine/src/instance/close.rs

@@ -3,12 +3,12 @@
 //! Close table logic of instance
 
 use log::{info, warn};
-use snafu::ResultExt;
+use snafu::{OptionExt, ResultExt};
 use table_engine::engine::CloseTableRequest;
 
 use crate::{
     instance::{
-        engine::{DoManifestSnapshot, FlushTable, Result},
+        engine::{DoManifestSnapshot, FlushTable, OperateClosedTable, Result},
         flush_compaction::{Flusher, TableFlushOptions},
     },
     manifest::{ManifestRef, SnapshotRequest},
@@ -37,8 +37,11 @@ impl Closer {
 
         // Flush table.
         let opts = TableFlushOptions::default();
-        let mut serial_exec = table_data.serial_exec.lock().await;
-        let flush_scheduler = serial_exec.flush_scheduler();
+        let mut serial_exec_ctx = table_data
+            .acquire_serial_exec_ctx()
+            .await
+            .context(OperateClosedTable)?;
+        let flush_scheduler = serial_exec_ctx.flush_scheduler();
 
         self.flusher
             .do_flush(flush_scheduler, &table_data, opts)
@@ -67,9 +70,10 @@ impl Closer {
         let removed_table = self.space.remove_table(&request.table_name);
         assert!(removed_table.is_some());
 
+        serial_exec_ctx.invalidate();
         info!(
-            "table:{}-{} has been removed from the space_id:{}",
-            table_data.name, table_data.id, self.space.id
+            "table:{} has been removed from the space_id:{}, table_id:{}",
+            table_data.name, self.space.id, table_data.id,
         );
         Ok(())
     }

analytic_engine/src/instance/drop.rs

@@ -3,12 +3,12 @@
 //! Drop table logic of instance
 
 use log::{info, warn};
-use snafu::ResultExt;
+use snafu::{OptionExt, ResultExt};
 use table_engine::engine::DropTableRequest;
 
 use crate::{
     instance::{
-        engine::{FlushTable, Result, WriteManifest},
+        engine::{FlushTable, OperateClosedTable, Result, WriteManifest},
         flush_compaction::{Flusher, TableFlushOptions},
         SpaceStoreRef,
     },
@@ -36,7 +36,10 @@ impl Dropper {
             }
         };
 
-        let mut serial_exec = table_data.serial_exec.lock().await;
+        let mut serial_exec_ctx = table_data
+            .acquire_serial_exec_ctx()
+            .await
+            .context(OperateClosedTable)?;
 
         if table_data.is_dropped() {
             warn!(
@@ -51,7 +54,7 @@ impl Dropper {
         //  be avoided.
 
         let opts = TableFlushOptions::default();
-        let flush_scheduler = serial_exec.flush_scheduler();
+        let flush_scheduler = serial_exec_ctx.flush_scheduler();
         self.flusher
             .do_flush(flush_scheduler, &table_data, opts)
             .await

analytic_engine/src/instance/engine.rs

@@ -23,44 +23,33 @@ use crate::{
 #[derive(Debug, Snafu)]
 #[snafu(visibility(pub(crate)))]
 pub enum Error {
-    #[snafu(display(
-        "The space of the table does not exist, space_id:{}, table:{}.\nBacktrace:\n{}",
-        space_id,
-        table,
-        backtrace,
-    ))]
+    #[snafu(display("The space of the table does not exist, space_id:{space_id}, table:{table}.\nBacktrace:\n{backtrace}"))]
     SpaceNotExist {
         space_id: SpaceId,
         table: String,
         backtrace: Backtrace,
     },
 
-    #[snafu(display("Failed to read meta update, table_id:{}, err:{}", table_id, source))]
+    #[snafu(display("Failed to read meta update, table_id:{table_id}, err:{source}"))]
     ReadMetaUpdate {
         table_id: TableId,
         source: GenericError,
     },
 
     #[snafu(display(
-        "Failed to recover table data, space_id:{}, table:{}, err:{}",
-        space_id,
-        table,
-        source
+        "Failed to recover table data, space_id:{space_id}, table:{table}, err:{source}"
     ))]
     RecoverTableData {
         space_id: SpaceId,
         table: String,
         source: crate::table::data::Error,
     },
 
-    #[snafu(display("Failed to read wal, err:{}", source))]
+    #[snafu(display("Failed to read wal, err:{source}"))]
     ReadWal { source: wal::manager::Error },
 
     #[snafu(display(
-        "Failed to apply log entry to memtable, table:{}, table_id:{}, err:{}",
-        table,
-        table_id,
-        source
+        "Failed to apply log entry to memtable, table:{table}, table_id:{table_id}, err:{source}",
     ))]
     ApplyMemTable {
         space_id: SpaceId,
@@ -70,11 +59,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Flush failed, space_id:{}, table:{}, table_id:{}, err:{}",
-        space_id,
-        table,
-        table_id,
-        source
+        "Flush failed, space_id:{space_id}, table:{table}, table_id:{table_id}, err:{source}",
     ))]
     FlushTable {
         space_id: SpaceId,
@@ -84,11 +69,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to persist meta update to manifest, space_id:{}, table:{}, table_id:{}, err:{}",
-        space_id,
-        table,
-        table_id,
-        source
+        "Failed to persist meta update to manifest, space_id:{space_id}, table:{table}, table_id:{table_id}, err:{source}",
     ))]
     WriteManifest {
         space_id: SpaceId,
@@ -98,11 +79,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to persist meta update to WAL, space_id:{}, table:{}, table_id:{}, err:{}",
-        space_id,
-        table,
-        table_id,
-        source
+        "Failed to persist meta update to WAL, space_id:{space_id}, table:{table}, table_id:{table_id}, err:{source}",
     ))]
     WriteWal {
         space_id: SpaceId,
@@ -112,11 +89,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Invalid options, space_id:{}, table:{}, table_id:{}, err:{}",
-        space_id,
-        table,
-        table_id,
-        source
+        "Invalid options, space_id:{space_id}, table:{table}, table_id:{table_id}, err:{source}",
     ))]
     InvalidOptions {
         space_id: SpaceId,
@@ -126,11 +99,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to create table data, space_id:{}, table:{}, table_id:{}, err:{}",
-        space_id,
-        table,
-        table_id,
-        source
+        "Failed to create table data, space_id:{space_id}, table:{table}, table_id:{table_id}, err:{source}",
     ))]
     CreateTableData {
         space_id: SpaceId,
@@ -140,11 +109,8 @@ pub enum Error {
     },
 
     #[snafu(display(
-    "Try to update schema to elder version, table:{}, current_version:{}, given_version:{}.\nBacktrace:\n{}",
-    table,
-    current_version,
-    given_version,
-    backtrace,
+        "Try to update schema to elder version, table:{table}, current_version:{current_version}, \
+        given_version:{given_version}.\nBacktrace:\n{backtrace}",
     ))]
     InvalidSchemaVersion {
         table: String,
@@ -154,11 +120,8 @@ pub enum Error {
     },
 
     #[snafu(display(
-    "Invalid previous schema version, table:{}, current_version:{}, pre_version:{}.\nBacktrace:\n{}",
-    table,
-    current_version,
-    pre_version,
-    backtrace,
+        "Invalid previous schema version, table:{table}, current_version:{current_version}, \
+        pre_version:{pre_version}.\nBacktrace:\n{backtrace}",
     ))]
     InvalidPreVersion {
         table: String,
@@ -167,21 +130,14 @@ pub enum Error {
         backtrace: Backtrace,
     },
 
-    #[snafu(display(
-        "Alter schema of a dropped table:{}.\nBacktrace:\n{}",
-        table,
-        backtrace
-    ))]
+    #[snafu(display("Alter schema of a dropped table:{table}.\nBacktrace:\n{backtrace}"))]
     AlterDroppedTable { table: String, backtrace: Backtrace },
 
-    #[snafu(display("Failed to store version edit, err:{}", source))]
+    #[snafu(display("Failed to store version edit, err:{source}"))]
     StoreVersionEdit { source: GenericError },
 
     #[snafu(display(
-        "Failed to encode payloads, table:{}, wal_location:{:?}, err:{}",
-        table,
-        wal_location,
-        source
+        "Failed to encode payloads, table:{table}, wal_location:{wal_location:?}, err:{source}"
     ))]
     EncodePayloads {
         table: String,
@@ -190,10 +146,7 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Failed to do manifest snapshot for table, space_id:{}, table:{}, err:{}",
-        space_id,
-        table,
-        source
+        "Failed to do manifest snapshot for table, space_id:{space_id}, table:{table}, err:{source}",
     ))]
     DoManifestSnapshot {
         space_id: SpaceId,
@@ -202,30 +155,31 @@ pub enum Error {
     },
 
     #[snafu(display(
-        "Table open failed and can not be created again, table:{}.\nBacktrace:\n{}",
-        table,
-        backtrace,
+        "Table open failed and can not be created again, table:{table}.\nBacktrace:\n{backtrace}",
     ))]
     CreateOpenFailedTable { table: String, backtrace: Backtrace },
 
-    #[snafu(display("Failed to open manifest, err:{}", source))]
+    #[snafu(display("Failed to open manifest, err:{source}"))]
     OpenManifest {
         source: crate::manifest::details::Error,
     },
 
-    #[snafu(display("Failed to find table, msg:{}.\nBacktrace:\n{}", msg, backtrace))]
+    #[snafu(display("Failed to find table, msg:{msg}.\nBacktrace:\n{backtrace}"))]
     TableNotExist { msg: String, backtrace: Backtrace },
 
-    #[snafu(display("Failed to open shard, msg:{}.\nBacktrace:\n{}", msg, backtrace))]
+    #[snafu(display("Failed to open shard, msg:{msg}.\nBacktrace:\n{backtrace}"))]
     OpenTablesOfShard { msg: String, backtrace: Backtrace },
 
-    #[snafu(display("Failed to replay wal, msg:{:?}, err:{}", msg, source))]
+    #[snafu(display("Try to operate a closed table.\nBacktrace:\n{backtrace}"))]
+    OperateClosedTable { backtrace: Backtrace },
+
+    #[snafu(display("Failed to replay wal, msg:{msg:?}, err:{source}"))]
     ReplayWalWithCause {
         msg: Option<String>,
         source: GenericError,
     },
 
-    #[snafu(display("Failed to replay wal, msg:{:?}.\nBacktrace:\n{}", msg, backtrace))]
+    #[snafu(display("Failed to replay wal, msg:{msg:?}.\nBacktrace:\n{backtrace}"))]
     ReplayWalNoCause {
         msg: Option<String>,
         backtrace: Backtrace,
@@ -264,6 +218,7 @@ impl From<Error> for table_engine::engine::Error {
             | Error::TableNotExist { .. }
             | Error::OpenTablesOfShard { .. }
             | Error::ReplayWalNoCause { .. }
+            | Error::OperateClosedTable { .. }
             | Error::ReplayWalWithCause { .. } => Self::Unexpected {
                 source: Box::new(err),
             },

analytic_engine/src/instance/mod.rs

@@ -28,14 +28,14 @@ use common_util::{
 };
 use log::{error, info};
 use mem_collector::MemUsageCollector;
-use snafu::{ResultExt, Snafu};
+use snafu::{Backtrace, OptionExt, ResultExt, Snafu};
 use table_engine::{engine::EngineRuntimes, table::FlushRequest};
 use tokio::sync::oneshot::{self, error::RecvError};
 use wal::manager::{WalLocation, WalManagerRef};
 
-use self::flush_compaction::{Flusher, TableFlushOptions};
 use crate::{
     compaction::{scheduler::CompactionSchedulerRef, TableCompactionRequest},
+    instance::flush_compaction::{Flusher, TableFlushOptions},
     manifest::ManifestRef,
     row_iter::IterOptions,
     space::{SpaceId, SpaceRef, SpacesRef},
@@ -66,6 +66,9 @@ pub enum Error {
         source: GenericError,
     },
 
+    #[snafu(display("Try to operate a closed table, table:{table}.\nBacktrace:\n{backtrace}"))]
+    OperateClosedTable { table: String, backtrace: Backtrace },
+
     #[snafu(display("Failed to receive {} result, table:{}, err:{}", op, table, source))]
     RecvManualOpResult {
         op: String,
@@ -195,7 +198,13 @@ impl Instance {
         };
 
         let flusher = self.make_flusher();
-        let mut serial_exec = table_data.serial_exec.lock().await;
+        let mut serial_exec =
+            table_data
+                .acquire_serial_exec_ctx()
+                .await
+                .context(OperateClosedTable {
+                    table: &table_data.name,
+                })?;
         let flush_scheduler = serial_exec.flush_scheduler();
         flusher
             .schedule_flush(flush_scheduler, table_data, flush_opts)