magi-code 0.77.1

use super::super::*;

/// A single owned outcome, published before its coalescible app wake-up.
/// Keep the handle until it finishes; receiving an outcome must never join it.
pub(super) struct CompletionWorker<T> {
    handle: JoinHandle<()>,
    receiver: Receiver<Result<T, String>>,
    consumed: bool,
}

impl<T: Send + 'static> CompletionWorker<T> {
    pub(super) fn spawn(
        name: &str,
        wake: Sender<TuiEvent>,
        task: impl FnOnce() -> Result<T, String> + Send + 'static,
    ) -> std::io::Result<Self> {
        let (sender, receiver) = crossbeam_channel::bounded(1);
        let handle = thread::Builder::new().name(name.into()).spawn(move || {
            let result = panic::catch_unwind(panic::AssertUnwindSafe(task))
                .unwrap_or_else(|_| Err("completion worker panicked".into()));
            // One producer and one result: this cannot wait for queue capacity.
            if sender.send(result).is_ok() {
                // Full means the app is already awake. Disconnection cannot
                // discard the owned outcome in the dedicated mailbox.
                let _ = wake.try_send(TuiEvent::WorkerOutcomeReady);
            }
        })?;
        Ok(Self {
            handle,
            receiver,
            consumed: false,
        })
    }

    pub(super) fn take_result(&mut self) -> Option<Result<T, String>> {
        if self.consumed {
            return None;
        }
        match self.receiver.try_recv() {
            Ok(result) => {
                self.consumed = true;
                Some(result)
            }
            Err(crossbeam_channel::TryRecvError::Disconnected) => {
                self.consumed = true;
                Some(Err("completion worker exited without an outcome".into()))
            }
            Err(crossbeam_channel::TryRecvError::Empty) => None,
        }
    }

    pub(super) fn is_finished(&self) -> bool {
        self.handle.is_finished()
    }

    pub(super) fn ready_to_reap(&self) -> bool {
        self.consumed && self.is_finished()
    }

    pub(super) fn join(self) -> Result<Option<T>, String> {
        self.handle
            .join()
            .map_err(|_| "completion worker panicked".to_string())?;
        if self.consumed {
            Ok(None)
        } else {
            self.receiver
                .recv()
                .map_err(|_| "completion worker exited without an outcome".to_string())?
                .map(Some)
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn outcome_can_be_applied_before_thread_exit_without_losing_join_ownership() {
        let (sender, receiver) = crossbeam_channel::bounded(1);
        let (release, wait) = crossbeam_channel::bounded(1);
        let handle = thread::spawn(move || {
            sender.send(Ok(String::from("saved"))).unwrap();
            wait.recv().unwrap();
        });
        let mut worker = CompletionWorker {
            handle,
            receiver,
            consumed: false,
        };
        let deadline = Instant::now() + Duration::from_secs(1);
        while worker.receiver.is_empty() {
            assert!(Instant::now() < deadline);
            thread::yield_now();
        }
        assert!(!worker.is_finished());
        assert_eq!(worker.take_result().unwrap().unwrap(), "saved");
        assert!(worker.take_result().is_none());
        assert!(!worker.ready_to_reap());
        release.send(()).unwrap();
        assert!(worker.join().unwrap().is_none());
    }

    #[test]
    fn saturated_or_disconnected_wake_queue_preserves_owned_result() {
        for disconnected in [false, true] {
            let (wake, events) = crossbeam_channel::bounded(1);
            wake.send(TuiEvent::WorkerOutcomeReady).unwrap();
            let events = if disconnected { None } else { Some(events) };
            let worker = CompletionWorker::spawn("test-completion", wake, || {
                Ok(String::from("durable result"))
            })
            .unwrap();
            let deadline = Instant::now() + Duration::from_secs(1);
            while !worker.is_finished() {
                assert!(Instant::now() < deadline, "wake delivery blocked worker");
                thread::yield_now();
            }
            assert_eq!(worker.join().unwrap().as_deref(), Some("durable result"));
            drop(events);
        }
    }

    #[test]
    fn panic_is_published_before_wake() {
        let (wake, events) = crossbeam_channel::bounded(1);
        let mut worker =
            CompletionWorker::<()>::spawn("test-panic", wake, || panic!("test failure")).unwrap();
        assert!(matches!(
            events.recv_timeout(Duration::from_secs(1)).unwrap(),
            TuiEvent::WorkerOutcomeReady
        ));
        assert!(worker.take_result().unwrap().is_err());
        assert!(worker.join().unwrap().is_none());
    }
}