aion-rs 0.31.0

//! Production [`WorkloopWaker`]: bring a suspended loop's current generation
//! to a live process (wake = load carry, run iteration, suspend — R13.3).
//!
//! The wake is the recovery machinery reused, not reinvented: the same
//! respawn-and-register path pause-resume uses for a non-resident run —
//! replay re-derives the generation's state from its own bounded history
//! segment, so the woken iteration continues exactly where its recorded
//! events (`WorkflowStarted` with the carry, `CadenceFired`) put it.

use aion_core::{Event, WorkflowId, WorkflowStatus, status_from_events};
use aion_store::EventStore;
use aion_store::visibility::VisibilityStore;
use async_trait::async_trait;
use std::sync::Arc;

use super::error::WorkloopError;
use super::service::WorkloopWaker;
use crate::durability::Recorder;
use crate::lifecycle::reopen::{self, ReopenWorkflowContext};
use crate::loader::WorkflowCatalog;
use crate::registry::Registry;
use crate::runtime::RuntimeHandle;
use crate::supervision::SupervisionTree;

/// Respawn-backed waker used by the engine's cadence service.
pub struct EngineWorkloopWaker {
    store: Arc<dyn EventStore>,
    visibility_store: Arc<dyn VisibilityStore>,
    catalog: Arc<WorkflowCatalog>,
    runtime: Arc<RuntimeHandle>,
    supervision: Arc<SupervisionTree>,
    registry: Arc<Registry>,
    search_attribute_schema: Arc<aion_core::SearchAttributeSchema>,
}

impl EngineWorkloopWaker {
    /// Builds the waker over the engine's recovery components.
    #[must_use]
    pub fn new(
        store: Arc<dyn EventStore>,
        visibility_store: Arc<dyn VisibilityStore>,
        catalog: Arc<WorkflowCatalog>,
        runtime: Arc<RuntimeHandle>,
        supervision: Arc<SupervisionTree>,
        registry: Arc<Registry>,
        search_attribute_schema: Arc<aion_core::SearchAttributeSchema>,
    ) -> Self {
        Self {
            store,
            visibility_store,
            catalog,
            runtime,
            supervision,
            registry,
            search_attribute_schema,
        }
    }
}

#[async_trait]
impl WorkloopWaker for EngineWorkloopWaker {
    async fn wake(&self, loop_id: &WorkflowId) -> Result<(), WorkloopError> {
        let history = self.store.read_history(loop_id).await?;
        let Some(run_id) = history.iter().rev().find_map(|event| match event {
            Event::WorkflowStarted { run_id, .. } => Some(run_id.clone()),
            _ => None,
        }) else {
            return Err(WorkloopError::Engine {
                reason: format!("workloop {loop_id} has no recorded generation to wake"),
            });
        };

        // Already resident: the generation is running its iteration; the
        // recorded fire is in history and replay/live code observes it there.
        if self
            .registry
            .get(loop_id, &run_id)
            .map_err(|error| WorkloopError::Engine {
                reason: error.to_string(),
            })?
            .is_some()
        {
            return Ok(());
        }

        let status = status_from_events(&history);
        if status != WorkflowStatus::Running {
            return Err(WorkloopError::Engine {
                reason: format!("workloop {loop_id} is {status:?}, not Running; refusing to wake"),
            });
        }

        // One-shot recorder at the head (no registry entry exists, so this is
        // the run's single writer), handed whole to respawn-and-register for
        // one-Recorder continuity — the same shape pause-resume uses.
        let head = history.iter().map(Event::seq).max().unwrap_or_default();
        let recorder = Recorder::resume_at(loop_id.clone(), Arc::clone(&self.store), head)
            .with_visibility(run_id.clone(), Arc::clone(&self.visibility_store));

        let context = ReopenWorkflowContext {
            store: Arc::clone(&self.store),
            visibility_store: Arc::clone(&self.visibility_store),
            catalog: Arc::clone(&self.catalog),
            runtime: &self.runtime,
            supervision: Arc::clone(&self.supervision),
            registry: &self.registry,
            search_attribute_schema: Arc::clone(&self.search_attribute_schema),
        };
        let segment = aion_core::run_segment(&history, &run_id);
        let rearm = reopen::rearmable_timers(segment);
        let handle = reopen::respawn_and_register(&context, loop_id, &run_id, &history, recorder)
            .await
            .map_err(|error| WorkloopError::Engine {
                reason: format!("waking workloop {loop_id} failed: {error}"),
            })?;
        reopen::rearm_reopened_timers(&context, loop_id, handle.pid(), &rearm)
            .await
            .map_err(|error| WorkloopError::Engine {
                reason: format!("re-arming woken workloop {loop_id} timers failed: {error}"),
            })?;
        Ok(())
    }
}