1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
use error::PoolErrorKind;
use pool_runtime::{Init, PoolRuntime};
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use stratum_apps::bitcoin_core_sv2::common::template_distribution_protocol::CancellationToken;
use tokio::sync::Notify;
use tracing::info;
use crate::config::PoolConfig;
pub mod channel_manager;
pub mod config;
pub mod downstream;
pub mod error;
mod io_task;
#[cfg(feature = "monitoring")]
mod monitoring;
mod pool_runtime;
pub mod template_receiver;
pub mod utils;
#[derive(Debug, Clone)]
pub struct PoolSv2 {
config: PoolConfig,
cancellation_token: CancellationToken,
shutdown_notify: Arc<Notify>,
is_alive: Arc<AtomicBool>,
}
#[cfg_attr(not(test), hotpath::measure_all)]
impl PoolSv2 {
pub fn new(config: PoolConfig) -> Self {
Self {
config,
cancellation_token: CancellationToken::new(),
shutdown_notify: Arc::new(Notify::new()),
is_alive: Arc::new(AtomicBool::new(true)),
}
}
/// Starts the Pool server and blocks asynchronously on the [`PoolRuntime`].
///
/// The startup and execution sequence follows:
/// 1. **Initialize:** Sets up the pool runtime state machine.
/// 2. **Bootstrap:** Configures internal channels, starts the Job Declarator Server (JDS),
/// connects to the Template Provider, and initializes the Channel Manager.
/// 3. **Run & Block:** Spawns active background loops/servers and blocks the caller while
/// awaiting the runtime's shutdown signal (e.g., Ctrl+C or program cancellation).
/// 4. **Teardown:** Performs a coordinated graceful cleanup of all services and tasks,
/// remaining blocked until all sub-services have exited.
///
/// If any error occurs during bootstrapping, the runtime automatically initiates a
/// graceful shutdown of any partially initialized components before returning the error.
pub async fn start(&self) -> Result<(), PoolErrorKind> {
let runtime = PoolRuntime::<Init>::new(self.clone())?;
let runtime = match runtime.bootstrap().await {
Ok(runtime) => runtime,
Err(err) => {
return Err(err);
}
};
runtime.wait_for_shutdown().await;
runtime.shutdown().await;
Ok(())
}
pub async fn shutdown(&self) {
if !self.is_alive.load(Ordering::Relaxed) {
return;
}
// The Notified future is guaranteed to receive wakeups from notify_waiters()
// as soon as it has been created, even if it has not yet been polled.
let notified = self.shutdown_notify.notified();
self.cancellation_token.cancel();
notified.await;
}
}
impl Drop for PoolSv2 {
fn drop(&mut self) {
info!("PoolSv2 dropped");
self.cancellation_token.cancel();
}
}