aion-rs 0.10.0

Transport-agnostic Aion workflow engine with durability, replay, timers, and supervision.
Documentation 
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

//! Channel-parking loop for the runtime's singleton process-exit drainer.

use std::sync::Weak;
use std::sync::atomic::Ordering;

use beamr::scheduler::{ExitEventRecvError, ExitEventSubscription, Scheduler};

use crate::EngineError;

use super::{ProcessExitObservationFailure, ProcessExitRegistry};

pub(super) fn run(
    registry: &Weak<ProcessExitRegistry>,
    scheduler: &Scheduler,
    subscription: &ExitEventSubscription,
) -> Result<(), EngineError> {
    let mut shutdown_resynchronized = false;
    loop {
        let Some(registry) = registry.upgrade() else {
            return Ok(());
        };
        if registry.stop_drainer.load(Ordering::Acquire) {
            if !shutdown_resynchronized {
                registry.resynchronize(scheduler)?;
                shutdown_resynchronized = true;
            }
            if registry.all_owned_processes_terminal()? {
                return Ok(());
            }
        }
        #[cfg(test)]
        registry.pause_if_requested();
        match subscription.recv_timeout(registry.park_bound) {
            Ok(event) => {
                if let Err(error) = registry.process_event(scheduler, event) {
                    tracing::error!(%error, "process exit drainer invariant failed");
                    registry
                        .fail_unobserved(ProcessExitObservationFailure::OutcomeMissingAfterEvent);
                    return Err(error);
                }
            }
            Err(ExitEventRecvError::Timeout) => {}
            Err(ExitEventRecvError::Disconnected) => {
                registry.fail_unobserved(ProcessExitObservationFailure::EventStreamDisconnected);
                return Err(EngineError::ProcessExitEventStreamDisconnected);
            }
        }
    }
}