refactor(targets): unify queue/connectivity handling and coverage (#2953)

Co-authored-by: Claude Opus 4.7 <noreply@anthropic.com>
Co-authored-by: marshawcoco <marshawcoco@gmail.com>

crates/targets/src/runtime/adapter.rs

@@ -0,0 +1,346 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::{
+    ReplayEvent, ReplayWorkerManager, RuntimeActivation, RuntimeStatusSnapshot, RuntimeTargetHealthSnapshot,
+    TargetRuntimeManager, activate_targets_with_replay, init_target_and_optionally_start_replay, start_replay_worker,
+};
+use crate::{Target, TargetError};
+use async_trait::async_trait;
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use std::future::Future;
+use std::pin::Pin;
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::sync::Semaphore;
+
+type ReplayHook<E> = Arc<dyn Fn(ReplayEvent<E>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;
+type ReplayStartObserver = Arc<dyn Fn(&str, bool) + Send + Sync>;
+
+/// Shared runtime contract for target plugins.
+#[async_trait]
+pub trait PluginRuntimeAdapter<E>: Send + Sync
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    async fn activate_with_replay(&self, targets: Vec<Box<dyn Target<E> + Send + Sync>>) -> RuntimeActivation<E>;
+
+    async fn replace_runtime_targets(
+        &self,
+        runtime: &mut TargetRuntimeManager<E>,
+        replay_workers: &mut ReplayWorkerManager,
+        activation: RuntimeActivation<E>,
+    ) -> Result<(), TargetError>;
+
+    async fn stop_replay_workers(&self, replay_workers: &mut ReplayWorkerManager);
+
+    fn snapshot_runtime_status(
+        &self,
+        runtime: &TargetRuntimeManager<E>,
+        replay_workers: &ReplayWorkerManager,
+    ) -> RuntimeStatusSnapshot;
+
+    async fn snapshot_runtime_health(&self, runtime: &TargetRuntimeManager<E>) -> Vec<RuntimeTargetHealthSnapshot>;
+
+    async fn shutdown(
+        &self,
+        runtime: &mut TargetRuntimeManager<E>,
+        replay_workers: &mut ReplayWorkerManager,
+    ) -> Result<(), TargetError>;
+}
+
+/// Built-in in-process runtime adapter that preserves the current replay and
+/// activation behavior while presenting a stable runtime contract to callers.
+#[derive(Clone)]
+pub struct BuiltinPluginRuntimeAdapter<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    replay_hook: ReplayHook<E>,
+    replay_start_observer: ReplayStartObserver,
+    replay_semaphore: Option<Arc<Semaphore>>,
+    batch_timeout: Duration,
+    idle_sleep: Duration,
+    stop_log_prefix: Arc<str>,
+}
+
+impl<E> BuiltinPluginRuntimeAdapter<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    pub fn new(
+        replay_hook: ReplayHook<E>,
+        replay_start_observer: ReplayStartObserver,
+        replay_semaphore: Option<Arc<Semaphore>>,
+        batch_timeout: Duration,
+        idle_sleep: Duration,
+        stop_log_prefix: impl Into<Arc<str>>,
+    ) -> Self {
+        Self {
+            replay_hook,
+            replay_start_observer,
+            replay_semaphore,
+            batch_timeout,
+            idle_sleep,
+            stop_log_prefix: stop_log_prefix.into(),
+        }
+    }
+}
+
+#[async_trait]
+impl<E> PluginRuntimeAdapter<E> for BuiltinPluginRuntimeAdapter<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    async fn activate_with_replay(&self, targets: Vec<Box<dyn Target<E> + Send + Sync>>) -> RuntimeActivation<E> {
+        let replay_hook = Arc::clone(&self.replay_hook);
+        let replay_start_observer = Arc::clone(&self.replay_start_observer);
+        let replay_semaphore = self.replay_semaphore.clone();
+        let batch_timeout = self.batch_timeout;
+        let idle_sleep = self.idle_sleep;
+
+        activate_targets_with_replay(targets, move |target| {
+            let replay_hook = Arc::clone(&replay_hook);
+            let replay_start_observer = Arc::clone(&replay_start_observer);
+            let replay_semaphore = replay_semaphore.clone();
+
+            async move {
+                init_target_and_optionally_start_replay(
+                    target,
+                    move |target_id, has_replay| replay_start_observer(target_id, has_replay),
+                    move |store, target| {
+                        start_replay_worker(
+                            store,
+                            target,
+                            Arc::clone(&replay_hook),
+                            replay_semaphore.clone(),
+                            batch_timeout,
+                            idle_sleep,
+                        )
+                    },
+                )
+                .await
+            }
+        })
+        .await
+    }
+
+    async fn replace_runtime_targets(
+        &self,
+        runtime: &mut TargetRuntimeManager<E>,
+        replay_workers: &mut ReplayWorkerManager,
+        activation: RuntimeActivation<E>,
+    ) -> Result<(), TargetError> {
+        self.stop_replay_workers(replay_workers).await;
+        runtime.clear_and_close().await;
+
+        for target in activation.targets {
+            runtime.add_arc(target);
+        }
+
+        *replay_workers = activation.replay_workers;
+        Ok(())
+    }
+
+    async fn stop_replay_workers(&self, replay_workers: &mut ReplayWorkerManager) {
+        replay_workers.stop_all(&self.stop_log_prefix).await;
+    }
+
+    fn snapshot_runtime_status(
+        &self,
+        runtime: &TargetRuntimeManager<E>,
+        replay_workers: &ReplayWorkerManager,
+    ) -> RuntimeStatusSnapshot {
+        runtime.status_snapshot(replay_workers)
+    }
+
+    async fn snapshot_runtime_health(&self, runtime: &TargetRuntimeManager<E>) -> Vec<RuntimeTargetHealthSnapshot> {
+        runtime.health_snapshots().await
+    }
+
+    async fn shutdown(
+        &self,
+        runtime: &mut TargetRuntimeManager<E>,
+        replay_workers: &mut ReplayWorkerManager,
+    ) -> Result<(), TargetError> {
+        self.stop_replay_workers(replay_workers).await;
+        runtime.clear_and_close().await;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{BuiltinPluginRuntimeAdapter, PluginRuntimeAdapter};
+    use crate::arn::TargetID;
+    use crate::store::{Key, QueueStore, Store};
+    use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta};
+    use crate::{StoreError, Target, TargetError};
+    use async_trait::async_trait;
+    use serde::{Serialize, de::DeserializeOwned};
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::time::Duration;
+    use tempfile::tempdir;
+
+    #[derive(Clone)]
+    struct TestTarget {
+        close_calls: Arc<AtomicUsize>,
+        id: TargetID,
+        init_fails: bool,
+        store: Option<QueueStore<QueuedPayload>>,
+    }
+
+    impl TestTarget {
+        fn new(id: &str, name: &str) -> Self {
+            Self {
+                close_calls: Arc::new(AtomicUsize::new(0)),
+                id: TargetID::new(id.to_string(), name.to_string()),
+                init_fails: false,
+                store: None,
+            }
+        }
+
+        fn with_failed_init(mut self) -> Self {
+            self.init_fails = true;
+            self
+        }
+
+        fn with_store(mut self) -> Self {
+            let dir = tempdir().expect("tempdir should be created for queue store tests");
+            let store = QueueStore::<QueuedPayload>::new(dir.path(), 16, ".queue");
+            store.open().expect("queue store should open");
+            self.store = Some(store);
+            self
+        }
+    }
+
+    #[async_trait]
+    impl<E> Target<E> for TestTarget
+    where
+        E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+    {
+        fn id(&self) -> TargetID {
+            self.id.clone()
+        }
+
+        async fn is_active(&self) -> Result<bool, TargetError> {
+            Ok(true)
+        }
+
+        async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
+            Ok(())
+        }
+
+        async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
+            Ok(())
+        }
+
+        async fn close(&self) -> Result<(), TargetError> {
+            self.close_calls.fetch_add(1, Ordering::SeqCst);
+            Ok(())
+        }
+
+        fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
+            self.store
+                .as_ref()
+                .map(|store| store as &(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync))
+        }
+
+        fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
+            Box::new(self.clone())
+        }
+
+        async fn init(&self) -> Result<(), TargetError> {
+            if self.init_fails {
+                return Err(TargetError::Configuration("forced init failure".to_string()));
+            }
+            Ok(())
+        }
+
+        fn is_enabled(&self) -> bool {
+            true
+        }
+    }
+
+    fn builtin_adapter() -> BuiltinPluginRuntimeAdapter<String> {
+        BuiltinPluginRuntimeAdapter::new(
+            Arc::new(|_event| Box::pin(async {})),
+            Arc::new(|_target_id, _has_replay| {}),
+            None,
+            Duration::from_millis(10),
+            Duration::from_millis(10),
+            "stopping test replay worker",
+        )
+    }
+
+    #[tokio::test]
+    async fn builtin_adapter_handles_empty_target_activation() {
+        let adapter = builtin_adapter();
+        let activation = adapter.activate_with_replay(Vec::new()).await;
+
+        assert!(activation.targets.is_empty());
+        assert!(activation.replay_workers.is_empty());
+    }
+
+    #[tokio::test]
+    async fn builtin_adapter_skips_non_store_target_when_init_fails() {
+        let adapter = builtin_adapter();
+        let target = TestTarget::new("primary", "webhook").with_failed_init();
+
+        let activation = adapter.activate_with_replay(vec![Box::new(target)]).await;
+
+        assert!(activation.targets.is_empty());
+        assert!(activation.replay_workers.is_empty());
+    }
+
+    #[tokio::test]
+    async fn builtin_adapter_keeps_store_backed_target_when_init_fails() {
+        let adapter = builtin_adapter();
+        let target = TestTarget::new("primary", "webhook").with_failed_init().with_store();
+
+        let activation = adapter.activate_with_replay(vec![Box::new(target)]).await;
+
+        assert_eq!(activation.targets.len(), 1);
+        assert_eq!(activation.replay_workers.len(), 1);
+    }
+
+    #[tokio::test]
+    async fn builtin_adapter_shutdown_clears_runtime_and_replay_workers() {
+        let adapter = builtin_adapter();
+        let target = TestTarget::new("primary", "webhook");
+        let close_calls = Arc::clone(&target.close_calls);
+        let mut runtime = crate::runtime::TargetRuntimeManager::new();
+        let mut replay_workers = crate::runtime::ReplayWorkerManager::new();
+
+        let activation = adapter.activate_with_replay(vec![Box::new(target)]).await;
+        adapter
+            .replace_runtime_targets(&mut runtime, &mut replay_workers, activation)
+            .await
+            .expect("replace_runtime_targets should succeed");
+
+        assert_eq!(runtime.len(), 1);
+        assert_eq!(replay_workers.len(), 0);
+
+        adapter
+            .shutdown(&mut runtime, &mut replay_workers)
+            .await
+            .expect("shutdown should succeed");
+
+        assert!(runtime.is_empty());
+        assert!(replay_workers.is_empty());
+        assert_eq!(close_calls.load(Ordering::SeqCst), 1);
+    }
+}

