diff --git a/src/mito2/src/engine/close_test.rs b/src/mito2/src/engine/close_test.rs index 4c06583b0b..0092634ec7 100644 --- a/src/mito2/src/engine/close_test.rs +++ b/src/mito2/src/engine/close_test.rs @@ -12,11 +12,17 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::sync::Arc; +use std::sync::atomic::Ordering; + +use common_base::Plugins; use store_api::region_engine::RegionEngine; use store_api::region_request::{RegionCloseRequest, RegionRequest}; use store_api::storage::RegionId; use crate::config::MitoConfig; +use crate::engine::flush_test::MockRegionHook; +use crate::engine::region_hook::RegionHookRef; use crate::test_util::{CreateRequestBuilder, TestEnv}; #[tokio::test] @@ -60,3 +66,40 @@ async fn test_engine_close_region_with_format(flat_format: bool) { .await .unwrap(); } + +#[tokio::test] +async fn test_region_hook_on_close() { + common_telemetry::init_default_ut_logging(); + let mut env = TestEnv::new().await; + + let hook = Arc::new(MockRegionHook::new()); + let plugins = Plugins::new(); + plugins.insert(hook.clone() as RegionHookRef); + + let engine = env + .create_engine_with_plugins(MitoConfig::default(), plugins) + .await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new().build(); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Sanity: no lifecycle events before closing besides the create open. + assert_eq!(hook.opened_count.load(Ordering::Relaxed), 1); + assert_eq!(hook.closed_count.load(Ordering::Relaxed), 0); + assert_eq!(hook.dropped_count.load(Ordering::Relaxed), 0); + + engine + .handle_request(region_id, RegionRequest::Close(RegionCloseRequest {})) + .await + .unwrap(); + + // Closing fires on_region_closed exactly once. + assert_eq!(hook.closed_count.load(Ordering::Relaxed), 1); + // Close must not be confused with drop. + assert_eq!(hook.dropped_count.load(Ordering::Relaxed), 0); + assert_eq!(hook.files_removed_count.load(Ordering::Relaxed), 0); +} diff --git a/src/mito2/src/engine/drop_test.rs b/src/mito2/src/engine/drop_test.rs index a34a5d1172..90afd66ba3 100644 --- a/src/mito2/src/engine/drop_test.rs +++ b/src/mito2/src/engine/drop_test.rs @@ -13,9 +13,11 @@ // limitations under the License. use std::sync::Arc; +use std::sync::atomic::Ordering; use std::time::Duration; use api::v1::Rows; +use common_base::Plugins; use common_meta::key::SchemaMetadataManager; use common_meta::kv_backend::KvBackendRef; use object_store::util::join_path; @@ -25,7 +27,9 @@ use store_api::storage::RegionId; use crate::config::MitoConfig; use crate::engine::MitoEngine; +use crate::engine::flush_test::MockRegionHook; use crate::engine::listener::DropListener; +use crate::engine::region_hook::RegionHookRef; use crate::test_util::{ CreateRequestBuilder, TestEnv, build_rows_for_key, flush_region, put_rows, rows_schema, }; @@ -302,3 +306,75 @@ async fn test_engine_drop_region_for_custom_store_with_format(flat_format: bool) .unwrap() ); } + +#[tokio::test] +async fn test_region_hook_on_drop() { + common_telemetry::init_default_ut_logging(); + let mut env = TestEnv::with_prefix("drop_hook").await; + + let hook = Arc::new(MockRegionHook::new()); + let plugins = Plugins::new(); + plugins.insert(hook.clone() as RegionHookRef); + + let engine = env + .create_engine_with_plugins(MitoConfig::default(), plugins) + .await; + + let region_id = RegionId::new(1, 1); + env.get_schema_metadata_manager() + .register_region_table_info( + region_id.table_id(), + "test_table", + "test_catalog", + "test_schema", + None, + env.get_kv_backend(), + ) + .await; + + let request = CreateRequestBuilder::new().build(); + let column_schemas = rows_schema(&request); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Put some rows so the region has active data (kept in memtable; we + // intentionally do not flush so the drop GC worker finds no parquet files + // and removes the directory immediately, avoiding the multi-minute GC wait). + let rows = Rows { + schema: column_schemas, + rows: build_rows_for_key("a", 0, 2, 0), + }; + put_rows(&engine, region_id, rows).await; + + // Sanity: nothing fired yet besides the open from create. + assert_eq!(hook.opened_count.load(Ordering::Relaxed), 1); + assert_eq!(hook.dropped_count.load(Ordering::Relaxed), 0); + assert_eq!(hook.closed_count.load(Ordering::Relaxed), 0); + + // full drop (partial_drop = false) so the GC worker physically deletes the + // directory and fires on_region_files_removed. + engine + .handle_request( + region_id, + RegionRequest::Drop(RegionDropRequest { + fast_path: false, + force: false, + partial_drop: false, + }), + ) + .await + .unwrap(); + + // on_region_dropped fires inline while handling the drop request. + hook.wait_for_dropped().await; + assert_eq!(hook.dropped_count.load(Ordering::Relaxed), 1); + // Drop must not be confused with close. + assert_eq!(hook.closed_count.load(Ordering::Relaxed), 0); + + // on_region_files_removed fires from the background GC worker once the + // directory is physically deleted. + hook.wait_for_files_removed().await; + assert_eq!(hook.files_removed_count.load(Ordering::Relaxed), 1); +} diff --git a/src/mito2/src/engine/flush_test.rs b/src/mito2/src/engine/flush_test.rs index 4870e8c2bb..dc0063a7e6 100644 --- a/src/mito2/src/engine/flush_test.rs +++ b/src/mito2/src/engine/flush_test.rs @@ -895,24 +895,46 @@ async fn test_update_topic_latest_entry_id(factory: Option) { } #[derive(Debug)] -struct MockRegionHook { - sst_written_count: AtomicUsize, - manifest_updated_count: AtomicUsize, +pub(super) struct MockRegionHook { + pub(super) sst_written_count: AtomicUsize, + pub(super) manifest_updated_count: AtomicUsize, + pub(super) opened_count: AtomicUsize, + pub(super) closed_count: AtomicUsize, + pub(super) dropped_count: AtomicUsize, + pub(super) files_removed_count: AtomicUsize, notify: Notify, + dropped_notify: Notify, + files_removed_notify: Notify, } impl MockRegionHook { - fn new() -> Self { + pub(super) fn new() -> Self { Self { sst_written_count: AtomicUsize::new(0), manifest_updated_count: AtomicUsize::new(0), + opened_count: AtomicUsize::new(0), + closed_count: AtomicUsize::new(0), + dropped_count: AtomicUsize::new(0), + files_removed_count: AtomicUsize::new(0), notify: Notify::new(), + dropped_notify: Notify::new(), + files_removed_notify: Notify::new(), } } async fn wait_for_manifest_update(&self) { self.notify.notified().await; } + + /// Waits until `on_region_dropped` has fired at least once. + pub(super) async fn wait_for_dropped(&self) { + self.dropped_notify.notified().await; + } + + /// Waits until `on_region_files_removed` has fired at least once. + pub(super) async fn wait_for_files_removed(&self) { + self.files_removed_notify.notified().await; + } } #[async_trait] @@ -966,6 +988,35 @@ impl RegionHook for MockRegionHook { ); self.notify.notify_one(); } + + async fn on_region_opened(&self, region_id: RegionId, _region_metadata: &RegionMetadataRef) { + self.opened_count.fetch_add(1, Ordering::Relaxed); + common_telemetry::info!("MockRegionHook::on_region_opened: region={}", region_id); + } + + async fn on_region_closed(&self, region_id: RegionId, _region_metadata: &RegionMetadataRef) { + self.closed_count.fetch_add(1, Ordering::Relaxed); + common_telemetry::info!("MockRegionHook::on_region_closed: region={}", region_id); + } + + async fn on_region_dropped(&self, region_id: RegionId, _region_metadata: &RegionMetadataRef) { + self.dropped_count.fetch_add(1, Ordering::Relaxed); + common_telemetry::info!("MockRegionHook::on_region_dropped: region={}", region_id); + self.dropped_notify.notify_one(); + } + + async fn on_region_files_removed( + &self, + region_id: RegionId, + _region_metadata: &RegionMetadataRef, + ) { + self.files_removed_count.fetch_add(1, Ordering::Relaxed); + common_telemetry::info!( + "MockRegionHook::on_region_files_removed: region={}", + region_id + ); + self.files_removed_notify.notify_one(); + } } #[tokio::test] diff --git a/src/mito2/src/engine/open_test.rs b/src/mito2/src/engine/open_test.rs index cf4bfb1ff1..ee0d956009 100644 --- a/src/mito2/src/engine/open_test.rs +++ b/src/mito2/src/engine/open_test.rs @@ -32,6 +32,8 @@ use tokio::sync::oneshot; use crate::compaction::compactor::{OpenCompactionRegionRequest, open_compaction_region}; use crate::config::MitoConfig; +use crate::engine::flush_test::MockRegionHook; +use crate::engine::region_hook::RegionHookRef; use crate::error; use crate::region::opener::{PartitionExprFetcher, PartitionExprFetcherRef}; use crate::region::options::RegionOptions; @@ -805,3 +807,36 @@ async fn test_open_keeps_none_without_fetcher() { let meta = engine.get_region(region_id).unwrap().metadata(); assert!(meta.partition_expr.is_none()); } + +#[tokio::test] +async fn test_region_hook_on_open() { + common_telemetry::init_default_ut_logging(); + let mut env = TestEnv::with_prefix("open_hook").await; + + let hook = Arc::new(MockRegionHook::new()); + let plugins = Plugins::new(); + plugins.insert(hook.clone() as RegionHookRef); + + let engine = env + .create_engine_with_plugins(MitoConfig::default(), plugins) + .await; + + let region_id = RegionId::new(1, 1); + let request = CreateRequestBuilder::new().build(); + let table_dir = request.table_dir.clone(); + engine + .handle_request(region_id, RegionRequest::Create(request)) + .await + .unwrap(); + + // Creating a region fires on_region_opened exactly once. + assert_eq!(hook.opened_count.load(Ordering::Relaxed), 1); + assert_eq!(hook.closed_count.load(Ordering::Relaxed), 0); + + // `reopen_region` closes then opens; the open must fire the hook again. + reopen_region(&engine, region_id, table_dir, false, Default::default()).await; + + assert_eq!(hook.opened_count.load(Ordering::Relaxed), 2); + assert_eq!(hook.closed_count.load(Ordering::Relaxed), 1); + assert_eq!(hook.dropped_count.load(Ordering::Relaxed), 0); +} diff --git a/src/mito2/src/engine/region_hook.rs b/src/mito2/src/engine/region_hook.rs index d1b28f6eba..c4b683ca83 100644 --- a/src/mito2/src/engine/region_hook.rs +++ b/src/mito2/src/engine/region_hook.rs @@ -16,7 +16,8 @@ //! //! ## Design //! -//! The [`RegionHook`] trait provides two methods with clear separation of concerns: +//! The [`RegionHook`] trait observes region activity through two categories of +//! callbacks — manifest/file observation and region lifecycle: //! //! - [`on_sst_files_written`]: Fires when mito2 physically writes SST **data files**. //! Provides per-file [`SstInfo`] + [`FileMeta`]; metadata richness varies by path @@ -26,6 +27,10 @@ //! Receives the full [`RegionMetaActionList`] so consumers can inspect what changed //! (file additions/removals, schema changes, truncation, partition expression changes, etc.). //! +//! - [`on_region_opened`] / [`on_region_closed`] / [`on_region_dropped`] / [`on_region_files_removed`]: +//! Region **lifecycle** callbacks for open, close, logical drop, and physical file removal. +//! See [Region lifecycle](#region-lifecycle) below. +//! //! Hook implementations are registered via the [`Plugins`](common_base::Plugins) system: //! ```ignore //! plugins.insert(Arc::new(MyHook) as RegionHookRef); @@ -56,7 +61,39 @@ //! //! The following paths do **not** trigger any hook: //! - Follower region sync / catchup (manifest read-only; followers don't author changes) -//! - GC / checkpoint / drop / remap (internal bookkeeping, not logical state changes) +//! - GC / checkpoint / remap (internal bookkeeping, not logical state changes) +//! +//! An explicit region **drop** does fire lifecycle hooks — see +//! [Region lifecycle](#region-lifecycle). +//! +//! ## Region lifecycle +//! +//! Beyond manifest/SST observation, the hook observes the high-level lifecycle of an +//! active region: +//! +//! | Event | Method | When | +//! |-------|--------|------| +//! | Open | [`on_region_opened`] | A create or open request registers the region as active (the counterpart to close/drop). Does not fire for the compactor's transient regions or the catch-up reopen. | +//! | Close | [`on_region_closed`] | A close request (or a close-after-flush) removes the region from the active set. Data files, manifest and WAL state are **preserved**; the region may be reopened. | +//! | Logical drop | [`on_region_dropped`] | A drop request has been handled: the region leaves the active set and its WAL entries are marked obsolete. Data files are **not yet deleted**. | +//! | Physical file removal | [`on_region_files_removed`] | The drop GC worker has deleted the region directory. Terminal file-lifecycle event. | +//! +//! Notes: +//! - `on_region_closed` / `on_region_dropped` run **inline in the region worker loop**, +//! so implementations must be fast (same contract as `on_manifest_updated`). +//! - `on_region_opened` runs inline in the worker loop on the **create** path, but on the +//! **open** path it fires inside the spawned open task (`common_runtime::spawn_global`), +//! i.e. concurrently with the worker loop — after WAL replay, before the region is +//! registered and its open request is acknowledged. Implementations must still be fast +//! and must not assume worker-loop-thread affinity or strict ordering against concurrent +//! requests to other regions. +//! - `on_region_files_removed` runs on the background drop GC task, outside the worker loop. +//! - When global GC is enabled and a normal table region is dropped with `partial_drop`, its +//! directory is left for global reclamation and `on_region_files_removed` is **not** fired +//! by the drop worker (observe it via the global GC path instead). +//! - Logical file removal (compaction, region edit, truncate) is already observable via +//! [`on_manifest_updated`] (`Edit.files_to_remove` / `Truncate` action); only the drop +//! worker's physical directory deletion needs a dedicated file hook. //! //! ## Invocation points //! @@ -77,11 +114,18 @@ //! //! ## Future work //! -//! A future `on_files_removed` hook may be added to observe file lifecycle end -//! (GC, drop, truncate, compaction removal). This is not yet implemented. +//! `on_region_files_removed` currently covers only the **drop** GC worker's physical +//! directory removal. A broader per-file `on_files_removed` hook covering compaction +//! removal, truncate, and the global GC reclamation path is not yet implemented +//! (though logical file removal is already observable via `on_manifest_updated`). +//! Role/leadership transitions (`on_region_role_changed`) are also not hooked. //! //! [`on_sst_files_written`]: RegionHook::on_sst_files_written //! [`on_manifest_updated`]: RegionHook::on_manifest_updated +//! [`on_region_opened`]: RegionHook::on_region_opened +//! [`on_region_closed`]: RegionHook::on_region_closed +//! [`on_region_dropped`]: RegionHook::on_region_dropped +//! [`on_region_files_removed`]: RegionHook::on_region_files_removed //! [`RegionManifestManager::update`]: crate::manifest::manager::RegionManifestManager::update //! [`ManifestContext::update_locked`]: crate::region::ManifestContext::update_locked //! [`ManifestContext::update_manifest`]: crate::region::ManifestContext::update_manifest @@ -228,6 +272,80 @@ pub trait RegionHook: Send + Sync + Debug { ) { let _ = (region_id, action_list, manifest_version); } + + /// Called once a region **open** or **create** succeeds, but **before** the + /// region is registered in the engine's active set (`insert_region` runs + /// immediately afterwards). + /// + /// Fires once when a region becomes active via a create or open request — + /// the natural counterpart to [`on_region_closed`] / [`on_region_dropped`]. + /// It does **not** fire for the compactor's transient compaction regions + /// (`open_compaction_region`), nor for the internal reopen performed during + /// follower catch-up / leadership promotion. + /// + /// On the **create** path it runs inline in the region worker loop; on the + /// **open** path it runs inside the spawned open task + /// (`common_runtime::spawn_global`), concurrently with the worker loop + /// (after WAL replay, before the region is registered/acknowledged). + /// Implementations must be fast and must **not** assume worker-loop-thread + /// affinity or strict ordering against concurrent requests to other regions. + /// + /// [`on_region_closed`]: RegionHook::on_region_closed + /// [`on_region_dropped`]: RegionHook::on_region_dropped + async fn on_region_opened(&self, region_id: RegionId, region_metadata: &RegionMetadataRef) { + let _ = (region_id, region_metadata); + } + + /// Called after a region is **closed** via a close request. + /// + /// The region is removed from the engine's active set, but its data files, + /// manifest, and WAL state are **preserved**; the region may be reopened + /// later. Fires once per successful close, after the region's background + /// tasks (flush/compaction) have been stopped. + /// + /// Fires for a region of **any** role (leader or follower) that is closed. + /// Does **not** fire when a region is dropped (see [`on_region_dropped`]). + /// + /// Runs inline in the region worker loop; implementations should be fast. + /// + /// [`on_region_dropped`]: RegionHook::on_region_dropped + async fn on_region_closed(&self, region_id: RegionId, region_metadata: &RegionMetadataRef) { + let _ = (region_id, region_metadata); + } + + /// Called after a region is **logically dropped** (a drop request has been + /// handled). + /// + /// The region is removed from the active set and its WAL entries are marked + /// obsolete. Its data files are **not yet deleted** — they are scheduled for + /// asynchronous removal by the GC worker. Observe physical deletion via + /// [`on_region_files_removed`]. + /// + /// Runs inline in the region worker loop; implementations should be fast. + /// + /// [`on_region_files_removed`]: RegionHook::on_region_files_removed + async fn on_region_dropped(&self, region_id: RegionId, region_metadata: &RegionMetadataRef) { + let _ = (region_id, region_metadata); + } + + /// Called after a dropped region's data files are **physically removed** by + /// the drop GC worker (the region directory has been deleted). + /// + /// This is the terminal event in a region's file lifecycle; no further + /// callbacks fire for this region id afterwards. Fires only when the drop + /// worker itself deletes the directory. When global GC is enabled and the + /// region is a normal table region dropped with `partial_drop`, the + /// directory is left for global reclamation and this hook is **not** fired + /// by the drop worker. + /// + /// Runs on a background task, outside the region worker loop. + async fn on_region_files_removed( + &self, + region_id: RegionId, + region_metadata: &RegionMetadataRef, + ) { + let _ = (region_id, region_metadata); + } } pub type RegionHookRef = Arc; diff --git a/src/mito2/src/worker/handle_close.rs b/src/mito2/src/worker/handle_close.rs index a616a0e32d..b582c27692 100644 --- a/src/mito2/src/worker/handle_close.rs +++ b/src/mito2/src/worker/handle_close.rs @@ -89,5 +89,13 @@ impl RegionWorkerLoop { // clean index build status. self.index_build_scheduler.on_region_closed(region_id).await; self.region_count.dec(); + + // Notify the region hook that the region has been closed. The region is + // fully stopped and unregistered, but its files/manifest are preserved. + // Runs inline; the hook contract requires it to be fast. + if let Some(hook) = region.manifest_ctx.hook() { + let metadata = region.metadata(); + hook.on_region_closed(region_id, &metadata).await; + } } } diff --git a/src/mito2/src/worker/handle_create.rs b/src/mito2/src/worker/handle_create.rs index 6e43e74a0e..bf254dc1e6 100644 --- a/src/mito2/src/worker/handle_create.rs +++ b/src/mito2/src/worker/handle_create.rs @@ -86,6 +86,12 @@ impl RegionWorkerLoop { self.region_count.inc(); + // Notify the region hook that the region has been opened (created). + // Fires before registration; allocates nothing when no hook is registered. + if let Some(hook) = region.manifest_ctx.hook() { + hook.on_region_opened(region_id, ®ion.metadata()).await; + } + // Insert the MitoRegion into the RegionMap. self.regions.insert_region(region); diff --git a/src/mito2/src/worker/handle_drop.rs b/src/mito2/src/worker/handle_drop.rs index 00a96b0d16..3d111776b8 100644 --- a/src/mito2/src/worker/handle_drop.rs +++ b/src/mito2/src/worker/handle_drop.rs @@ -23,10 +23,12 @@ use object_store::util::join_path; use object_store::{EntryMode, ObjectStore}; use snafu::ResultExt; use store_api::logstore::LogStore; +use store_api::metadata::RegionMetadataRef; use store_api::region_request::{AffectedRows, PathType}; use store_api::storage::RegionId; use tokio::time::sleep; +use crate::engine::region_hook::RegionHookRef; use crate::error::{OpenDalSnafu, Result}; use crate::region::{RegionLeaderState, RegionMapRef}; use crate::worker::{DROPPING_MARKER_FILE, RegionWorkerLoop}; @@ -106,6 +108,22 @@ where self.region_count.dec(); + // Notify registered hooks that the region has been logically dropped, and + // prepare a payload for the background GC task to fire the terminal + // `on_region_files_removed`. When no hook is registered this allocates + // nothing — no metadata snapshot is taken and no payload is built. + let hook_payload = match region.manifest_ctx.hook() { + Some(hook) => { + let region_metadata = region.metadata(); + hook.on_region_dropped(region_id, ®ion_metadata).await; + Some(DropHookPayload { + hook, + metadata: region_metadata, + }) + } + None => None, + }; + let object_store = region.access_layer.object_store().clone(); let dropping_regions = self.dropping_regions.clone(); let listener = self.listener.clone(); @@ -122,6 +140,7 @@ where object_store, dropping_regions, partial_drop, + hook_payload.as_ref(), ) .await } else { @@ -135,6 +154,7 @@ where object_store, dropping_regions, gc_duration, + hook_payload.as_ref(), ) .await }; @@ -156,6 +176,19 @@ where } } +/// Carries the region hook and region metadata into the background GC task so +/// it can fire [`RegionHook::on_region_files_removed`] once the dropped region's +/// directory is physically deleted. +/// +/// Only constructed when a hook is registered; the task receives it as +/// `Option<&DropHookPayload>` so the no-hook path allocates nothing. +/// +/// [`RegionHook::on_region_files_removed`]: crate::engine::region_hook::RegionHook::on_region_files_removed +struct DropHookPayload { + hook: RegionHookRef, + metadata: RegionMetadataRef, +} + /// Background GC task to remove the entire region path once one of the following /// conditions is true: /// - It finds there is no parquet file left. @@ -172,6 +205,7 @@ async fn later_drop_task_without_global_gc( object_store: ObjectStore, dropping_regions: RegionMapRef, gc_duration: Duration, + hook_payload: Option<&DropHookPayload>, ) -> bool { remove_region_with_retry( region_id, @@ -180,6 +214,7 @@ async fn later_drop_task_without_global_gc( dropping_regions, Some(gc_duration), false, + hook_payload, ) .await } @@ -191,6 +226,7 @@ async fn remove_region_with_retry( dropping_regions: std::sync::Arc, gc_duration: Option, mut force: bool, + hook_payload: Option<&DropHookPayload>, ) -> bool { for _ in 0..MAX_RETRY_TIMES { let result = remove_region_dir_once(®ion_path, &object_store, force).await; @@ -204,6 +240,15 @@ async fn remove_region_with_retry( Ok(true) => { dropping_regions.remove_region(region_id); info!("Region {} is dropped, force: {}", region_path, force); + // The region directory has been physically deleted; fire the + // terminal file-lifecycle event. The partial-drop/global-GC path + // never reaches here (it does not delete the directory itself). + if let Some(hook_payload) = hook_payload { + hook_payload + .hook + .on_region_files_removed(region_id, &hook_payload.metadata) + .await; + } return true; } Ok(false) => (), @@ -230,6 +275,7 @@ async fn later_drop_task_with_global_gc( object_store: ObjectStore, dropping_regions: RegionMapRef, partial_drop: bool, + hook_payload: Option<&DropHookPayload>, ) -> bool { // For metadata regions or regions marked for full deletion (such as when dropping a table) // the region directory is forcefully removed immediately. @@ -243,6 +289,7 @@ async fn later_drop_task_with_global_gc( dropping_regions, None, true, + hook_payload, ) .await } else { diff --git a/src/mito2/src/worker/handle_open.rs b/src/mito2/src/worker/handle_open.rs index 3d3520b0cc..48fef5ddd6 100644 --- a/src/mito2/src/worker/handle_open.rs +++ b/src/mito2/src/worker/handle_open.rs @@ -149,6 +149,13 @@ impl RegionWorkerLoop { ); region_count.inc(); + // Notify the region hook that the region has been opened. + // Fires before registration; allocates nothing when no hook + // is registered. + if let Some(hook) = region.manifest_ctx.hook() { + hook.on_region_opened(region_id, ®ion.metadata()).await; + } + // Insert the Region into the RegionMap. regions.insert_region(region);