fix: avoid any updates after table is closed (#998)

## Rationale
Part of #990.

Some background jobs are still allowed to execute, and it will lead to
data corrupted when a table is migrated between different nodes because
of multiple writers for the same table.

## Detailed Changes
Introduce a flag called `invalid` in the table data to denote whether
the serial executor is valid, and this flag is protected with the
`TableOpSerialExecutor` in table data, and the `TableOpSerialExecutor`
won't be acquired if the flag is set, that is to say, any table
operation including updating manifest, altering table and so on, can't
be executed after the flag is set because these operations require the
`TableOpSerialExecutor`. Finally, the flag will be set when the table is
closed.

Author: ShiKaiWi

analytic_engine/src/compaction/scheduler.rs

@@ -650,7 +650,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)