crates/targets/src/runtime/mod.rs

@@ -0,0 +1,641 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+pub mod adapter;
+pub mod sidecar;
+pub mod sidecar_protocol;
+
+use crate::Target;
+use crate::arn::TargetID;
+use crate::store::{Key, Store, ensure_store_entry_raw_readable};
+use crate::target::QueuedPayload;
+use crate::target::TargetDeliverySnapshot;
+use crate::{StoreError, TargetError};
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use std::sync::Arc;
+use std::{collections::HashMap, fmt::Debug};
+use std::{future::Future, pin::Pin, time::Duration};
+use tokio::sync::{Semaphore, mpsc};
+
+/// Shared target trait object used by the runtime manager.
+pub type SharedTarget<E> = Arc<dyn Target<E> + Send + Sync>;
+type ReplayHook<E> = Arc<dyn Fn(ReplayEvent<E>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync>;
+
+#[derive(Debug, Default)]
+pub struct ReplayWorkerManager {
+    cancellers: HashMap<String, mpsc::Sender<()>>,
+}
+
+impl ReplayWorkerManager {
+    pub fn new() -> Self {
+        Self {
+            cancellers: HashMap::new(),
+        }
+    }
+
+    pub fn insert(&mut self, target_id: String, cancel_tx: mpsc::Sender<()>) {
+        self.cancellers.insert(target_id, cancel_tx);
+    }
+
+    pub fn len(&self) -> usize {
+        self.cancellers.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.cancellers.is_empty()
+    }
+
+    pub fn snapshot(&self, target_count: usize) -> RuntimeStatusSnapshot {
+        RuntimeStatusSnapshot {
+            replay_worker_count: self.len(),
+            target_count,
+        }
+    }
+
+    pub async fn stop_all(&mut self, log_prefix: &str) {
+        for (target_id, cancel_tx) in self.cancellers.drain() {
+            tracing::info!(target_id = %target_id, "{log_prefix}");
+            let _ = cancel_tx.send(()).await;
+        }
+    }
+}
+
+pub struct RuntimeActivation<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    pub replay_workers: ReplayWorkerManager,
+    pub targets: Vec<SharedTarget<E>>,
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct RuntimeStatusSnapshot {
+    pub replay_worker_count: usize,
+    pub target_count: usize,
+}
+
+/// A read-only runtime snapshot for a target instance.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct RuntimeTargetSnapshot {
+    pub failed_messages: u64,
+    pub queue_length: u64,
+    pub target_id: String,
+    pub target_type: String,
+    pub total_messages: u64,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RuntimeTargetHealthState {
+    Disabled,
+    Error,
+    Offline,
+    Online,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct RuntimeTargetHealthSnapshot {
+    pub enabled: bool,
+    pub error_message: Option<String>,
+    pub state: RuntimeTargetHealthState,
+    pub target_id: String,
+    pub target_type: String,
+}
+
+pub enum ReplayEvent<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    Delivered {
+        key: Key,
+        target: SharedTarget<E>,
+    },
+    RetryableError {
+        error: TargetError,
+        key: Key,
+        retry_count: usize,
+        target: SharedTarget<E>,
+    },
+    Dropped {
+        key: Key,
+        reason: String,
+        target: SharedTarget<E>,
+    },
+    PermanentFailure {
+        error: TargetError,
+        key: Key,
+        target: SharedTarget<E>,
+    },
+    RetryExhausted {
+        key: Key,
+        target: SharedTarget<E>,
+    },
+    UnreadableEntry {
+        error: StoreError,
+        key: Key,
+        target: SharedTarget<E>,
+    },
+}
+
+/// Shared runtime container for managing instantiated targets.
+///
+/// This intentionally focuses on low-risk shared lifecycle primitives first:
+/// add/remove/close/list/snapshot. Replay workers and reload orchestration can
+/// be layered on top in later phases.
+pub struct TargetRuntimeManager<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    targets: HashMap<String, SharedTarget<E>>,
+}
+
+impl<E> Default for TargetRuntimeManager<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<E> Debug for TargetRuntimeManager<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("TargetRuntimeManager")
+            .field("target_count", &self.targets.len())
+            .finish()
+    }
+}
+
+impl<E> TargetRuntimeManager<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    pub fn new() -> Self {
+        Self { targets: HashMap::new() }
+    }
+
+    pub fn add_arc(&mut self, target: SharedTarget<E>) -> Option<SharedTarget<E>> {
+        let key = target.id().to_string();
+        self.targets.insert(key, target)
+    }
+
+    pub fn add_boxed(&mut self, target: Box<dyn Target<E> + Send + Sync>) -> Option<SharedTarget<E>> {
+        self.add_arc(Arc::from(target))
+    }
+
+    pub fn get(&self, key: &str) -> Option<SharedTarget<E>> {
+        self.targets.get(key).cloned()
+    }
+
+    pub fn get_by_target_id(&self, target_id: &TargetID) -> Option<SharedTarget<E>> {
+        self.get(&target_id.to_string())
+    }
+
+    pub fn remove(&mut self, key: &str) -> Option<SharedTarget<E>> {
+        self.targets.remove(key)
+    }
+
+    pub fn remove_by_target_id(&mut self, target_id: &TargetID) -> Option<SharedTarget<E>> {
+        self.remove(&target_id.to_string())
+    }
+
+    pub fn clear(&mut self) {
+        self.targets.clear();
+    }
+
+    pub async fn remove_and_close(&mut self, key: &str) -> Option<SharedTarget<E>> {
+        let target = self.targets.remove(key)?;
+        if let Err(err) = target.close().await {
+            tracing::error!(target_id = %key, error = %err, "Failed to close target during removal");
+        }
+        Some(target)
+    }
+
+    pub async fn remove_by_target_id_and_close(&mut self, target_id: &TargetID) -> Option<SharedTarget<E>> {
+        self.remove_and_close(&target_id.to_string()).await
+    }
+
+    pub async fn clear_and_close(&mut self) {
+        let target_ids: Vec<String> = self.targets.keys().cloned().collect();
+        for target_id in target_ids {
+            let _ = self.remove_and_close(&target_id).await;
+        }
+        self.targets.clear();
+    }
+
+    pub fn target_ids(&self) -> Vec<TargetID> {
+        self.targets.values().map(|target| target.id()).collect()
+    }
+
+    pub fn keys(&self) -> Vec<String> {
+        self.targets.keys().cloned().collect()
+    }
+
+    pub fn values(&self) -> Vec<SharedTarget<E>> {
+        self.targets.values().cloned().collect()
+    }
+
+    pub fn len(&self) -> usize {
+        self.targets.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.targets.is_empty()
+    }
+
+    pub fn snapshots(&self) -> Vec<RuntimeTargetSnapshot> {
+        let mut snapshots = Vec::with_capacity(self.targets.len());
+        for target in self.targets.values() {
+            let delivery = target.delivery_snapshot();
+            let target_id = target.id();
+            snapshots.push(snapshot_from_delivery(target_id, delivery));
+        }
+        snapshots.sort_by(|a, b| a.target_id.cmp(&b.target_id));
+        snapshots
+    }
+
+    pub fn status_snapshot(&self, replay_workers: &ReplayWorkerManager) -> RuntimeStatusSnapshot {
+        replay_workers.snapshot(self.len())
+    }
+
+    pub async fn health_snapshots(&self) -> Vec<RuntimeTargetHealthSnapshot> {
+        let mut snapshots = Vec::with_capacity(self.targets.len());
+        for target in self.targets.values() {
+            let enabled = target.is_enabled();
+            let target_id = target.id();
+            let (state, error_message) = if !enabled {
+                (RuntimeTargetHealthState::Disabled, None)
+            } else {
+                match target.is_active().await {
+                    Ok(true) => (RuntimeTargetHealthState::Online, None),
+                    Ok(false) => (RuntimeTargetHealthState::Offline, None),
+                    Err(err) => (RuntimeTargetHealthState::Error, Some(err.to_string())),
+                }
+            };
+
+            snapshots.push(RuntimeTargetHealthSnapshot {
+                enabled,
+                error_message,
+                state,
+                target_id: target_id.to_string(),
+                target_type: target_id.name,
+            });
+        }
+        snapshots.sort_by(|a, b| a.target_id.cmp(&b.target_id));
+        snapshots
+    }
+}
+
+fn snapshot_from_delivery(target_id: TargetID, delivery: TargetDeliverySnapshot) -> RuntimeTargetSnapshot {
+    RuntimeTargetSnapshot {
+        failed_messages: delivery.failed_messages,
+        queue_length: delivery.queue_length,
+        target_id: target_id.to_string(),
+        target_type: target_id.name,
+        total_messages: delivery.total_messages,
+    }
+}
+
+pub async fn init_target_and_optionally_start_replay<E, F, G>(
+    target: Box<dyn Target<E> + Send + Sync>,
+    on_replay_start: F,
+    start_replay: G,
+) -> Option<(SharedTarget<E>, Option<mpsc::Sender<()>>)>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+    F: FnOnce(&str, bool),
+    G: FnOnce(Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send>, SharedTarget<E>) -> mpsc::Sender<()>,
+{
+    let target_id = target.id().to_string();
+    let has_store = target.store().is_some();
+
+    if let Err(err) = target.init().await {
+        tracing::error!(target_id = %target_id, error = %err, "Failed to initialize target");
+        if !has_store {
+            return None;
+        }
+        tracing::warn!(
+            target_id = %target_id,
+            "Proceeding with store-backed target despite init failure"
+        );
+    }
+
+    let shared: SharedTarget<E> = Arc::from(target);
+    if !shared.is_enabled() {
+        on_replay_start(&target_id, false);
+        return Some((shared, None));
+    }
+
+    let cancel = shared
+        .store()
+        .map(|store| start_replay(store.boxed_clone(), Arc::clone(&shared)));
+    on_replay_start(&target_id, cancel.is_some());
+    Some((shared, cancel))
+}
+
+pub async fn activate_targets_with_replay<E, F, Fut>(
+    targets: Vec<Box<dyn Target<E> + Send + Sync>>,
+    mut activate_one: F,
+) -> RuntimeActivation<E>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+    F: FnMut(Box<dyn Target<E> + Send + Sync>) -> Fut,
+    Fut: Future<Output = Option<(SharedTarget<E>, Option<mpsc::Sender<()>>)>>,
+{
+    let mut replay_workers = ReplayWorkerManager::new();
+    let mut shared_targets = Vec::new();
+
+    for target in targets {
+        if let Some((shared_target, cancel_tx)) = activate_one(target).await {
+            let target_id = shared_target.id().to_string();
+            if let Some(cancel_tx) = cancel_tx {
+                replay_workers.insert(target_id, cancel_tx);
+            }
+            shared_targets.push(shared_target);
+        }
+    }
+
+    RuntimeActivation {
+        replay_workers,
+        targets: shared_targets,
+    }
+}
+
+pub fn start_replay_worker<E>(
+    mut store: Box<dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send>,
+    target: SharedTarget<E>,
+    hook: ReplayHook<E>,
+    semaphore: Option<Arc<Semaphore>>,
+    batch_timeout: Duration,
+    idle_sleep: Duration,
+) -> mpsc::Sender<()>
+where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    let (cancel_tx, cancel_rx) = mpsc::channel(1);
+
+    tokio::spawn(async move {
+        stream_replay_worker(&mut *store, target, cancel_rx, hook, semaphore, batch_timeout, idle_sleep).await;
+    });
+
+    cancel_tx
+}
+
+async fn stream_replay_worker<E>(
+    store: &mut (dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send),
+    target: SharedTarget<E>,
+    mut cancel_rx: mpsc::Receiver<()>,
+    hook: ReplayHook<E>,
+    semaphore: Option<Arc<Semaphore>>,
+    batch_timeout: Duration,
+    idle_sleep: Duration,
+) where
+    E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+{
+    const MAX_RETRIES: usize = 5;
+    const BASE_RETRY_DELAY: Duration = Duration::from_secs(2);
+
+    let mut batch_keys = Vec::with_capacity(1);
+    let mut last_flush = tokio::time::Instant::now();
+
+    loop {
+        if cancel_rx.try_recv().is_ok() {
+            return;
+        }
+
+        let keys = store.list();
+        if keys.is_empty() {
+            if !batch_keys.is_empty() && last_flush.elapsed() >= batch_timeout {
+                process_replay_batch(&mut batch_keys, target.clone(), &hook, semaphore.clone()).await;
+                last_flush = tokio::time::Instant::now();
+            }
+            tokio::time::sleep(idle_sleep).await;
+            continue;
+        }
+
+        for key in keys {
+            if cancel_rx.try_recv().is_ok() {
+                if !batch_keys.is_empty() {
+                    process_replay_batch(&mut batch_keys, target.clone(), &hook, semaphore.clone()).await;
+                }
+                return;
+            }
+
+            match ensure_store_entry_raw_readable(&*store, &key) {
+                Ok(true) => {}
+                Ok(false) => continue,
+                Err(err) => {
+                    hook(ReplayEvent::UnreadableEntry {
+                        error: err,
+                        key,
+                        target: target.clone(),
+                    })
+                    .await;
+                    continue;
+                }
+            }
+
+            batch_keys.push(key);
+            if !batch_keys.is_empty() || last_flush.elapsed() >= batch_timeout {
+                process_replay_batch(&mut batch_keys, target.clone(), &hook, semaphore.clone()).await;
+                last_flush = tokio::time::Instant::now();
+            }
+        }
+
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+
+    async fn process_replay_batch<E>(
+        batch_keys: &mut Vec<Key>,
+        target: SharedTarget<E>,
+        hook: &ReplayHook<E>,
+        semaphore: Option<Arc<Semaphore>>,
+    ) where
+        E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+    {
+        if batch_keys.is_empty() {
+            return;
+        }
+
+        let _permit = match semaphore {
+            Some(ref semaphore) => match semaphore.clone().acquire_owned().await {
+                Ok(permit) => Some(permit),
+                Err(err) => {
+                    tracing::error!(error = %err, "Failed to acquire replay semaphore permit");
+                    return;
+                }
+            },
+            None => None,
+        };
+
+        for key in batch_keys.iter() {
+            let mut retry_count = 0usize;
+            let mut success = false;
+
+            while retry_count < MAX_RETRIES && !success {
+                match target.send_from_store(key.clone()).await {
+                    Ok(_) => {
+                        hook(ReplayEvent::Delivered {
+                            key: key.clone(),
+                            target: target.clone(),
+                        })
+                        .await;
+                        success = true;
+                    }
+                    Err(err) => match err {
+                        TargetError::NotConnected | TargetError::Timeout(_) => {
+                            retry_count += 1;
+                            hook(ReplayEvent::RetryableError {
+                                error: err,
+                                key: key.clone(),
+                                retry_count,
+                                target: target.clone(),
+                            })
+                            .await;
+
+                            let jitter = Duration::from_millis(key.to_string().len() as u64 % 500);
+                            let backoff = 1u32 << retry_count as u32;
+                            tokio::time::sleep(BASE_RETRY_DELAY * backoff + jitter).await;
+                        }
+                        TargetError::Dropped(reason) => {
+                            hook(ReplayEvent::Dropped {
+                                key: key.clone(),
+                                reason,
+                                target: target.clone(),
+                            })
+                            .await;
+                            break;
+                        }
+                        other => {
+                            hook(ReplayEvent::PermanentFailure {
+                                error: other,
+                                key: key.clone(),
+                                target: target.clone(),
+                            })
+                            .await;
+                            break;
+                        }
+                    },
+                }
+            }
+
+            if retry_count >= MAX_RETRIES && !success {
+                hook(ReplayEvent::RetryExhausted {
+                    key: key.clone(),
+                    target: target.clone(),
+                })
+                .await;
+            }
+        }
+
+        batch_keys.clear();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::TargetRuntimeManager;
+    use crate::StoreError;
+    use crate::arn::TargetID;
+    use crate::store::{Key, Store};
+    use crate::target::{EntityTarget, QueuedPayload, QueuedPayloadMeta};
+    use crate::{Target, TargetError};
+    use async_trait::async_trait;
+    use serde::{Serialize, de::DeserializeOwned};
+    use std::sync::Arc;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    #[derive(Clone)]
+    struct TestTarget {
+        id: TargetID,
+        close_calls: Arc<AtomicUsize>,
+    }
+
+    impl TestTarget {
+        fn new(id: &str, name: &str) -> Self {
+            Self {
+                id: TargetID::new(id.to_string(), name.to_string()),
+                close_calls: Arc::new(AtomicUsize::new(0)),
+            }
+        }
+    }
+
+    #[async_trait]
+    impl<E> Target<E> for TestTarget
+    where
+        E: Send + Sync + 'static + Clone + Serialize + DeserializeOwned,
+    {
+        fn id(&self) -> TargetID {
+            self.id.clone()
+        }
+
+        async fn is_active(&self) -> Result<bool, TargetError> {
+            Ok(true)
+        }
+
+        async fn save(&self, _event: Arc<EntityTarget<E>>) -> Result<(), TargetError> {
+            Ok(())
+        }
+
+        async fn send_raw_from_store(&self, _key: Key, _body: Vec<u8>, _meta: QueuedPayloadMeta) -> Result<(), TargetError> {
+            Ok(())
+        }
+
+        async fn close(&self) -> Result<(), TargetError> {
+            self.close_calls.fetch_add(1, Ordering::SeqCst);
+            Ok(())
+        }
+
+        fn store(&self) -> Option<&(dyn Store<QueuedPayload, Error = StoreError, Key = Key> + Send + Sync)> {
+            None
+        }
+
+        fn clone_dyn(&self) -> Box<dyn Target<E> + Send + Sync> {
+            Box::new(self.clone())
+        }
+
+        fn is_enabled(&self) -> bool {
+            true
+        }
+    }
+
+    #[tokio::test]
+    async fn runtime_manager_removes_and_closes_target() {
+        let mut manager = TargetRuntimeManager::<String>::new();
+        let target = TestTarget::new("primary", "webhook");
+        let close_calls = Arc::clone(&target.close_calls);
+
+        manager.add_boxed(Box::new(target));
+        assert_eq!(manager.len(), 1);
+
+        let removed = manager.remove_and_close("primary:webhook").await;
+        assert!(removed.is_some());
+        assert_eq!(manager.len(), 0);
+        assert_eq!(close_calls.load(Ordering::SeqCst), 1);
+    }
+
+    #[test]
+    fn runtime_manager_snapshots_targets() {
+        let mut manager = TargetRuntimeManager::<String>::new();
+        manager.add_boxed(Box::new(TestTarget::new("primary", "webhook")));
+
+        let snapshots = manager.snapshots();
+        assert_eq!(snapshots.len(), 1);
+        assert_eq!(snapshots[0].target_id, "primary:webhook");
+        assert_eq!(snapshots[0].target_type, "webhook");
+    }
+}

crates/targets/src/runtime/sidecar.rs

@@ -0,0 +1,171 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::TargetDomain;
+use crate::runtime::sidecar_protocol::SidecarHandshake;
+use serde::{Deserialize, Serialize};
+use std::time::Duration;
+
+const DEFAULT_FAILURE_THRESHOLD: usize = 3;
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub struct SidecarPluginRuntime {
+    pub endpoint: String,
+    pub handshake: SidecarHandshake,
+    pub healthy: bool,
+    pub failure_count: usize,
+    pub degraded_to_builtin: bool,
+    pub last_error: Option<String>,
+}
+
+impl SidecarPluginRuntime {
+    pub fn new(endpoint: impl Into<String>, handshake: SidecarHandshake) -> Self {
+        Self {
+            endpoint: endpoint.into(),
+            handshake,
+            healthy: false,
+            failure_count: 0,
+            degraded_to_builtin: false,
+            last_error: None,
+        }
+    }
+
+    pub fn enable(&mut self, expected_plugin_id: &str, required_domain: TargetDomain) -> Result<(), String> {
+        self.handshake.validate(expected_plugin_id)?;
+        if !self.handshake.supported_domains.contains(&required_domain) {
+            return Err(format!(
+                "sidecar plugin {} does not support required domain {:?}",
+                self.handshake.plugin_id, required_domain
+            ));
+        }
+
+        self.healthy = true;
+        self.degraded_to_builtin = false;
+        self.last_error = None;
+        self.failure_count = 0;
+        Ok(())
+    }
+
+    pub fn mark_unhealthy(&mut self) {
+        self.healthy = false;
+    }
+
+    pub fn record_failure(&mut self, error: impl Into<String>) {
+        self.failure_count = self.failure_count.saturating_add(1);
+        self.healthy = false;
+        self.last_error = Some(error.into());
+        if self.failure_count >= DEFAULT_FAILURE_THRESHOLD {
+            self.degraded_to_builtin = true;
+        }
+    }
+
+    pub fn send_with_timeout(&mut self, operation_timeout: Duration, simulated_latency: Duration) -> Result<(), String> {
+        if simulated_latency > operation_timeout {
+            self.record_failure(format!(
+                "sidecar send timeout after {:?} (budget {:?})",
+                simulated_latency, operation_timeout
+            ));
+            return Err(self
+                .last_error
+                .clone()
+                .unwrap_or_else(|| "sidecar timeout without recorded error".to_string()));
+        }
+        self.healthy = true;
+        self.last_error = None;
+        Ok(())
+    }
+
+    pub fn shutdown(&mut self) {
+        self.healthy = false;
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::SidecarPluginRuntime;
+    use crate::TargetDomain;
+    use crate::runtime::sidecar_protocol::{SIDECAR_RUNTIME_PROTOCOL_VERSION, SidecarHandshake, SidecarPluginCapability};
+    use std::time::Duration;
+
+    fn notify_sidecar_handshake() -> SidecarHandshake {
+        SidecarHandshake {
+            protocol_version: SIDECAR_RUNTIME_PROTOCOL_VERSION.to_string(),
+            plugin_id: "external:webhook".to_string(),
+            plugin_version: "1.2.3".to_string(),
+            supported_domains: vec![TargetDomain::Notify],
+            capabilities: vec![
+                SidecarPluginCapability::HealthCheck,
+                SidecarPluginCapability::SendEvent,
+                SidecarPluginCapability::Shutdown,
+            ],
+        }
+    }
+
+    #[test]
+    fn sidecar_runtime_enable_marks_runtime_healthy() {
+        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
+
+        runtime
+            .enable("external:webhook", TargetDomain::Notify)
+            .expect("sidecar runtime should enable");
+
+        assert!(runtime.healthy);
+    }
+
+    #[test]
+    fn sidecar_runtime_enable_rejects_domain_mismatch() {
+        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
+
+        let result = runtime.enable("external:webhook", TargetDomain::Audit);
+
+        assert!(result.is_err());
+        assert!(!runtime.healthy);
+    }
+
+    #[test]
+    fn sidecar_runtime_shutdown_marks_runtime_unhealthy() {
+        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
+        runtime
+            .enable("external:webhook", TargetDomain::Notify)
+            .expect("sidecar runtime should enable");
+
+        runtime.shutdown();
+
+        assert!(!runtime.healthy);
+    }
+
+    #[test]
+    fn sidecar_runtime_degrades_to_builtin_after_failure_threshold() {
+        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
+
+        runtime.record_failure("send failed");
+        runtime.record_failure("send failed again");
+        runtime.record_failure("send failed third time");
+
+        assert!(runtime.degraded_to_builtin);
+        assert!(!runtime.healthy);
+        assert_eq!(runtime.failure_count, 3);
+    }
+
+    #[test]
+    fn sidecar_runtime_send_timeout_records_last_error() {
+        let mut runtime = SidecarPluginRuntime::new("grpc://127.0.0.1:50051", notify_sidecar_handshake());
+
+        let result = runtime.send_with_timeout(Duration::from_millis(50), Duration::from_millis(75));
+
+        assert!(result.is_err());
+        assert_eq!(runtime.last_error.as_deref(), Some("sidecar send timeout after 75ms (budget 50ms)"));
+    }
+}

crates/targets/src/runtime/sidecar_protocol.rs

@@ -0,0 +1,106 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::TargetDomain;
+use serde::{Deserialize, Serialize};
+
+pub const SIDECAR_RUNTIME_PROTOCOL_VERSION: &str = "rustfs.target-runtime.v1";
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum SidecarPluginCapability {
+    HealthCheck,
+    SendEvent,
+    Shutdown,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub struct SidecarHandshake {
+    pub protocol_version: String,
+    pub plugin_id: String,
+    pub plugin_version: String,
+    pub supported_domains: Vec<TargetDomain>,
+    pub capabilities: Vec<SidecarPluginCapability>,
+}
+
+impl SidecarHandshake {
+    pub fn validate(&self, expected_plugin_id: &str) -> Result<(), String> {
+        if self.protocol_version != SIDECAR_RUNTIME_PROTOCOL_VERSION {
+            return Err(format!(
+                "unsupported sidecar protocol version: expected {}, got {}",
+                SIDECAR_RUNTIME_PROTOCOL_VERSION, self.protocol_version
+            ));
+        }
+
+        if self.plugin_id != expected_plugin_id {
+            return Err(format!(
+                "sidecar plugin id mismatch: expected {}, got {}",
+                expected_plugin_id, self.plugin_id
+            ));
+        }
+
+        for capability in [
+            SidecarPluginCapability::HealthCheck,
+            SidecarPluginCapability::SendEvent,
+            SidecarPluginCapability::Shutdown,
+        ] {
+            if !self.capabilities.contains(&capability) {
+                return Err(format!("sidecar handshake missing required capability: {:?}", capability));
+            }
+        }
+
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{SIDECAR_RUNTIME_PROTOCOL_VERSION, SidecarHandshake, SidecarPluginCapability};
+    use crate::TargetDomain;
+
+    #[test]
+    fn sidecar_handshake_accepts_expected_contract() {
+        let handshake = SidecarHandshake {
+            protocol_version: SIDECAR_RUNTIME_PROTOCOL_VERSION.to_string(),
+            plugin_id: "external:webhook".to_string(),
+            plugin_version: "1.2.3".to_string(),
+            supported_domains: vec![TargetDomain::Notify],
+            capabilities: vec![
+                SidecarPluginCapability::HealthCheck,
+                SidecarPluginCapability::SendEvent,
+                SidecarPluginCapability::Shutdown,
+            ],
+        };
+
+        assert!(handshake.validate("external:webhook").is_ok());
+    }
+
+    #[test]
+    fn sidecar_handshake_rejects_protocol_mismatch() {
+        let handshake = SidecarHandshake {
+            protocol_version: "rustfs.target-runtime.v0".to_string(),
+            plugin_id: "external:webhook".to_string(),
+            plugin_version: "1.2.3".to_string(),
+            supported_domains: vec![TargetDomain::Notify],
+            capabilities: vec![
+                SidecarPluginCapability::HealthCheck,
+                SidecarPluginCapability::SendEvent,
+                SidecarPluginCapability::Shutdown,
+            ],
+        };
+
+        assert!(handshake.validate("external:webhook").is_err());
+    }
+}