use crate::state_table::SessionId;
use std::sync::Arc;
use tracing::{debug, error, info};
use crate::{
adapters::{dialog_adapter::DialogAdapter, media_adapter::MediaAdapter},
cleanup_diag::{self, CleanupStage},
session_store::{SessionState, SessionStore},
state_table::{Action, EventTemplate, EventType, StateKey, Transition, MASTER_TABLE},
types::CallState,
};
use super::{actions, guards};
#[derive(Debug, Clone)]
pub struct ProcessEventResult {
pub old_state: CallState,
pub next_state: Option<CallState>,
pub transition: Option<Transition>,
pub actions_executed: Vec<Action>,
pub events_published: Vec<EventTemplate>,
}
pub struct StateMachine {
table: Arc<crate::state_table::MasterStateTable>,
pub(crate) store: Arc<SessionStore>,
dialog_adapter: Arc<DialogAdapter>,
media_adapter: Arc<MediaAdapter>,
event_tx: Option<tokio::sync::mpsc::Sender<SessionEvent>>,
auto_180_ringing: bool,
}
#[derive(Debug, Clone)]
pub enum PendingOptionsSlot {
Invite(Arc<crate::api::send::outbound_call::OutboundCallOptionsSnapshot>),
ReInvite(Arc<rvoip_sip_dialog::api::unified::ReInviteRequestOptions>),
Register(Arc<rvoip_sip_dialog::api::unified::RegisterRequestOptions>),
Refer(Arc<rvoip_sip_dialog::api::unified::ReferRequestOptions>),
Bye(Arc<rvoip_sip_dialog::api::unified::ByeRequestOptions>),
Cancel(Arc<rvoip_sip_dialog::api::unified::CancelRequestOptions>),
Notify(Arc<rvoip_sip_dialog::api::unified::NotifyRequestOptions>),
Subscribe(Arc<rvoip_sip_dialog::api::unified::SubscribeRequestOptions>),
Info(Arc<rvoip_sip_dialog::api::unified::InfoRequestOptions>),
Update(Arc<rvoip_sip_dialog::api::unified::UpdateRequestOptions>),
Message(Arc<rvoip_sip_dialog::api::unified::MessageRequestOptions>),
Options(Arc<rvoip_sip_dialog::api::unified::OptionsRequestOptions>),
}
impl PendingOptionsSlot {
pub fn method(&self) -> rvoip_sip_core::Method {
use rvoip_sip_core::Method;
match self {
Self::Invite(_) | Self::ReInvite(_) => Method::Invite,
Self::Register(_) => Method::Register,
Self::Refer(_) => Method::Refer,
Self::Bye(_) => Method::Bye,
Self::Cancel(_) => Method::Cancel,
Self::Notify(_) => Method::Notify,
Self::Subscribe(_) => Method::Subscribe,
Self::Info(_) => Method::Info,
Self::Update(_) => Method::Update,
Self::Message(_) => Method::Message,
Self::Options(_) => Method::Options,
}
}
}
fn state_machine_stage_for_event(event: &EventType) -> CleanupStage {
match event {
EventType::IncomingCall { .. } | EventType::IncomingCallAutoAccept { .. } => {
CleanupStage::StateMachineIncomingCall
}
EventType::AcceptCall => CleanupStage::StateMachineAcceptCall,
EventType::DialogBYE | EventType::DialogTerminated | EventType::DialogCANCEL => {
CleanupStage::StateMachineTerminalEvent
}
_ => CleanupStage::StateMachineOtherEvent,
}
}
fn state_machine_event_name(event: &EventType) -> &'static str {
match event {
EventType::IncomingCall { .. } => "IncomingCall",
EventType::IncomingCallAutoAccept { .. } => "IncomingCallAutoAccept",
EventType::AcceptCall => "AcceptCall",
EventType::DialogBYE => "DialogBYE",
EventType::DialogTerminated => "DialogTerminated",
EventType::InternalCheckReady => "InternalCheckReady",
EventType::DialogCreated { .. } => "DialogCreated",
EventType::Dialog200OK => "Dialog200OK",
EventType::DialogACK => "DialogACK",
EventType::DialogCANCEL => "DialogCANCEL",
_ => "Other",
}
}
#[derive(Debug, Clone)]
pub enum SessionEvent {
StateChanged {
session_id: SessionId,
old_state: CallState,
new_state: CallState,
},
MediaFlowEstablished {
session_id: SessionId,
local_addr: String,
remote_addr: String,
direction: crate::state_table::MediaFlowDirection,
},
CallEstablished {
session_id: SessionId,
},
CallTerminated {
session_id: SessionId,
},
CallCancelled {
session_id: SessionId,
},
CallOnHold {
session_id: SessionId,
},
CallResumed {
session_id: SessionId,
},
Custom {
session_id: SessionId,
event: String,
},
}
impl StateMachine {
pub fn new(
table: Arc<crate::state_table::MasterStateTable>,
store: Arc<SessionStore>,
dialog_adapter: Arc<DialogAdapter>,
media_adapter: Arc<MediaAdapter>,
) -> Self {
Self {
table,
store,
dialog_adapter,
media_adapter,
event_tx: None, auto_180_ringing: true,
}
}
pub fn new_with_adapters(
store: Arc<SessionStore>,
dialog_adapter: Arc<DialogAdapter>,
media_adapter: Arc<MediaAdapter>,
event_tx: tokio::sync::mpsc::Sender<SessionEvent>,
) -> Self {
Self {
table: MASTER_TABLE.clone(),
store,
dialog_adapter,
media_adapter,
event_tx: Some(event_tx),
auto_180_ringing: true,
}
}
pub fn new_with_custom_table(
table: Arc<crate::state_table::MasterStateTable>,
store: Arc<SessionStore>,
dialog_adapter: Arc<DialogAdapter>,
media_adapter: Arc<MediaAdapter>,
event_tx: tokio::sync::mpsc::Sender<SessionEvent>,
auto_180_ringing: bool,
) -> Self {
Self {
table,
store,
dialog_adapter,
media_adapter,
event_tx: Some(event_tx),
auto_180_ringing,
}
}
pub fn has_transition(&self, key: &StateKey) -> bool {
self.table.has_transition(key)
}
pub async fn stage_outbound_options(
&self,
session_id: &SessionId,
slot: PendingOptionsSlot,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
let mut session = self.store.get_session(session_id).await.map_err(|e| {
Box::<dyn std::error::Error + Send + Sync>::from(format!(
"stage_outbound_options: session {} not found: {}",
session_id, e
))
})?;
let method = slot.method();
let occupied = match &slot {
PendingOptionsSlot::Invite(_) => session.pending_invite_options.is_some(),
PendingOptionsSlot::ReInvite(_) => session.pending_reinvite_options.is_some(),
PendingOptionsSlot::Register(_) => session.pending_register_options.is_some(),
PendingOptionsSlot::Refer(_) => session.pending_refer_options.is_some(),
PendingOptionsSlot::Bye(_) => session.pending_bye_options.is_some(),
PendingOptionsSlot::Cancel(_) => session.pending_cancel_options.is_some(),
PendingOptionsSlot::Notify(_) => session.pending_notify_options.is_some(),
PendingOptionsSlot::Subscribe(_) => session.pending_subscribe_options.is_some(),
PendingOptionsSlot::Info(_) => session.pending_info_options.is_some(),
PendingOptionsSlot::Update(_) => session.pending_update_options.is_some(),
PendingOptionsSlot::Message(_) => session.pending_message_options.is_some(),
PendingOptionsSlot::Options(_) => session.pending_options_options.is_some(),
};
if occupied {
return Err(crate::errors::SessionError::Conflict { method }.into());
}
match slot {
PendingOptionsSlot::Invite(a) => session.pending_invite_options = Some(a),
PendingOptionsSlot::ReInvite(a) => session.pending_reinvite_options = Some(a),
PendingOptionsSlot::Register(a) => session.pending_register_options = Some(a),
PendingOptionsSlot::Refer(a) => session.pending_refer_options = Some(a),
PendingOptionsSlot::Bye(a) => session.pending_bye_options = Some(a),
PendingOptionsSlot::Cancel(a) => session.pending_cancel_options = Some(a),
PendingOptionsSlot::Notify(a) => session.pending_notify_options = Some(a),
PendingOptionsSlot::Subscribe(a) => session.pending_subscribe_options = Some(a),
PendingOptionsSlot::Info(a) => session.pending_info_options = Some(a),
PendingOptionsSlot::Update(a) => session.pending_update_options = Some(a),
PendingOptionsSlot::Message(a) => session.pending_message_options = Some(a),
PendingOptionsSlot::Options(a) => session.pending_options_options = Some(a),
}
self.store.update_session(session).await?;
Ok(())
}
pub async fn process_event(
&self,
session_id: &SessionId,
event: EventType,
) -> Result<ProcessEventResult, Box<dyn std::error::Error + Send + Sync>> {
let guard = cleanup_diag::stage_guard(
state_machine_stage_for_event(&event),
format!("{}:{}", session_id, state_machine_event_name(&event)),
);
let result = self.process_event_inner(session_id, event).await;
match &result {
Ok(_) => guard.finish_success(),
Err(_) => guard.finish_failure(),
}
result
}
async fn process_event_inner(
&self,
session_id: &SessionId,
event: EventType,
) -> Result<ProcessEventResult, Box<dyn std::error::Error + Send + Sync>> {
use std::collections::VecDeque;
const MAX_INTERNAL_EVENTS: usize = 32;
let mut queue = VecDeque::new();
queue.push_back(event);
let mut first_result = None;
let mut processed = 0usize;
while let Some(event) = queue.pop_front() {
processed += 1;
if processed > MAX_INTERNAL_EVENTS {
return Err(crate::errors::SessionError::InternalError(format!(
"state-machine internal event limit ({}) exceeded for session {}",
MAX_INTERNAL_EVENTS, session_id
))
.into());
}
let result = self
.process_one_event(session_id, event, &mut queue)
.await?;
if first_result.is_none() {
first_result = Some(result);
}
}
first_result.ok_or_else(|| {
crate::errors::SessionError::InternalError(format!(
"state-machine queue was empty for session {}",
session_id
))
.into()
})
}
async fn process_one_event(
&self,
session_id: &SessionId,
event: EventType,
queued_follow_up_events: &mut std::collections::VecDeque<EventType>,
) -> Result<ProcessEventResult, Box<dyn std::error::Error + Send + Sync>> {
use crate::session_store::{ActionRecord, GuardResult, TransitionRecord};
use std::time::Instant;
debug!("Processing event {:?} for session {}", event, session_id);
let transition_start = Instant::now();
let mut session = match self.store.get_session(session_id).await {
Ok(s) => s,
Err(e) => {
debug!("Failed to get session {}: {}", session_id, e);
return Err(
crate::errors::SessionError::SessionNotFound(session_id.to_string()).into(),
);
}
};
let old_state = session.call_state;
let mut guards_evaluated = Vec::new();
let mut actions_executed_history = Vec::new();
let mut errors = Vec::new();
match &event {
EventType::MakeCall { target } => {
session.remote_uri = Some(target.clone());
}
EventType::IncomingCall { from, sdp }
| EventType::IncomingCallAutoAccept { from, sdp } => {
session.remote_uri = Some(from.clone());
if let Some(sdp_data) = sdp {
session.remote_sdp = Some(sdp_data.clone());
}
}
EventType::RejectCall { status, reason } => {
session.reject_status = Some(*status);
session.reject_reason = Some(reason.clone());
}
EventType::RedirectCall { status, contacts } => {
session.redirect_response_status = Some(*status);
session.redirect_response_contacts = contacts.clone();
}
EventType::SendEarlyMedia { sdp } => {
if let Some(sdp_data) = sdp {
session.early_media_sdp = Some(sdp_data.clone());
}
}
EventType::AuthRequired {
status_code,
challenge,
method,
} => {
session.pending_auth = Some((*status_code, challenge.clone()));
session.pending_auth_method = if method.is_empty() {
None
} else {
Some(method.clone())
};
}
EventType::SessionIntervalTooSmall { min_se_secs } => {
session.session_timer_min_se = if *min_se_secs > 0 {
Some(*min_se_secs)
} else {
None
};
}
EventType::Dialog3xxRedirect { targets, .. } => {
for t in targets {
if !session.redirect_targets.contains(t) {
session.redirect_targets.push(t.clone());
}
}
}
EventType::TransferRequested {
refer_to,
transfer_type,
transaction_id,
} => {
session.transfer_target = Some(refer_to.clone());
session.transfer_notify_dialog = session.dialog_id.clone();
session.refer_transaction_id = Some(transaction_id.clone());
debug!(
"Set transfer target from REFER: {}, type: {:?}, transaction: {}",
refer_to, transfer_type, transaction_id
);
}
EventType::ReinviteReceived { sdp } => {
if session.call_state == crate::types::CallState::Active
&& session.pending_reinvite.is_some()
{
info!(
"RFC 3261 §14.1 UAS-side glare: peer re-INVITE arrived while \
our builder-API re-INVITE is in flight on session {} — \
responding 491 Request Pending",
session.session_id
);
if let Err(e) = self
.dialog_adapter
.send_response(&session.session_id, 491, None)
.await
{
tracing::warn!(
"Failed to send 491 Request Pending for session {}: {}",
session.session_id,
e
);
}
return Ok(ProcessEventResult {
old_state,
next_state: None,
transition: None,
actions_executed: vec![],
events_published: vec![],
});
}
if let Some(sdp_data) = sdp {
session.remote_sdp = Some(sdp_data.clone());
session.sdp_negotiated = false;
}
}
EventType::UpdateReceived { sdp } => {
if let Some(sdp_data) = sdp {
session.remote_sdp = Some(sdp_data.clone());
session.sdp_negotiated = false;
}
}
_ => {}
}
let key = StateKey {
role: session.role,
state: session.call_state,
event: event.clone(),
};
let transition = match self.table.get(&key) {
Some(t) => t,
None => {
debug!("No transition defined for {:?}", key);
if session.history.is_some() {
let now = Instant::now();
let record = TransitionRecord {
sequence: 0, timestamp: now,
timestamp_ms: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64,
from_state: old_state,
event: event.clone(),
to_state: Some(old_state),
guards_evaluated: vec![],
actions_executed: vec![],
duration_ms: transition_start.elapsed().as_millis() as u64,
errors: vec![format!("No transition defined for {:?}", key)],
events_published: vec![],
};
session.record_transition(record);
self.store.update_session(session).await?;
}
return Ok(ProcessEventResult {
old_state,
next_state: None,
transition: None,
actions_executed: vec![],
events_published: vec![],
});
}
};
for guard in &transition.guards {
let guard_start = Instant::now();
let satisfied = guards::check_guard(guard, &session).await;
let guard_duration = guard_start.elapsed().as_millis() as u64;
guards_evaluated.push(GuardResult {
guard: guard.clone(),
passed: satisfied,
evaluation_time_us: guard_duration * 1000,
});
if !satisfied {
debug!("Guard {:?} not satisfied, skipping transition", guard);
if session.history.is_some() {
let now = Instant::now();
let record = TransitionRecord {
sequence: 0,
timestamp: now,
timestamp_ms: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64,
from_state: old_state,
event: event.clone(),
to_state: Some(old_state),
guards_evaluated,
actions_executed: vec![],
duration_ms: transition_start.elapsed().as_millis() as u64,
errors: vec![format!("Guard {:?} not satisfied", guard)],
events_published: vec![],
};
session.record_transition(record);
self.store.update_session(session).await?;
}
return Ok(ProcessEventResult {
old_state,
next_state: None,
transition: None,
actions_executed: vec![],
events_published: vec![],
});
}
}
info!("Executing transition for {:?} + {:?}", old_state, event);
if let Some(new_state) = transition.next_state {
info!("State transition: {:?} -> {:?}", old_state, new_state);
session.call_state = new_state;
session.entered_state_at = Instant::now();
if new_state == crate::types::CallState::Terminated {
session.pending_invite_options = None;
session.pending_reinvite_options = None;
session.pending_register_options = None;
session.pending_refer_options = None;
session.pending_bye_options = None;
session.pending_cancel_options = None;
session.pending_notify_options = None;
session.pending_subscribe_options = None;
session.pending_info_options = None;
session.pending_update_options = None;
session.pending_message_options = None;
session.pending_options_options = None;
}
self.store.update_session(session.clone()).await?;
}
let mut actions_executed = Vec::new();
for action in &transition.actions {
if self.should_skip_action(action) {
continue;
}
let action_start = Instant::now();
let result = Box::pin(actions::execute_action(
action,
&mut session,
&self.dialog_adapter,
&self.media_adapter,
&self.store,
self.auto_180_ringing,
&None, ))
.await;
let action_duration = action_start.elapsed().as_millis() as u64;
let (success, error_opt, exec_error) = match result {
Ok(outcome) => {
actions_executed.push(action.clone());
queued_follow_up_events.extend(outcome.follow_up_events);
(true, None, None)
}
Err(e) => {
let error_msg = format!("Failed to execute action {:?}: {}", action, e);
error!("{}", error_msg);
errors.push(error_msg.clone());
(false, Some(error_msg), Some(e))
}
};
actions_executed_history.push(ActionRecord {
action: action.clone(),
success,
execution_time_us: action_duration * 1000,
error: error_opt,
});
if !success {
if session.history.is_some() {
let now = Instant::now();
let record = TransitionRecord {
sequence: 0,
timestamp: now,
timestamp_ms: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64,
from_state: old_state,
event: event.clone(),
to_state: Some(old_state),
guards_evaluated,
actions_executed: actions_executed_history,
duration_ms: transition_start.elapsed().as_millis() as u64,
errors,
events_published: vec![],
};
session.record_transition(record);
self.store.update_session(session).await?;
}
return Err(exec_error.unwrap());
}
}
let next_state = transition.next_state;
if session.history.is_some() {
let now = Instant::now();
let record = TransitionRecord {
sequence: 0,
timestamp: now,
timestamp_ms: std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64,
from_state: old_state,
event: event.clone(),
to_state: next_state,
guards_evaluated,
actions_executed: actions_executed_history,
duration_ms: transition_start.elapsed().as_millis() as u64,
errors,
events_published: transition.publish_events.clone(),
};
session.record_transition(record);
}
session.apply_condition_updates(&transition.condition_updates);
if let Ok(current) = self.store.get_session(session_id).await {
if current.call_state != session.call_state {
debug!(
"session {} call_state changed during action phase ({:?} -> {:?}); preserving store value",
session_id, session.call_state, current.call_state
);
session.call_state = current.call_state;
session.entered_state_at = current.entered_state_at;
}
if current.sdp_origin_version > session.sdp_origin_version {
session.sdp_origin_session_id = current.sdp_origin_session_id;
session.sdp_origin_version = current.sdp_origin_version;
}
if session.media_security.is_none() {
session.media_security = current.media_security;
}
}
self.store.update_session(session.clone()).await?;
if let Some(ref event_tx) = self.event_tx {
for event_template in &transition.publish_events {
let event = self
.instantiate_event(event_template, &session, old_state)
.await;
let guard = cleanup_diag::stage_guard(
CleanupStage::StateMachineEventPublish,
&session.session_id.0,
);
match event_tx.send(event).await {
Ok(()) => guard.finish_success(),
Err(e) => {
guard.finish_failure();
error!("Failed to publish event: {}", e);
}
}
}
}
let session = self
.store
.get_session(session_id)
.await
.map_err(|e| format!("Failed to reload session after actions: {}", e))?;
let all_conditions_met = session.all_conditions_met();
let call_established_triggered = session.call_established_triggered;
self.store
.update_session(session)
.await
.map_err(|e| format!("Failed to save session state: {}", e))?;
if all_conditions_met && !call_established_triggered {
debug!("All conditions met, triggering InternalCheckReady");
queued_follow_up_events.push_back(EventType::InternalCheckReady);
}
Ok(ProcessEventResult {
old_state,
next_state: transition.next_state,
transition: Some(transition.clone()),
actions_executed,
events_published: transition.publish_events.clone(),
})
}
fn should_skip_action(&self, action: &Action) -> bool {
matches!(action, Action::SendSIPResponse(180, _)) && !self.auto_180_ringing
}
async fn instantiate_event(
&self,
template: &EventTemplate,
session: &SessionState,
old_state: CallState,
) -> SessionEvent {
match template {
EventTemplate::StateChanged => SessionEvent::StateChanged {
session_id: session.session_id.clone(),
old_state,
new_state: session.call_state,
},
EventTemplate::MediaFlowEstablished => {
let negotiated = session.negotiated_config.as_ref();
SessionEvent::MediaFlowEstablished {
session_id: session.session_id.clone(),
local_addr: negotiated
.map(|n| n.local_addr.to_string())
.unwrap_or_default(),
remote_addr: negotiated
.map(|n| n.remote_addr.to_string())
.unwrap_or_default(),
direction: crate::state_table::MediaFlowDirection::Both,
}
}
EventTemplate::CallEstablished => SessionEvent::CallEstablished {
session_id: session.session_id.clone(),
},
EventTemplate::CallTerminated => SessionEvent::CallTerminated {
session_id: session.session_id.clone(),
},
EventTemplate::CallCancelled => SessionEvent::CallCancelled {
session_id: session.session_id.clone(),
},
EventTemplate::CallOnHold => SessionEvent::CallOnHold {
session_id: session.session_id.clone(),
},
EventTemplate::CallResumed => SessionEvent::CallResumed {
session_id: session.session_id.clone(),
},
EventTemplate::Custom(event) => SessionEvent::Custom {
session_id: session.session_id.clone(),
event: event.clone(),
},
_ => SessionEvent::Custom {
session_id: session.session_id.clone(),
event: format!("{:?}", template),
},
}
}
}