liminal-server 0.20.0

use std::any::Any;
use std::sync::{Arc, mpsc};
use std::thread::{self, JoinHandle};
use std::time::Duration;

use liminal::durability::StoreCounts;

use crate::ServerError;
use crate::server::shutdown::ShutdownHandle;

use super::cost::{CostWake, SelfWork};
use super::publisher::{CompletionFailureSignal, CostSink, PublisherStop, run_cost_publisher};

/// Owned records must finish their durable drain after peers have closed.
/// A separate stall bound covers that storage work without borrowing a peer-close deadline.
const OWNED_DURABLE_DRAIN_STOP: Duration = Duration::from_secs(5);

/// Owns the one publisher executor and its explicit completion signal.
pub(in crate::server) struct PublisherRuntime {
    stop: Arc<PublisherStop>,
    completed: mpsc::Receiver<()>,
    thread: Option<JoinHandle<Result<(), ServerError>>>,
    shutdown: ShutdownHandle,
    completion_error: Option<CompletionError>,
    on_completion_failure: Option<CompletionFailureSignal>,
    drain_bound_hits: u64,
    #[cfg(test)]
    completion_received: bool,
}

enum CompletionError {
    Owner(String),
    PublisherResult(String),
}

impl PublisherRuntime {
    pub(in crate::server) fn start<Sink>(
        wake: Arc<CostWake>,
        sink: Sink,
        shutdown: ShutdownHandle,
    ) -> Result<Self, ServerError>
    where
        Sink: CostSink + Send + 'static,
    {
        let on_completion_failure = sink.completion_failure_signal();
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_time()
            .build()
            .map_err(|error| owner_error("runtime construction", &error.to_string()))?;
        let stop = Arc::new(PublisherStop::default());
        let thread_stop = Arc::clone(&stop);
        let fatal_shutdown = shutdown.clone();
        let (completion, completed) = mpsc::sync_channel(1);
        let thread = thread::Builder::new()
            .name("liminal-self-streams".to_owned())
            .spawn(move || {
                let completion_guard = PublisherCompletion {
                    completion,
                    shutdown: fatal_shutdown,
                };
                let mut owned_sink = sink;
                let owned_runtime = runtime;
                let result = owned_runtime.block_on(run_cost_publisher(
                    &wake,
                    &thread_stop,
                    (StoreCounts::default(), SelfWork::default()),
                    tokio::time::sleep,
                    &mut owned_sink,
                ));
                if result.is_err() {
                    completion_guard.shutdown.initiate();
                }
                result
            })
            .map_err(|error| owner_error("thread construction", &error.to_string()))?;
        Ok(Self {
            stop,
            completed,
            thread: Some(thread),
            shutdown,
            completion_error: None,
            on_completion_failure,
            drain_bound_hits: 0,
            #[cfg(test)]
            completion_received: false,
        })
    }

    pub(in crate::server) fn finish(&mut self) -> Result<(), ServerError> {
        self.finish_with_bound(OWNED_DURABLE_DRAIN_STOP)
    }

    pub(super) fn finish_with_bound(&mut self, bound: Duration) -> Result<(), ServerError> {
        tracing::debug!("self-stream publisher stop requested");
        self.stop.request();
        if let Some(error) = &self.completion_error {
            return Err(match error {
                CompletionError::Owner(message) => ServerError::ListenerAccept {
                    message: message.clone(),
                },
                CompletionError::PublisherResult(first_error) => {
                    ServerError::SelfStreamPublisherAlreadyFinished {
                        first_error: first_error.clone(),
                    }
                }
            });
        }
        if self.thread.is_none() {
            return Ok(());
        }
        if let Err(error) = self.completed.recv_timeout(bound) {
            if matches!(error, mpsc::RecvTimeoutError::Timeout) {
                let Some(hits) = self.drain_bound_hits.checked_add(1) else {
                    return Err(
                        self.fail_completion("owned durable drain bound counter", "count overflow")
                    );
                };
                self.drain_bound_hits = hits;
                return Err(self.fail_completion("owned durable drain bound", &error.to_string()));
            }
            return Err(self.fail_completion("completion signal", &error.to_string()));
        }
        #[cfg(test)]
        {
            self.completion_received = true;
        }
        tracing::debug!("self-stream publisher completion received");
        let Some(thread) = self.thread.take() else {
            return Err(owner_error("join", "completed publisher thread is absent"));
        };
        let result = thread.join().map_err(|panic| {
            self.shutdown.initiate();
            let message = format!(
                "self-stream join panic failed: {}",
                panic_message(panic.as_ref())
            );
            self.completion_error = Some(CompletionError::Owner(message.clone()));
            ServerError::ListenerAccept { message }
        })?;
        if let Err(error) = &result {
            self.completion_error = Some(CompletionError::PublisherResult(error.to_string()));
        }
        result
    }

    fn fail_completion(&mut self, phase: &str, detail: &str) -> ServerError {
        let message = format!("self-stream {phase} failed: {detail}");
        self.completion_error = Some(CompletionError::Owner(message.clone()));
        let error = ServerError::ListenerAccept { message };
        if let Some(signal) = &self.on_completion_failure {
            signal(&error);
        }
        self.shutdown.initiate();
        error
    }

    #[cfg(test)]
    pub(super) fn request_stop_for_test(&self) {
        self.stop.request();
    }

    #[cfg(test)]
    pub(super) fn join_completed_for_test(&mut self) -> Result<(), ServerError> {
        self.completed
            .recv()
            .map_err(|error| owner_error("completion signal", &error.to_string()))?;
        let thread = self
            .thread
            .take()
            .ok_or_else(|| owner_error("join", "completed publisher thread is absent"))?;
        thread
            .join()
            .map_err(|panic| owner_error("join panic", panic_message(panic.as_ref())))?
    }

    #[cfg(test)]
    pub(super) fn completion_counts(&self) -> (u64, u64, u64) {
        (
            u64::from(self.completion_received),
            u64::from(self.completion_error.is_some()),
            self.drain_bound_hits,
        )
    }
}

struct PublisherCompletion {
    completion: mpsc::SyncSender<()>,
    shutdown: ShutdownHandle,
}

impl Drop for PublisherCompletion {
    fn drop(&mut self) {
        if thread::panicking() {
            tracing::error!("self-stream publisher thread panicked");
            self.shutdown.initiate();
        }
        if let Err(error) = self.completion.send(()) {
            tracing::error!(%error, "self-stream completion receiver closed");
            self.shutdown.initiate();
        }
    }
}

fn panic_message(panic: &(dyn Any + Send)) -> &str {
    match panic.downcast_ref::<String>() {
        Some(message) => message,
        None => match panic.downcast_ref::<&str>() {
            Some(message) => message,
            None => "non-string panic",
        },
    }
}

impl Drop for PublisherRuntime {
    fn drop(&mut self) {
        if self.thread.is_none() {
            return;
        }
        if self.completion_error.is_some() {
            tracing::error!(
                "self-stream publisher join remains unconfirmed after completion failure"
            );
            self.shutdown.initiate();
        } else if let Err(error) = self.finish() {
            tracing::error!(%error, "self-stream publisher teardown failed");
            self.shutdown.initiate();
        }
    }
}

fn owner_error(phase: &str, message: &str) -> ServerError {
    ServerError::ListenerAccept {
        message: format!("self-stream {phase} failed: {message}"),
    }
}