aion-rs 0.31.0

//! Delivering the signals that arrived while a workflow was not yet resident.
//!
//! Split out of `start.rs` as a named module: the start path is at the 500-line
//! production cap this codebase holds itself to, and this is a self-contained
//! unit rather than a slice taken to hit a number — one [`EngineHandle`]
//! adapter and the one call that drives it. Every other seam on that trait is
//! deliberately refused here, because a handoff flush is allowed to wake a
//! process and nothing else: it must not record, spawn, terminate, or arm.

use aion_core::{Event, WorkflowId};

use super::StartWorkflowContext;
use crate::engine_seam::{
    ChildWorkflowSpawnRequest, ChildWorkflowSpawnResult, EngineHandle, EngineSeamError,
    TimerWheelEntry, WorkflowMailboxMessage, WorkflowProcessHandle, WorkflowResidency,
};
use crate::registry::{HandleResidency, Registry, WorkflowHandle};
use crate::runtime::RuntimeHandle;

pub(super) fn deliver_deferred_signals(context: &StartWorkflowContext, handle: &WorkflowHandle) {
    let Some(handoff) = &context.signal_handoff else {
        return;
    };
    let adapter = StartResumeEngineHandle {
        runtime: &context.runtime,
        registry: &context.registry,
    };
    if let Err(error) = handoff.deliver_deferred(&adapter, handle.workflow_id()) {
        tracing::warn!(
            workflow_id = %handle.workflow_id(),
            error = %error,
            "failed to flush deferred signals after workflow became resident"
        );
    }
}

struct StartResumeEngineHandle<'a> {
    runtime: &'a RuntimeHandle,
    registry: &'a Registry,
}

impl EngineHandle for StartResumeEngineHandle<'_> {
    fn resolve_workflow(
        &self,
        workflow_id: &WorkflowId,
    ) -> Result<WorkflowResidency, EngineSeamError> {
        // aion#213: the workflow's ONE handle, or a typed refusal — see
        // `Registry::sole_handle`.
        let handle =
            self.registry
                .sole_handle(workflow_id)
                .map_err(|error| EngineSeamError::Delivery {
                    reason: error.to_string(),
                })?;
        match handle {
            Some(handle) if handle.residency() == HandleResidency::Resident => Ok(
                WorkflowResidency::Resident(WorkflowProcessHandle::new(handle.pid())),
            ),
            Some(_) => Ok(WorkflowResidency::NonResident),
            None => Ok(WorkflowResidency::Unknown),
        }
    }

    fn deliver_workflow_message(
        &self,
        process: WorkflowProcessHandle,
        message: WorkflowMailboxMessage,
    ) -> Result<(), EngineSeamError> {
        match message {
            WorkflowMailboxMessage::SignalReceived { .. } => self
                .runtime
                .deliver_signal_received(process.pid())
                .map_err(|error| EngineSeamError::Delivery {
                    reason: error.to_string(),
                }),
            other => Err(EngineSeamError::Delivery {
                reason: format!("unsupported resume handoff message: {other:?}"),
            }),
        }
    }

    fn spawn_child_workflow(
        &self,
        request: ChildWorkflowSpawnRequest,
    ) -> Result<ChildWorkflowSpawnResult, EngineSeamError> {
        let _ = request;
        Err(EngineSeamError::ChildSpawn {
            reason: "start resume handoff cannot spawn child workflows".to_owned(),
        })
    }

    fn terminate_linked_child_workflow(
        &self,
        parent_workflow_id: &WorkflowId,
        child_process: WorkflowProcessHandle,
        correlation: u64,
    ) -> Result<(), EngineSeamError> {
        let _ = (parent_workflow_id, child_process, correlation);
        Err(EngineSeamError::ChildTermination {
            reason: "start resume handoff cannot terminate child workflows".to_owned(),
        })
    }

    fn terminate_linked_activity(
        &self,
        parent_workflow_id: &WorkflowId,
        activity_process: crate::Pid,
        correlation: u64,
    ) -> Result<(), EngineSeamError> {
        let _ = (parent_workflow_id, activity_process, correlation);
        Err(EngineSeamError::ChildTermination {
            reason: "start resume handoff cannot terminate activities".to_owned(),
        })
    }

    fn arm_timer(&self, entry: TimerWheelEntry) -> Result<(), EngineSeamError> {
        let _ = entry;
        Err(EngineSeamError::TimerWheel {
            reason: "start resume handoff cannot arm timers".to_owned(),
        })
    }

    fn disarm_timer(
        &self,
        process: WorkflowProcessHandle,
        timer_id: &aion_core::TimerId,
    ) -> Result<(), EngineSeamError> {
        let _ = (process, timer_id);
        Err(EngineSeamError::TimerWheel {
            reason: "start resume handoff cannot disarm timers".to_owned(),
        })
    }

    fn record_workflow_event(
        &self,
        workflow_id: &WorkflowId,
        event: Event,
    ) -> Result<crate::engine_seam::RecordOutcome, EngineSeamError> {
        let _ = (workflow_id, event);
        Err(EngineSeamError::Recorder {
            reason: "start resume handoff cannot record workflow events".to_owned(),
        })
    }

    fn record_redelivered_timer_fire(
        &self,
        workflow_id: &WorkflowId,
        timer_id: &aion_core::TimerId,
    ) -> Result<crate::engine_seam::RedeliveredFire, EngineSeamError> {
        let _ = (workflow_id, timer_id);
        Err(EngineSeamError::Recorder {
            reason: "start resume handoff cannot answer timer redeliveries".to_owned(),
        })
    }
}