diff --git a/crates/bifrost/src/appender.rs b/crates/bifrost/src/appender.rs index 1182046a1a..cec3d5fe8b 100644 --- a/crates/bifrost/src/appender.rs +++ b/crates/bifrost/src/appender.rs @@ -16,7 +16,7 @@ use std::time::{Duration, Instant}; use bytes::BytesMut; use tracing::{debug, info, instrument}; -use restate_core::{Metadata, MetadataKind, TaskCenter}; +use restate_core::{Metadata, TaskCenter}; use restate_futures_util::overdue::OverdueLoggingExt; use restate_types::Versioned; use restate_types::config::Configuration; @@ -448,12 +448,17 @@ impl Appender { } } + let mut wait_for_writable_segment = bifrost_inner.log_chain_watcher().subscribe( + log_id, + crate::log_chain_watcher::ChainCondition::WritableSegmentAfter(sealed_segment), + ); + tokio::select! { biased; // if error it means that metadata manager has stopped. We are shutting down. // the check for shutdown in the loop above will catch if this happened and bubble // up the shutdown error. - _ = metadata.wait_for_version(MetadataKind::Logs, log_metadata_version.next()) => { + _ = wait_for_writable_segment.wait() => { // do not advance the sleep duration. Successive metadata updates that are // irrelavant to this loglet should not increase the sleep duration. } diff --git a/crates/bifrost/src/bifrost.rs b/crates/bifrost/src/bifrost.rs index 30fa22e69a..d616cdc343 100644 --- a/crates/bifrost/src/bifrost.rs +++ b/crates/bifrost/src/bifrost.rs @@ -34,6 +34,7 @@ use restate_types::storage::StorageEncode; use crate::appender::Appender; use crate::background_appender::BackgroundAppender; +use crate::log_chain_watcher::LogChainWatcherHandle; use crate::log_chain_writer::LogChainCommand; use crate::loglet::{FindTailOptions, LogletProvider, OperationError}; use crate::loglet_wrapper::LogletWrapper; @@ -294,15 +295,17 @@ pub struct BifrostInner { pub(crate) providers: OnceLock>>>, shutting_down: AtomicBool, pub(crate) read_stream_registry: crate::read_stream_registry::ActiveReadStreamRegistry, + log_chain_watcher: LogChainWatcherHandle, } impl BifrostInner { - pub fn new(watchdog: WatchdogSender) -> Self { + pub fn new(watchdog: WatchdogSender, log_chain_watcher: LogChainWatcherHandle) -> Self { Self { watchdog, providers: Default::default(), shutting_down: AtomicBool::new(false), read_stream_registry: Default::default(), + log_chain_watcher, } } @@ -660,6 +663,10 @@ impl BifrostInner { loglet, )) } + + pub fn log_chain_watcher(&self) -> &LogChainWatcherHandle { + &self.log_chain_watcher + } } /// Result of a lookup by lsn in the chain diff --git a/crates/bifrost/src/lib.rs b/crates/bifrost/src/lib.rs index 45ab9d1734..da70528856 100644 --- a/crates/bifrost/src/lib.rs +++ b/crates/bifrost/src/lib.rs @@ -13,6 +13,7 @@ mod background_appender; mod bifrost; mod bifrost_admin; mod error; +mod log_chain_watcher; mod log_chain_writer; pub mod loglet; mod loglet_wrapper; diff --git a/crates/bifrost/src/log_chain_watcher.rs b/crates/bifrost/src/log_chain_watcher.rs new file mode 100644 index 0000000000..782b550f7a --- /dev/null +++ b/crates/bifrost/src/log_chain_watcher.rs @@ -0,0 +1,679 @@ +// Copyright (c) 2023 - 2026 Restate Software, Inc., Restate GmbH. +// All rights reserved. +// +// Use of this software is governed by the Business Source License +// included in the LICENSE file. +// +// As of the Change Date specified in that file, in accordance with +// the Business Source License, use of this software will be governed +// by the Apache License, Version 2.0. + +use std::collections::{HashMap, VecDeque}; + +use tokio::sync::{mpsc, oneshot}; +use tracing::trace; + +use restate_core::{Metadata, MetadataKind, TaskCenter, TaskKind, cancellation_watcher}; +use restate_types::logs::{ + LogId, + metadata::{Chain, Logs, SegmentIndex}, +}; + +// TODO: Remove when all the subscriptions are in place +#[allow(dead_code)] +#[derive(Debug)] +pub enum ChainCondition { + /// Fires whenever the segment with the given index is considered sealed. + Sealed(SegmentIndex), + /// Fires whenever there's a writable segment after the given index. + WritableSegmentAfter(SegmentIndex), +} + +#[derive(Debug)] +pub(crate) struct Subscription { + log_id: LogId, + condition: ChainCondition, + sender: oneshot::Sender<()>, + #[cfg(test)] + /// A test only notification for when the subscription request is processed by the watcher. + armed: Option>, +} + +impl Subscription { + /// Fires the test-only notification signaling that the watcher has processed + /// this subscription request. No-op outside of tests. + fn notify_armed(&mut self) { + #[cfg(test)] + if let Some(tx) = self.armed.take() { + let _ = tx.send(()); + } + } + + fn notify(self) { + let _ = self.sender.send(()); + } +} + +enum SubscriptionHandleInner { + Waiting(oneshot::Receiver<()>), + Completed, +} + +pub struct SubscriptionHandle(SubscriptionHandleInner); +impl SubscriptionHandle { + fn new(receiver: oneshot::Receiver<()>) -> Self { + Self(SubscriptionHandleInner::Waiting(receiver)) + } + + /// Waits for the subscription condition to be satisfied. + /// + /// This is cancel-safe: if the returned future is dropped before it resolves, + /// the subscription is left untouched and a subsequent `wait()` will resume + /// waiting. Once it has resolved, every later call returns immediately. + pub async fn wait(&mut self) -> Result<(), oneshot::error::RecvError> { + match &mut self.0 { + SubscriptionHandleInner::Waiting(receiver) => { + let result = std::pin::Pin::new(receiver).await; + self.0 = SubscriptionHandleInner::Completed; + result + } + SubscriptionHandleInner::Completed => Ok(()), + } + } + + #[cfg(test)] + pub fn try_recv(&mut self) -> Result<(), oneshot::error::TryRecvError> { + match &mut self.0 { + SubscriptionHandleInner::Waiting(receiver) => receiver.try_recv(), + SubscriptionHandleInner::Completed => Ok(()), + } + } +} + +/// The main and only public interface for the log chain watcher. Allows callers to +/// subscribe to chain events for a particular log. +pub struct LogChainWatcherHandle { + sender: mpsc::UnboundedSender, +} + +impl LogChainWatcherHandle { + pub fn new(sender: mpsc::UnboundedSender) -> Self { + Self { sender } + } + + /// Subscribe to chain events for a particular log. + /// + /// Note: If the condition is already satisfied by the time it reaches the watcher, the subscription + /// will fire immediately without observing a new logs config update. + #[must_use] + pub fn subscribe(&self, log_id: LogId, condition: ChainCondition) -> SubscriptionHandle { + let (sender, receiver) = oneshot::channel(); + let _ = self.sender.send(Subscription { + log_id, + condition, + sender, + #[cfg(test)] + armed: None, + }); + SubscriptionHandle::new(receiver) + } + + /// Same as `subscribe` but waits until the subscription is accepted by the watcher. + /// This is a test only util. + #[cfg(test)] + #[must_use] + pub async fn subscribe_and_wait_for_armed( + &self, + log_id: LogId, + condition: ChainCondition, + ) -> SubscriptionHandle { + let (armed_tx, armed_rx) = oneshot::channel(); + let (sender, receiver) = oneshot::channel(); + let _ = self.sender.send(Subscription { + log_id, + condition, + sender, + armed: Some(armed_tx), + }); + armed_rx.await.unwrap(); + SubscriptionHandle::new(receiver) + } +} + +pub(crate) type LogChainWatcherReceiver = mpsc::UnboundedReceiver; + +/// LogChainWatcher is a background task that allows users to subscribe only to particular +/// events that they are interested in on the log chain. This amortizes the cost of subscribing +/// to chain events across all the watchers, and reduces the number of irrelevant wakeups on watchers +/// had they subscribed to the entire logs config instead. +pub(crate) struct LogChainWatcher { + inbound: LogChainWatcherReceiver, + subscriptions: HashMap>, +} + +impl LogChainWatcher { + pub fn start(inbound: LogChainWatcherReceiver) -> anyhow::Result<()> { + let watcher = Self { + inbound, + subscriptions: HashMap::default(), + }; + + TaskCenter::spawn( + TaskKind::BifrostLogChainWatcher, + "bifrost-logchainwatcher", + watcher.run(), + )?; + Ok(()) + } + + async fn run(mut self) -> anyhow::Result<()> { + let shutdown = cancellation_watcher(); + tokio::pin!(shutdown); + trace!("Bifrost log chain watcher started"); + + let metadata = Metadata::current(); + let mut logs = metadata.updateable_logs_metadata(); + + let mut logs_watcher = metadata.watch(MetadataKind::Logs); + + loop { + tokio::select! { + _ = &mut shutdown => { + return Ok(()); + } + _ = logs_watcher.changed() => { + let logs = logs.live_load(); + self.on_config_change(logs); + } + Some(mut sub) = self.inbound.recv() => { + // use it as an opportunity to reap potentially dropped subs + self.maybe_reap_dropped_subs(); + + let logs = logs.live_load(); + let chain = logs.chain(&sub.log_id); + match chain { + // If the subscription condition is satisfied already, notify immediately. + Some(chain) if Self::is_condition_satisfied(&sub, chain) => { + sub.notify_armed(); + sub.notify(); + } + // Either the chain is not yet found, or the condition is not satisfied. + _ => { + self.register_subscription(sub); + } + } + } + } + } + } + + fn register_subscription(&mut self, mut sub: Subscription) { + sub.notify_armed(); + self.subscriptions + .entry(sub.log_id) + .or_default() + .push_back(sub); + } + + fn on_config_change(&mut self, logs: &Logs) { + for (log_id, subs) in self.subscriptions.iter_mut() { + let Some(chain) = logs.chain(log_id) else { + // reap dropped subs + Self::reap_dropped_subs_inner(subs); + continue; + }; + + let mut idx = 0; + while idx < subs.len() { + let sub = &subs[idx]; + if sub.sender.is_closed() { + subs.swap_remove_back(idx); + continue; + } + if Self::is_condition_satisfied(sub, chain) { + let sub = subs.swap_remove_back(idx).unwrap(); + sub.notify(); + } else { + idx += 1; + } + } + } + } + + fn is_condition_satisfied(sub: &Subscription, chain: &Chain) -> bool { + match sub.condition { + ChainCondition::Sealed(sealed_segment) => { + let tail_segment = chain.tail(); + // If the tail segment index is greater than the subscribed to segment index, + // then the subscribed to segment is for sure sealed. + if tail_segment.index() > sealed_segment { + true + } else if tail_segment.index() == sealed_segment { + tail_segment.config.kind.is_seal_marker() + } else { + false + } + } + ChainCondition::WritableSegmentAfter(idx) => { + let tail_segment = chain.non_special_tail(); + match tail_segment { + Some(tail_segment) => { + tail_segment.index() > idx && tail_segment.tail_lsn.is_none() + } + None => false, + } + } + } + } + + fn maybe_reap_dropped_subs(&mut self) { + // probabilistically drop subs that are no longer needed. + // The main full cleanup happens in on_config_change(). So this + // is just a fallback in case we have a stable logs config for a while. + // Cleanup will happen with 5% chance. + if rand::random_bool(0.95) { + return; + } + self.subscriptions.retain(|_, subs| { + Self::reap_dropped_subs_inner(subs); + !subs.is_empty() + }); + } + + fn reap_dropped_subs_inner(subs: &mut VecDeque) { + let mut idx = 0; + while idx < subs.len() { + if subs[idx].sender.is_closed() { + subs.swap_remove_back(idx); + } else { + idx += 1; + } + } + } +} + +#[cfg(test)] +mod tests { + use std::sync::Arc; + use std::time::Duration; + + use restate_core::{MetadataWriter, TestCoreEnv}; + use restate_types::logs::Lsn; + use restate_types::logs::builder::LogsBuilder; + use restate_types::logs::metadata::{LogletParams, ProviderKind, SealMetadata}; + + use super::*; + + async fn update_logs_metadata( + metadata_writer: &mut MetadataWriter, + f: impl FnOnce(&mut LogsBuilder), + ) -> anyhow::Result<()> { + let mut builder = Metadata::current() + .updateable_logs_metadata() + .live_load() + .clone() + .try_into_builder()?; + f(&mut builder); + metadata_writer.update(Arc::new(builder.build())).await?; + Ok(()) + } + + async fn init() -> anyhow::Result<(LogChainWatcherHandle, MetadataWriter)> { + let env = TestCoreEnv::create_with_single_node(1, 1).await; + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + LogChainWatcher::start(rx)?; + + Ok((LogChainWatcherHandle::new(tx), env.metadata_writer)) + } + + #[restate_core::test] + async fn seal_subscription() -> anyhow::Result<()> { + let (handle, mut metadata_writer) = init().await?; + + // Initially, two segments, and segment 1 is opened. + update_logs_metadata(&mut metadata_writer, |builder| { + let idx = builder + .chain(LogId::new(1)) + .unwrap() + .append_segment( + Lsn::from(1), + ProviderKind::InMemory, + LogletParams::from("test2"), + ) + .unwrap(); + assert_eq!(idx, SegmentIndex::from(1)); + }) + .await?; + + // Subscribing to seal event should fire when the chain gets sealed. + { + let mut sub = handle + .subscribe_and_wait_for_armed( + LogId::new(1), + ChainCondition::Sealed(SegmentIndex::from(1)), + ) + .await; + assert_eq!(sub.try_recv(), Err(oneshot::error::TryRecvError::Empty)); + + update_logs_metadata(&mut metadata_writer, |builder| { + builder + .chain(LogId::new(1)) + .unwrap() + .seal(Lsn::from(15), &SealMetadata::default()) + .unwrap(); + }) + .await?; + + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), sub.wait()).await, + Ok(Ok(())) + ); + } + + // Subscribing to a seal event on an already sealed segment should fire immediately. + { + let mut sub = + handle.subscribe(LogId::new(1), ChainCondition::Sealed(SegmentIndex::from(1))); + + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), sub.wait()).await, + Ok(Ok(())) + ); + } + + // A new writable segment should still consider segment 1 sealed. + { + update_logs_metadata(&mut metadata_writer, |builder| { + let idx = builder + .chain(LogId::new(1)) + .unwrap() + .append_segment( + Lsn::from(15), + ProviderKind::InMemory, + LogletParams::from("test2"), + ) + .unwrap(); + assert_eq!(idx, SegmentIndex::from(2)); + }) + .await?; + + let mut sub = + handle.subscribe(LogId::new(1), ChainCondition::Sealed(SegmentIndex::from(1))); + + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), sub.wait()).await, + Ok(Ok(())) + ); + } + + Ok(()) + } + + #[restate_core::test] + async fn writable_segment_after_subscription() -> anyhow::Result<()> { + let (handle, mut metadata_writer) = init().await?; + + // Create a chain with a single sealed segment. + update_logs_metadata(&mut metadata_writer, |builder| { + let idx = builder + .chain(LogId::new(1)) + .unwrap() + .append_segment( + Lsn::from(1), + ProviderKind::InMemory, + LogletParams::from("test2"), + ) + .unwrap(); + assert_eq!(idx, SegmentIndex::from(1)); + builder + .chain(LogId::new(1)) + .unwrap() + .seal(Lsn::from(10), &SealMetadata::default()) + .unwrap(); + }) + .await?; + + // Subscribing to a chain writable event should fire when the chain is writable. + { + let mut sub = handle + .subscribe_and_wait_for_armed( + LogId::new(1), + ChainCondition::WritableSegmentAfter(SegmentIndex::from(1)), + ) + .await; + assert_eq!(sub.try_recv(), Err(oneshot::error::TryRecvError::Empty)); + + update_logs_metadata(&mut metadata_writer, |builder| { + let idx = builder + .chain(LogId::new(1)) + .unwrap() + .append_segment( + Lsn::from(10), + ProviderKind::InMemory, + LogletParams::from("test2"), + ) + .unwrap(); + assert_eq!(idx, SegmentIndex::from(2)); + }) + .await?; + + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), sub.wait()).await, + Ok(Ok(())) + ); + } + + // Subscribing to a chain writable event on an already writable segment should fire immediately. + { + let mut sub = handle.subscribe( + LogId::new(1), + ChainCondition::WritableSegmentAfter(SegmentIndex::from(1)), + ); + + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), sub.wait()).await, + Ok(Ok(())) + ); + } + + Ok(()) + } + + #[restate_core::test] + async fn non_existent_log_sub() -> anyhow::Result<()> { + let (handle, mut metadata_writer) = init().await?; + + let mut sub = handle + .subscribe_and_wait_for_armed( + LogId::new(1000), + ChainCondition::WritableSegmentAfter(SegmentIndex::from(0)), + ) + .await; + assert_eq!(sub.try_recv(), Err(oneshot::error::TryRecvError::Empty)); + + update_logs_metadata(&mut metadata_writer, |builder| { + builder + .add_log( + LogId::new(1000), + Chain::new(ProviderKind::InMemory, LogletParams::from("test2")), + ) + .unwrap(); + + let idx = builder + .chain(LogId::new(1000)) + .unwrap() + .append_segment( + Lsn::from(1), + ProviderKind::InMemory, + LogletParams::from("test2"), + ) + .unwrap(); + assert_eq!(idx, SegmentIndex::from(1)); + }) + .await?; + + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), sub.wait()).await, + Ok(Ok(())) + ); + + Ok(()) + } + + #[restate_core::test] + async fn dropped_subscriptions() -> anyhow::Result<()> { + let (handle, mut metadata_writer) = init().await?; + + // Two subscriptions, one is dropped while the other is still active. + let mut sub1 = handle + .subscribe_and_wait_for_armed( + LogId::new(1), + ChainCondition::Sealed(SegmentIndex::from(1)), + ) + .await; + assert_eq!(sub1.try_recv(), Err(oneshot::error::TryRecvError::Empty)); + + let mut sub2 = handle + .subscribe_and_wait_for_armed( + LogId::new(1), + ChainCondition::Sealed(SegmentIndex::from(1)), + ) + .await; + assert_eq!(sub2.try_recv(), Err(oneshot::error::TryRecvError::Empty)); + + // Drop sub1 + drop(sub1); + + update_logs_metadata(&mut metadata_writer, |builder| { + builder + .chain(LogId::new(1)) + .unwrap() + .append_segment( + Lsn::from(1), + ProviderKind::InMemory, + LogletParams::from("test2"), + ) + .unwrap(); + + builder + .chain(LogId::new(1)) + .unwrap() + .seal(Lsn::from(10), &SealMetadata::default()) + .unwrap(); + }) + .await?; + + // sub2 should fire as expected, with no problems from the dropped sub1. + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), sub2.wait()).await, + Ok(Ok(())) + ); + + Ok(()) + } + + #[restate_core::test] + async fn multiple_subs() -> anyhow::Result<()> { + let (handle, mut metadata_writer) = init().await?; + + // Initial chain state: + // log-1 -> 2 segments, segment 1 is open + // log-2 -> 1 sealed segment + update_logs_metadata(&mut metadata_writer, |builder| { + let idx = builder + .chain(LogId::new(1)) + .unwrap() + .append_segment( + Lsn::from(1), + ProviderKind::InMemory, + LogletParams::from("test2"), + ) + .unwrap(); + assert_eq!(idx, SegmentIndex::from(1)); + let idx = builder + .chain(LogId::new(1)) + .unwrap() + .append_segment( + Lsn::from(1), + ProviderKind::InMemory, + LogletParams::from("test2"), + ) + .unwrap(); + assert_eq!(idx, SegmentIndex::from(2)); + + builder + .chain(LogId::new(2)) + .unwrap() + .seal(Lsn::from(10), &SealMetadata::default()) + .unwrap(); + }) + .await?; + + // Sub 1 is waiting for segment 2 of log-1 to be sealed + let mut log1_sub = handle + .subscribe_and_wait_for_armed( + LogId::new(1), + ChainCondition::Sealed(SegmentIndex::from(2)), + ) + .await; + + // Sub 2 is waiting for the chain of log-2 to be writable + let mut log2_sub = handle + .subscribe_and_wait_for_armed( + LogId::new(2), + ChainCondition::WritableSegmentAfter(SegmentIndex::from(0)), + ) + .await; + + assert_eq!( + log1_sub.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + ); + assert_eq!( + log2_sub.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + ); + + // Seal segment 2 of log-1 + update_logs_metadata(&mut metadata_writer, |builder| { + builder + .chain(LogId::new(1)) + .unwrap() + .seal(Lsn::from(10), &SealMetadata::default()) + .unwrap(); + }) + .await?; + + // Sub 1 is notified of the sealed segment + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), log1_sub.wait()).await, + Ok(Ok(())) + ); + assert_eq!( + log2_sub.try_recv(), + Err(oneshot::error::TryRecvError::Empty) + ); + + // Add a writable segment to log-2 + update_logs_metadata(&mut metadata_writer, |builder| { + builder + .chain(LogId::new(2)) + .unwrap() + .append_segment( + Lsn::from(10), + ProviderKind::InMemory, + LogletParams::from("test2"), + ) + .unwrap(); + }) + .await?; + + // Sub 2 is notified of the writable segment + assert_eq!( + tokio::time::timeout(Duration::from_secs(1), log2_sub.wait()).await, + Ok(Ok(())) + ); + + Ok(()) + } +} diff --git a/crates/bifrost/src/service.rs b/crates/bifrost/src/service.rs index 1de3534bbc..100a1a5197 100644 --- a/crates/bifrost/src/service.rs +++ b/crates/bifrost/src/service.rs @@ -23,6 +23,7 @@ use restate_types::live::BoxLiveLoad; use restate_types::logs::metadata::ProviderKind; use crate::bifrost::BifrostInner; +use crate::log_chain_watcher::{LogChainWatcher, LogChainWatcherHandle, LogChainWatcherReceiver}; #[cfg(any(test, feature = "memory-loglet"))] use crate::providers::memory_loglet; use crate::watchdog::{Watchdog, WatchdogCommand}; @@ -32,6 +33,7 @@ pub struct BifrostService { inner: Arc, watchdog_tx: mpsc::UnboundedSender, watchdog_rx: mpsc::UnboundedReceiver, + log_chain_watcher_rx: LogChainWatcherReceiver, metadata_writer: MetadataWriter, factories: HashMap>, } @@ -39,7 +41,11 @@ pub struct BifrostService { impl BifrostService { pub fn new(metadata_writer: MetadataWriter) -> Self { let (watchdog_tx, watchdog_rx) = tokio::sync::mpsc::unbounded_channel(); - let inner = Arc::new(BifrostInner::new(watchdog_tx.clone())); + let (log_chain_watcher_tx, log_chain_watcher_rx) = tokio::sync::mpsc::unbounded_channel(); + let inner = Arc::new(BifrostInner::new( + watchdog_tx.clone(), + LogChainWatcherHandle::new(log_chain_watcher_tx), + )); Self { inner, @@ -47,6 +53,7 @@ impl BifrostService { watchdog_rx, metadata_writer, factories: HashMap::with_capacity(ProviderKind::LENGTH), + log_chain_watcher_rx, } } @@ -151,6 +158,7 @@ impl BifrostService { // We spawn the watchdog as a background long-running task Watchdog::start(self.inner, self.watchdog_rx, self.metadata_writer)?; + LogChainWatcher::start(self.log_chain_watcher_rx)?; // Bifrost started! Ok(()) diff --git a/crates/core/src/task_center/task_kind.rs b/crates/core/src/task_center/task_kind.rs index 06299a5c41..d4ba549036 100644 --- a/crates/core/src/task_center/task_kind.rs +++ b/crates/core/src/task_center/task_kind.rs @@ -124,6 +124,8 @@ pub enum TaskKind { // -- Bifrost Tasks #[strum(props(runtime = "default"))] BifrostWatchdog, + #[strum(props(runtime = "default"))] + BifrostLogChainWatcher, /// A background task that the system needs for its operation. The task requires a system /// shutdown on errors and the system will wait for its graceful cancellation on shutdown. #[strum(props(runtime = "default"))]