// file: crates/ksp-worker-raw-transaction-ingest-lib/src/runtime.rs
// version: 18

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().clone());
}

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;