663 lines
31 KiB
Rust
663 lines
31 KiB
Rust
// file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs
|
|
// version: 17
|
|
|
|
type PersistencePort = std::sync::Arc<dyn crate::RawTransactionIngestPersistencePort + 'static>;
|
|
type PersistenceTasks = tokio::task::JoinSet<ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>>;
|
|
type ProcessingFrontierReceiver = tokio::sync::watch::Receiver<crate::RawTransactionIngestProcessingFrontierProjection>;
|
|
type SourceTasks = tokio::task::JoinSet<ksp_core_lib::Result<()>>;
|
|
|
|
/// Runtime-neutral boxed future resolving after one RAW transaction ingest Worker has fully reached a terminal lifecycle state.
|
|
pub type RawTransactionIngestTerminalFuture<'a> =
|
|
std::pin::Pin<std::boxed::Box<dyn std::future::Future<Output = ksp_core_lib::Result<ksp_worker_api::WorkerState>> + std::marker::Send + 'a>>;
|
|
|
|
/// Cloneable external control handle for one continuous RAW transaction ingest Worker.
|
|
#[derive(Clone)]
|
|
pub struct RawTransactionIngestHandle {
|
|
snapshots: crate::RawTransactionIngestSnapshotSource,
|
|
stop_sender: tokio::sync::watch::Sender<bool>,
|
|
stop_token: ksp_worker_api::WorkerStopToken,
|
|
}
|
|
|
|
impl crate::RawTransactionIngestHandle {
|
|
/// Requests cooperative stop and returns `true` only for the first request accepted by the live runtime.
|
|
#[must_use]
|
|
pub fn request_stop(&self) -> bool {
|
|
if !self.stop_token.request_stop() {
|
|
return false;
|
|
}
|
|
return self.stop_sender.send(true).is_ok();
|
|
}
|
|
|
|
/// Returns an independent latest-value source for concrete RAW transaction ingest Worker snapshots.
|
|
#[must_use]
|
|
pub fn snapshot_source(&self) -> crate::RawTransactionIngestSnapshotSource {
|
|
return self.snapshots.clone();
|
|
}
|
|
|
|
/// Returns an independent source implementing the common [`ksp_worker_api::WorkerSnapshotSource`] projection.
|
|
#[must_use]
|
|
pub fn worker_snapshot_source(&self) -> crate::RawTransactionIngestSnapshotSource {
|
|
return self.snapshots.clone();
|
|
}
|
|
|
|
/// Waits until the private runtime task has published one terminal lifecycle state after draining owned work.
|
|
#[must_use]
|
|
pub fn wait_terminal(&self) -> crate::RawTransactionIngestTerminalFuture<'_> {
|
|
let source = self.snapshots.clone();
|
|
return std::boxed::Box::pin(async move {
|
|
loop {
|
|
let current = source.current();
|
|
let state = current.worker_snapshot().state();
|
|
if state.is_terminal() {
|
|
return std::result::Result::Ok(state);
|
|
}
|
|
let observed = current.worker_snapshot().sequence();
|
|
let changed = source.wait_for_change(observed).await;
|
|
let state = changed.worker_snapshot().state();
|
|
if state.is_terminal() {
|
|
return std::result::Result::Ok(state);
|
|
}
|
|
if changed.worker_snapshot().sequence() == observed && source.is_closed() {
|
|
return std::result::Result::Err(crate::runtime_error("terminal.closed_before_terminal"));
|
|
}
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
impl std::fmt::Debug for crate::RawTransactionIngestHandle {
|
|
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
|
let snapshot = self.snapshots.current();
|
|
return formatter
|
|
.debug_struct("RawTransactionIngestHandle")
|
|
.field("stop_requested", &self.stop_token.is_stop_requested())
|
|
.field("sequence", &snapshot.worker_snapshot().sequence())
|
|
.field("state", &snapshot.worker_snapshot().state())
|
|
.finish();
|
|
}
|
|
}
|
|
|
|
/// Entry point owning synchronous validation and task launch for one RAW transaction ingest Worker run.
|
|
pub struct RawTransactionIngestWorker;
|
|
|
|
impl crate::RawTransactionIngestWorker {
|
|
/// Starts one Worker on the caller-owned current Tokio runtime while retaining the caller-owned Store facade through an `Arc`.
|
|
pub fn start(
|
|
settings: crate::RawTransactionIngestSettings,
|
|
store: std::sync::Arc<ksp_store_lib::Store>,
|
|
) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle> {
|
|
let runtime = match current_runtime_handle() {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let store_snapshot = store.runtime_snapshot();
|
|
if let std::result::Result::Err(error) = validate_store_network(&settings, store_snapshot.network()) {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
return start_foundation(settings, runtime, std::option::Option::Some(store));
|
|
}
|
|
|
|
/// Starts one Worker with caller-composed runtime resources.
|
|
///
|
|
/// Every validated source in the bounded runtime-resource aggregate is started concurrently under one private source supervisor. Any configured source
|
|
/// failure remains terminal for the Worker because this release does not infer equivalent coverage or failover between source families.
|
|
pub fn start_with_runtime_resources(
|
|
settings: crate::RawTransactionIngestSettings,
|
|
store: std::sync::Arc<ksp_store_lib::Store>,
|
|
runtime_resources: crate::RawTransactionIngestRuntimeResources,
|
|
) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle> {
|
|
let runtime = match current_runtime_handle() {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(error) => return std::result::Result::Err(error),
|
|
};
|
|
let store_snapshot = store.runtime_snapshot();
|
|
if let std::result::Result::Err(error) = validate_store_network(&settings, store_snapshot.network()) {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
if let std::result::Result::Err(error) = runtime_resources.validate_network(settings.network()) {
|
|
return std::result::Result::Err(error);
|
|
}
|
|
let source_total = runtime_resources.source_count();
|
|
let source_settings = settings.clone();
|
|
let (processing_frontier_sender, processing_frontier_receiver) =
|
|
tokio::sync::watch::channel(crate::RawTransactionIngestProcessingFrontierProjection::empty());
|
|
let port: PersistencePort = store;
|
|
return start_foundation_with_port_source_spawner_and_frontier(
|
|
settings,
|
|
runtime,
|
|
std::option::Option::Some(port),
|
|
std::option::Option::Some(processing_frontier_receiver),
|
|
source_total,
|
|
move |children, stop_receiver, admission_sender| {
|
|
let _abort_handle = children.spawn(async move {
|
|
return runtime_resources.run_live_sources(source_settings, stop_receiver, admission_sender, processing_frontier_sender).await;
|
|
});
|
|
},
|
|
);
|
|
}
|
|
}
|
|
|
|
fn begin_stopping(
|
|
lifecycle: &mut ksp_worker_api::WorkerLifecycle,
|
|
snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
|
|
admission_queue_depth: usize,
|
|
in_flight_persistence: usize,
|
|
) -> std::option::Option<ksp_core_lib::ErrorCode> {
|
|
if lifecycle.mark_stopping().is_err() {
|
|
return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
|
|
}
|
|
if let std::result::Result::Err(error) = snapshots.publish_state(lifecycle.state(), admission_queue_depth, in_flight_persistence) {
|
|
return std::option::Option::Some(error.code());
|
|
}
|
|
return std::option::Option::None;
|
|
}
|
|
|
|
fn current_runtime_handle() -> ksp_core_lib::Result<tokio::runtime::Handle> {
|
|
return match tokio::runtime::Handle::try_current() {
|
|
std::result::Result::Ok(value) => std::result::Result::Ok(value),
|
|
std::result::Result::Err(_) => std::result::Result::Err(crate::runtime_error("start.runtime_unavailable")),
|
|
};
|
|
}
|
|
|
|
async fn drain_children(
|
|
state: ksp_worker_api::WorkerState,
|
|
children: &mut SourceTasks,
|
|
snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
|
|
) -> std::option::Option<ksp_core_lib::ErrorCode> {
|
|
let mut fault = std::option::Option::None;
|
|
while let std::option::Option::Some(joined) = children.join_next().await {
|
|
let current = source_completion(joined, state, 0, 0, snapshots);
|
|
fault = merge_fault(fault, current);
|
|
}
|
|
return fault;
|
|
}
|
|
|
|
async fn drain_persistence(
|
|
lifecycle: &ksp_worker_api::WorkerLifecycle,
|
|
admission: &crate::RawTransactionAdmission,
|
|
persistence: &mut PersistenceTasks,
|
|
snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
|
|
) -> std::option::Option<ksp_core_lib::ErrorCode> {
|
|
let mut fault = std::option::Option::None;
|
|
while let std::option::Option::Some(joined) = persistence.join_next().await {
|
|
let current = persistence_completion(joined, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots);
|
|
fault = merge_fault(fault, current);
|
|
}
|
|
return fault;
|
|
}
|
|
|
|
async fn drain_admission_and_persistence(
|
|
settings: &crate::RawTransactionIngestSettings,
|
|
lifecycle: &ksp_worker_api::WorkerLifecycle,
|
|
admission: &mut crate::RawTransactionAdmission,
|
|
persistence: &mut PersistenceTasks,
|
|
persistence_convergence: &std::sync::Arc<crate::RawTransactionIngestPersistenceConvergence>,
|
|
port: &std::option::Option<PersistencePort>,
|
|
snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
|
|
) -> std::option::Option<ksp_core_lib::ErrorCode> {
|
|
let mut fault = std::option::Option::None;
|
|
admission.close();
|
|
loop {
|
|
while persistence.len() >= settings.persistence_concurrency() {
|
|
let joined = persistence.join_next().await;
|
|
let current = match joined {
|
|
std::option::Option::Some(value) => persistence_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots),
|
|
std::option::Option::None => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID),
|
|
};
|
|
fault = merge_fault(fault, current);
|
|
}
|
|
let received = admission.receive(settings.network()).await;
|
|
let backpressure_wait_observed = admission.take_backpressure_wait_observed();
|
|
match received {
|
|
std::result::Result::Ok(std::option::Option::Some(acquisition)) => {
|
|
let spawned = spawn_persistence(persistence, persistence_convergence, port, acquisition);
|
|
let published = snapshots.record_admission_success(lifecycle.state(), admission.queue_depth(), persistence.len(), backpressure_wait_observed);
|
|
if !spawned && fault.is_none() {
|
|
fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
|
|
}
|
|
if let std::result::Result::Err(error) = published
|
|
&& fault.is_none()
|
|
{
|
|
fault = std::option::Option::Some(error.code());
|
|
}
|
|
},
|
|
std::result::Result::Ok(std::option::Option::None) => break,
|
|
std::result::Result::Err(_) => {
|
|
let published = snapshots.record_admission_failure(lifecycle.state(), admission.queue_depth(), persistence.len(), backpressure_wait_observed);
|
|
match published {
|
|
std::result::Result::Ok(()) => {
|
|
if fault.is_none() {
|
|
fault = std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
|
|
}
|
|
},
|
|
std::result::Result::Err(error) => {
|
|
if fault.is_none() {
|
|
fault = std::option::Option::Some(error.code());
|
|
}
|
|
},
|
|
}
|
|
},
|
|
}
|
|
}
|
|
let persistence_fault = drain_persistence(lifecycle, admission, persistence, snapshots).await;
|
|
fault = merge_fault(fault, persistence_fault);
|
|
return fault;
|
|
}
|
|
|
|
async fn drain_owned_work(
|
|
settings: &crate::RawTransactionIngestSettings,
|
|
lifecycle: &ksp_worker_api::WorkerLifecycle,
|
|
children: &mut SourceTasks,
|
|
admission: &mut crate::RawTransactionAdmission,
|
|
persistence: &mut PersistenceTasks,
|
|
persistence_convergence: &std::sync::Arc<crate::RawTransactionIngestPersistenceConvergence>,
|
|
port: &std::option::Option<PersistencePort>,
|
|
snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
|
|
) -> std::option::Option<ksp_core_lib::ErrorCode> {
|
|
let drain = async {
|
|
let mut fault = drain_admission_and_persistence(settings, lifecycle, admission, persistence, persistence_convergence, port, snapshots).await;
|
|
let child_fault = drain_children(lifecycle.state(), children, snapshots).await;
|
|
fault = merge_fault(fault, child_fault);
|
|
return fault;
|
|
};
|
|
let timed = tokio::time::timeout(settings.shutdown_drain_timeout(), drain).await;
|
|
return match timed {
|
|
std::result::Result::Ok(fault) => fault,
|
|
std::result::Result::Err(_) => {
|
|
admission.close();
|
|
persistence.abort_all();
|
|
children.abort_all();
|
|
while persistence.join_next().await.is_some() {}
|
|
while children.join_next().await.is_some() {}
|
|
std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT)
|
|
},
|
|
};
|
|
}
|
|
|
|
fn finish_faulted(
|
|
lifecycle: &mut ksp_worker_api::WorkerLifecycle,
|
|
snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
|
|
code: ksp_core_lib::ErrorCode,
|
|
) {
|
|
if lifecycle.fault(code).is_err() {
|
|
snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
|
|
return;
|
|
}
|
|
if snapshots.publish_state(lifecycle.state(), 0, 0).is_err() {
|
|
snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED));
|
|
}
|
|
return;
|
|
}
|
|
|
|
fn finish_stopped(lifecycle: &mut ksp_worker_api::WorkerLifecycle, snapshots: &mut crate::RawTransactionIngestSnapshotPublisher) {
|
|
if lifecycle.mark_stopped().is_err() {
|
|
snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
|
|
return;
|
|
}
|
|
if snapshots.publish_state(lifecycle.state(), 0, 0).is_err() {
|
|
snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED));
|
|
}
|
|
return;
|
|
}
|
|
|
|
fn merge_fault(
|
|
first: std::option::Option<ksp_core_lib::ErrorCode>,
|
|
next: std::option::Option<ksp_core_lib::ErrorCode>,
|
|
) -> std::option::Option<ksp_core_lib::ErrorCode> {
|
|
if first.is_some() {
|
|
return first;
|
|
}
|
|
return next;
|
|
}
|
|
|
|
fn persistence_completion(
|
|
joined: std::result::Result<ksp_core_lib::Result<crate::RawTransactionIngestPersistenceOutcome>, tokio::task::JoinError>,
|
|
state: ksp_worker_api::WorkerState,
|
|
admission_queue_depth: usize,
|
|
in_flight_persistence: usize,
|
|
snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
|
|
) -> std::option::Option<ksp_core_lib::ErrorCode> {
|
|
return match joined {
|
|
std::result::Result::Ok(std::result::Result::Ok(outcome)) => {
|
|
match snapshots.record_persistence_success(state, admission_queue_depth, in_flight_persistence, outcome) {
|
|
std::result::Result::Ok(()) => std::option::Option::None,
|
|
std::result::Result::Err(error) => std::option::Option::Some(error.code()),
|
|
}
|
|
},
|
|
std::result::Result::Ok(std::result::Result::Err(error)) => {
|
|
let code = error.code();
|
|
let published = snapshots.record_persistence_fault(state, admission_queue_depth, in_flight_persistence, code);
|
|
match published {
|
|
std::result::Result::Ok(()) => std::option::Option::Some(code),
|
|
std::result::Result::Err(snapshot_error) => std::option::Option::Some(snapshot_error.code()),
|
|
}
|
|
},
|
|
std::result::Result::Err(_) => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID),
|
|
};
|
|
}
|
|
|
|
fn source_completion(
|
|
joined: std::result::Result<ksp_core_lib::Result<()>, tokio::task::JoinError>,
|
|
state: ksp_worker_api::WorkerState,
|
|
admission_queue_depth: usize,
|
|
in_flight_persistence: usize,
|
|
snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
|
|
) -> std::option::Option<ksp_core_lib::ErrorCode> {
|
|
let source_code = match joined {
|
|
std::result::Result::Ok(std::result::Result::Ok(())) => return std::option::Option::None,
|
|
std::result::Result::Ok(std::result::Result::Err(error)) if error.code() == crate::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED => {
|
|
crate::ERROR_CODE_RAW_TRANSACTION_INGEST_COUNTER_EXHAUSTED
|
|
},
|
|
std::result::Result::Ok(std::result::Result::Err(_)) | std::result::Result::Err(_) => crate::ERROR_CODE_RAW_TRANSACTION_INGEST_SOURCE_FAILED,
|
|
};
|
|
let published = snapshots.record_source_failure(state, admission_queue_depth, in_flight_persistence);
|
|
return match published {
|
|
std::result::Result::Ok(()) => std::option::Option::Some(source_code),
|
|
std::result::Result::Err(error) => std::option::Option::Some(error.code()),
|
|
};
|
|
}
|
|
|
|
async fn run_supervisor<Spawner>(
|
|
settings: crate::RawTransactionIngestSettings,
|
|
mut lifecycle: ksp_worker_api::WorkerLifecycle,
|
|
port: std::option::Option<PersistencePort>,
|
|
mut stop_receiver: tokio::sync::watch::Receiver<bool>,
|
|
mut snapshots: crate::RawTransactionIngestSnapshotPublisher,
|
|
mut processing_frontier_receiver: std::option::Option<ProcessingFrontierReceiver>,
|
|
source_spawner: Spawner,
|
|
) where
|
|
Spawner:
|
|
FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver<bool>, tokio::sync::mpsc::Sender<crate::RawTransactionIngress>) + std::marker::Send + 'static,
|
|
{
|
|
if *stop_receiver.borrow() {
|
|
let stopping_fault = begin_stopping(&mut lifecycle, &mut snapshots, 0, 0);
|
|
match stopping_fault {
|
|
std::option::Option::Some(code) => finish_faulted(&mut lifecycle, &mut snapshots, code),
|
|
std::option::Option::None => finish_stopped(&mut lifecycle, &mut snapshots),
|
|
}
|
|
return;
|
|
}
|
|
if lifecycle.mark_running().is_err() {
|
|
snapshots.force_terminal(ksp_worker_api::WorkerState::Faulted(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID));
|
|
return;
|
|
}
|
|
if let std::result::Result::Err(error) = snapshots.publish_state(lifecycle.state(), 0, 0) {
|
|
finish_faulted(&mut lifecycle, &mut snapshots, error.code());
|
|
return;
|
|
}
|
|
let mut children = SourceTasks::new();
|
|
let mut persistence = PersistenceTasks::new();
|
|
let persistence_convergence = std::sync::Arc::new(crate::RawTransactionIngestPersistenceConvergence::new(
|
|
settings.admission_queue_capacity().max(settings.persistence_concurrency()),
|
|
));
|
|
let (source_stop_sender, source_stop_receiver) = tokio::sync::watch::channel(false);
|
|
let (mut admission, admission_sender) = crate::RawTransactionAdmission::new(settings.admission_queue_capacity());
|
|
source_spawner(&mut children, source_stop_receiver, admission_sender);
|
|
let mut fault = supervise_until_stop(
|
|
&settings,
|
|
&lifecycle,
|
|
&source_stop_sender,
|
|
&mut stop_receiver,
|
|
&mut children,
|
|
&mut admission,
|
|
&mut persistence,
|
|
&persistence_convergence,
|
|
&port,
|
|
&mut snapshots,
|
|
&mut processing_frontier_receiver,
|
|
)
|
|
.await;
|
|
source_stop_sender.send_replace(true);
|
|
let stopping_fault = begin_stopping(&mut lifecycle, &mut snapshots, admission.queue_depth(), persistence.len());
|
|
fault = merge_fault(fault, stopping_fault);
|
|
let drain_fault =
|
|
drain_owned_work(&settings, &lifecycle, &mut children, &mut admission, &mut persistence, &persistence_convergence, &port, &mut snapshots).await;
|
|
match drain_fault {
|
|
std::option::Option::Some(code) if code == crate::ERROR_CODE_RAW_TRANSACTION_INGEST_DRAIN_TIMEOUT => {
|
|
fault = std::option::Option::Some(code);
|
|
},
|
|
std::option::Option::Some(code) => {
|
|
fault = merge_fault(fault, std::option::Option::Some(code));
|
|
},
|
|
std::option::Option::None => {},
|
|
}
|
|
match fault {
|
|
std::option::Option::Some(code) => finish_faulted(&mut lifecycle, &mut snapshots, code),
|
|
std::option::Option::None => finish_stopped(&mut lifecycle, &mut snapshots),
|
|
}
|
|
return;
|
|
}
|
|
|
|
fn spawn_persistence(
|
|
persistence: &mut PersistenceTasks,
|
|
persistence_convergence: &std::sync::Arc<crate::RawTransactionIngestPersistenceConvergence>,
|
|
port: &std::option::Option<PersistencePort>,
|
|
acquisition: ksp_raw_transaction_lib::RawTransactionAcquisition,
|
|
) -> bool {
|
|
let port = match port {
|
|
std::option::Option::Some(value) => std::sync::Arc::clone(value),
|
|
std::option::Option::None => return false,
|
|
};
|
|
let persistence_convergence = std::sync::Arc::clone(persistence_convergence);
|
|
let _abort_handle = persistence.spawn(async move {
|
|
return crate::persist_raw_transaction_ingest_converged_acquisition(port.as_ref(), acquisition, persistence_convergence.as_ref()).await;
|
|
});
|
|
return true;
|
|
}
|
|
|
|
fn start_foundation(
|
|
settings: crate::RawTransactionIngestSettings,
|
|
runtime: tokio::runtime::Handle,
|
|
store_guard: std::option::Option<std::sync::Arc<ksp_store_lib::Store>>,
|
|
) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle> {
|
|
return start_foundation_with_source_spawner(settings, runtime, store_guard, |_children, _stop_receiver, _admission_sender| {});
|
|
}
|
|
|
|
fn start_foundation_with_port_and_source_spawner<Spawner>(
|
|
settings: crate::RawTransactionIngestSettings,
|
|
runtime: tokio::runtime::Handle,
|
|
port: std::option::Option<PersistencePort>,
|
|
source_spawner: Spawner,
|
|
) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle>
|
|
where
|
|
Spawner:
|
|
FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver<bool>, tokio::sync::mpsc::Sender<crate::RawTransactionIngress>) + std::marker::Send + 'static,
|
|
{
|
|
return start_foundation_with_port_source_spawner_and_frontier(settings, runtime, port, std::option::Option::None, 0, source_spawner);
|
|
}
|
|
|
|
fn start_foundation_with_port_source_spawner_and_frontier<Spawner>(
|
|
settings: crate::RawTransactionIngestSettings,
|
|
runtime: tokio::runtime::Handle,
|
|
port: std::option::Option<PersistencePort>,
|
|
processing_frontier_receiver: std::option::Option<ProcessingFrontierReceiver>,
|
|
source_total: usize,
|
|
source_spawner: Spawner,
|
|
) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle>
|
|
where
|
|
Spawner:
|
|
FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver<bool>, tokio::sync::mpsc::Sender<crate::RawTransactionIngress>) + std::marker::Send + 'static,
|
|
{
|
|
let kind = match ksp_worker_api::WorkerKindCode::new(crate::RAW_TRANSACTION_INGEST_WORKER_KIND_CODE) {
|
|
std::result::Result::Ok(value) => value,
|
|
std::result::Result::Err(_) => return std::result::Result::Err(crate::runtime_error("start.worker_kind_invalid")),
|
|
};
|
|
let mut lifecycle = ksp_worker_api::WorkerLifecycle::new(settings.worker_id().clone(), kind);
|
|
if lifecycle.start().is_err() {
|
|
return std::result::Result::Err(crate::runtime_error("start.lifecycle_invalid"));
|
|
}
|
|
let stop_token = ksp_worker_api::WorkerStopToken::new();
|
|
let (stop_sender, stop_receiver) = tokio::sync::watch::channel(false);
|
|
let (snapshots, snapshot_source) = crate::RawTransactionIngestSnapshotPublisher::new(&settings, &lifecycle, source_total);
|
|
let handle = crate::RawTransactionIngestHandle { snapshots: snapshot_source, stop_sender, stop_token };
|
|
std::mem::drop(runtime.spawn(run_supervisor(settings, lifecycle, port, stop_receiver, snapshots, processing_frontier_receiver, source_spawner)));
|
|
return std::result::Result::Ok(handle);
|
|
}
|
|
|
|
fn start_foundation_with_source_spawner<Spawner>(
|
|
settings: crate::RawTransactionIngestSettings,
|
|
runtime: tokio::runtime::Handle,
|
|
store_guard: std::option::Option<std::sync::Arc<ksp_store_lib::Store>>,
|
|
source_spawner: Spawner,
|
|
) -> ksp_core_lib::Result<crate::RawTransactionIngestHandle>
|
|
where
|
|
Spawner:
|
|
FnOnce(&mut SourceTasks, tokio::sync::watch::Receiver<bool>, tokio::sync::mpsc::Sender<crate::RawTransactionIngress>) + std::marker::Send + 'static,
|
|
{
|
|
let port = match store_guard {
|
|
std::option::Option::Some(store) => {
|
|
let port: PersistencePort = store;
|
|
std::option::Option::Some(port)
|
|
},
|
|
std::option::Option::None => std::option::Option::None,
|
|
};
|
|
return start_foundation_with_port_and_source_spawner(settings, runtime, port, source_spawner);
|
|
}
|
|
|
|
async fn supervise_until_stop(
|
|
settings: &crate::RawTransactionIngestSettings,
|
|
lifecycle: &ksp_worker_api::WorkerLifecycle,
|
|
source_stop_sender: &tokio::sync::watch::Sender<bool>,
|
|
stop_receiver: &mut tokio::sync::watch::Receiver<bool>,
|
|
children: &mut SourceTasks,
|
|
admission: &mut crate::RawTransactionAdmission,
|
|
persistence: &mut PersistenceTasks,
|
|
persistence_convergence: &std::sync::Arc<crate::RawTransactionIngestPersistenceConvergence>,
|
|
port: &std::option::Option<PersistencePort>,
|
|
snapshots: &mut crate::RawTransactionIngestSnapshotPublisher,
|
|
processing_frontier_receiver: &mut std::option::Option<ProcessingFrontierReceiver>,
|
|
) -> std::option::Option<ksp_core_lib::ErrorCode> {
|
|
let mut admission_open = true;
|
|
loop {
|
|
if *stop_receiver.borrow() {
|
|
source_stop_sender.send_replace(true);
|
|
break;
|
|
}
|
|
if children.is_empty() && !admission_open && persistence.is_empty() {
|
|
let changed = stop_receiver.changed().await;
|
|
if changed.is_err() || *stop_receiver.borrow() {
|
|
source_stop_sender.send_replace(true);
|
|
break;
|
|
}
|
|
continue;
|
|
}
|
|
tokio::select! {
|
|
biased;
|
|
changed = stop_receiver.changed() => {
|
|
if changed.is_err() || *stop_receiver.borrow() {
|
|
source_stop_sender.send_replace(true);
|
|
break;
|
|
}
|
|
}
|
|
processing_frontier = wait_processing_frontier(processing_frontier_receiver) => {
|
|
match processing_frontier {
|
|
std::option::Option::Some(projection) => {
|
|
let published = snapshots.record_processing_frontier(
|
|
lifecycle.state(),
|
|
admission.queue_depth(),
|
|
persistence.len(),
|
|
projection,
|
|
);
|
|
if let std::result::Result::Err(error) = published {
|
|
source_stop_sender.send_replace(true);
|
|
return std::option::Option::Some(error.code());
|
|
}
|
|
},
|
|
std::option::Option::None => {
|
|
*processing_frontier_receiver = std::option::Option::None;
|
|
},
|
|
}
|
|
}
|
|
joined = children.join_next(), if !children.is_empty() => {
|
|
let source_fault = match joined {
|
|
std::option::Option::Some(value) => source_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots),
|
|
std::option::Option::None => std::option::Option::None,
|
|
};
|
|
if source_fault.is_some() {
|
|
source_stop_sender.send_replace(true);
|
|
return source_fault;
|
|
}
|
|
}
|
|
joined = persistence.join_next(), if !persistence.is_empty() => {
|
|
match joined {
|
|
std::option::Option::Some(value) => {
|
|
let fault = persistence_completion(value, lifecycle.state(), admission.queue_depth(), persistence.len(), snapshots);
|
|
if fault.is_some() {
|
|
source_stop_sender.send_replace(true);
|
|
return fault;
|
|
}
|
|
},
|
|
std::option::Option::None => {},
|
|
}
|
|
}
|
|
received = admission.receive(settings.network()), if admission_open && persistence.len() < settings.persistence_concurrency() => {
|
|
match received {
|
|
std::result::Result::Ok(std::option::Option::Some(acquisition)) => {
|
|
let backpressure_wait_observed = admission.take_backpressure_wait_observed();
|
|
let spawned = spawn_persistence(persistence, persistence_convergence, port, acquisition);
|
|
let published = snapshots.record_admission_success(
|
|
lifecycle.state(),
|
|
admission.queue_depth(),
|
|
persistence.len(),
|
|
backpressure_wait_observed,
|
|
);
|
|
if !spawned {
|
|
source_stop_sender.send_replace(true);
|
|
return std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID);
|
|
}
|
|
if let std::result::Result::Err(error) = published {
|
|
source_stop_sender.send_replace(true);
|
|
return std::option::Option::Some(error.code());
|
|
}
|
|
},
|
|
std::result::Result::Ok(std::option::Option::None) => {
|
|
let _backpressure_wait_observed = admission.take_backpressure_wait_observed();
|
|
admission_open = false;
|
|
},
|
|
std::result::Result::Err(_) => {
|
|
let backpressure_wait_observed = admission.take_backpressure_wait_observed();
|
|
let published = snapshots.record_admission_failure(
|
|
lifecycle.state(),
|
|
admission.queue_depth(),
|
|
persistence.len(),
|
|
backpressure_wait_observed,
|
|
);
|
|
source_stop_sender.send_replace(true);
|
|
return match published {
|
|
std::result::Result::Ok(()) => std::option::Option::Some(crate::ERROR_CODE_RAW_TRANSACTION_INGEST_RUNTIME_INVALID),
|
|
std::result::Result::Err(error) => std::option::Option::Some(error.code()),
|
|
};
|
|
},
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return std::option::Option::None;
|
|
}
|
|
|
|
async fn wait_processing_frontier(
|
|
receiver: &mut std::option::Option<ProcessingFrontierReceiver>,
|
|
) -> std::option::Option<crate::RawTransactionIngestProcessingFrontierProjection> {
|
|
let receiver = match receiver.as_mut() {
|
|
std::option::Option::Some(value) => value,
|
|
std::option::Option::None => {
|
|
return std::future::pending::<std::option::Option<crate::RawTransactionIngestProcessingFrontierProjection>>().await;
|
|
},
|
|
};
|
|
if receiver.changed().await.is_err() {
|
|
return std::option::Option::None;
|
|
}
|
|
return std::option::Option::Some(*receiver.borrow_and_update());
|
|
}
|
|
|
|
fn validate_store_network(settings: &crate::RawTransactionIngestSettings, store_network: &ksp_store_lib::RawNetworkId) -> ksp_core_lib::Result<()> {
|
|
if settings.network() != store_network {
|
|
return std::result::Result::Err(crate::runtime_error("start.store_network_mismatch"));
|
|
}
|
|
return std::result::Result::Ok(());
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "../unit_tests/runtime.rs"]
|
|
mod tests;
|