//! Transport-neutral dispatch from protocol requests to image API operations. use std::num::NonZeroUsize; use runa_actor::{ActorId, Event, Message, Status}; use runa_image_api::ImageApi; use runa_protocol::{ ActorSnapshotDto, ActorStatusDto, BehaviorDto, BehaviorVersionDto, Command, ErrorCode, EventDto, ExecutionSnapshotDto, MessageDto, PROTOCOL_VERSION, ProtocolError, Reply, Request, Response, }; pub struct Server { image: ImageApi, } impl Default for Server { fn default() -> Self { Self::new() } } impl Server { pub fn new() -> Self { Self { image: ImageApi::new(), } } pub fn with_image_api(image: ImageApi) -> Self { Self { image } } pub fn dispatch(&mut self, request: Request) -> Response { let request_id = request.request_id.clone(); let result = request .validate() .and_then(|()| self.dispatch_command(request.command)); Response { version: PROTOCOL_VERSION, request_id, result, } } fn dispatch_command(&mut self, command: Command) -> Result { match command { Command::PublishBehavior { name, expected_revision, source, } => self .image .publish_behavior(&name, expected_revision, &source) .map(behavior_dto) .map(Reply::Behavior) .map_err(api_error), Command::RollbackBehavior { name, expected_revision, target_revision, } => self .image .rollback_behavior(&name, expected_revision, target_revision) .map(behavior_dto) .map(Reply::Behavior) .map_err(api_error), Command::GetBehavior { name } => self .image .behavior(&name) .map(behavior_dto) .map(Reply::Behavior) .ok_or_else(|| ProtocolError::new(ErrorCode::NotFound, "unknown behavior")), Command::GetBehaviors => Ok(Reply::Behaviors( self.image .behaviors() .into_iter() .map(behavior_dto) .collect(), )), Command::GetBehaviorSource { name, revision } => self .image .behavior_source(&name, revision) .map(|source| Reply::BehaviorSource { source: source.to_owned(), }) .map_err(api_error), Command::Spawn { behavior_name, state, mailbox_capacity, } => { let capacity = NonZeroUsize::new(mailbox_capacity) .ok_or_else(|| invalid_request("mailbox_capacity must be positive"))?; self.image .spawn(&behavior_name, state, capacity) .map(|actor_id| Reply::ActorSpawned { actor_id: actor_id.0, }) .map_err(api_error) } Command::Send { actor_id, message } => self .image .send(ActorId(actor_id), message_from_dto(message)) .map(|()| Reply::Acknowledged) .map_err(api_error), Command::Tick { ticks, instruction_budget, } => { let budget = NonZeroUsize::new(instruction_budget) .ok_or_else(|| invalid_request("instruction_budget must be positive"))?; self.image.tick(ticks, budget); Ok(Reply::TickCompleted { ticks }) } Command::InspectActor { actor_id } => self .image .inspect(ActorId(actor_id)) .map(|snapshot| actor_snapshot_dto(&snapshot)) .map(Reply::Actor) .map_err(api_error), Command::ListActors => Ok(Reply::Actors( self.image.actors().iter().map(actor_snapshot_dto).collect(), )), Command::GetEvents => Ok(Reply::Events( self.image.events().iter().map(event_dto).collect(), )), Command::StopActor { actor_id } => self .image .stop(ActorId(actor_id)) .map(|()| Reply::Acknowledged) .map_err(api_error), Command::RestartActor { actor_id } => self .image .restart(ActorId(actor_id)) .map(|()| Reply::Acknowledged) .map_err(api_error), Command::PauseActor { actor_id } => self .image .pause(ActorId(actor_id)) .map(|()| Reply::Acknowledged) .map_err(api_error), Command::ResumeActor { actor_id } => self .image .resume(ActorId(actor_id)) .map(|()| Reply::Acknowledged) .map_err(api_error), Command::CancelActor { actor_id } => self .image .cancel(ActorId(actor_id)) .map(|()| Reply::Acknowledged) .map_err(api_error), Command::InstallBehavior { actor_id, behavior_name, } => self .image .install(ActorId(actor_id), &behavior_name) .map(|implementation_id| Reply::BehaviorInstalled { implementation_id: implementation_id.to_string(), }) .map_err(api_error), } } } fn behavior_dto(behavior: runa_code::BehaviorSnapshot) -> BehaviorDto { BehaviorDto { name: behavior.name, revision: behavior.revision, current_implementation_id: behavior.current.to_string(), history: behavior .history .into_iter() .map(|version| BehaviorVersionDto { revision: version.revision, implementation_id: version.implementation_id.to_string(), }) .collect(), } } fn actor_snapshot_dto(snapshot: &runa_actor::ActorSnapshot) -> ActorSnapshotDto { ActorSnapshotDto { actor_id: snapshot.id.0, mailbox_id: snapshot.mailbox.0, implementation_id: snapshot.implementation.to_string(), incarnation: snapshot.incarnation, state: snapshot.state, status: match &snapshot.status { Status::Idle => ActorStatusDto::Idle, Status::Running => ActorStatusDto::Running, Status::Awaiting => ActorStatusDto::Awaiting, Status::Paused => ActorStatusDto::Paused, Status::Stopped => ActorStatusDto::Stopped, Status::Failed(message) => ActorStatusDto::Failed(message.clone()), }, execution: snapshot.execution.map(|execution| ExecutionSnapshotDto { vm_steps: execution.vm_steps, instruction_pointer: execution.instruction_pointer, call_depth: execution.call_depth, }), queued: snapshot.queued.iter().copied().map(message_dto).collect(), reserved: snapshot.reserved.map(message_dto), } } fn message_dto(message: Message) -> MessageDto { MessageDto { kind: message.kind, value: message.value, } } fn message_from_dto(message: MessageDto) -> Message { Message { kind: message.kind, value: message.value, } } fn event_dto(event: &Event) -> EventDto { match event { Event::Spawned(actor_id) => EventDto::Spawned { actor_id: actor_id.0, }, Event::Committed(actor_id) => EventDto::Committed { actor_id: actor_id.0, }, Event::Failed(actor_id, message) => EventDto::Failed { actor_id: actor_id.0, message: message.clone(), }, Event::Stopped(actor_id) => EventDto::Stopped { actor_id: actor_id.0, }, Event::Restarted(actor_id) => EventDto::Restarted { actor_id: actor_id.0, }, Event::Paused(actor_id) => EventDto::Paused { actor_id: actor_id.0, }, Event::Resumed(actor_id) => EventDto::Resumed { actor_id: actor_id.0, }, Event::InvocationCancelled(actor_id) => EventDto::InvocationCancelled { actor_id: actor_id.0, }, Event::BehaviorInstalled(actor_id, implementation_id) => EventDto::BehaviorInstalled { actor_id: actor_id.0, implementation_id: implementation_id.to_string(), }, Event::StateMigrated(actor_id, previous_state, current_state) => EventDto::StateMigrated { actor_id: actor_id.0, previous_state: *previous_state, current_state: *current_state, }, } } fn api_error(error: impl std::fmt::Display) -> ProtocolError { let message = error.to_string(); let code = if message.starts_with("behavior revision conflict:") { ErrorCode::Conflict } else if matches!( message.as_str(), "unknown actor" | "unknown behavior" | "unknown behavior revision" | "unknown target behavior revision" | "unknown implementation" ) { ErrorCode::NotFound } else if matches!( message.as_str(), "mailbox full" | "effect exceeds mailbox capacity" ) { ErrorCode::CapacityExceeded } else if message == "invalid behavior name" { ErrorCode::InvalidRequest } else { ErrorCode::RuntimeFailure }; ProtocolError::new(code, message) } fn invalid_request(message: impl Into) -> ProtocolError { ProtocolError::new(ErrorCode::InvalidRequest, message) } #[cfg(test)] mod tests { use super::*; use runa_protocol::{Command, PROTOCOL_VERSION}; fn request(command: Command) -> Request { Request { version: PROTOCOL_VERSION, request_id: "test-request".to_owned(), command, } } fn publish(server: &mut Server) -> BehaviorDto { let response = server.dispatch(request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: None, source: "pub fn handle(state, kind, value) { (state + value, []) }".to_owned(), })); match response.result.unwrap() { Reply::Behavior(behavior) => behavior, other => panic!("unexpected reply: {other:?}"), } } #[test] fn dispatches_behavior_actor_and_event_operations() { let mut server = Server::new(); let behavior = publish(&mut server); assert_eq!(behavior.revision, 1); assert_eq!(behavior.history.len(), 1); let spawned = server.dispatch(request(Command::Spawn { behavior_name: "counter".to_owned(), state: 5, mailbox_capacity: 2, })); let actor_id = match spawned.result.unwrap() { Reply::ActorSpawned { actor_id } => actor_id, other => panic!("unexpected reply: {other:?}"), }; assert_eq!(actor_id, 1); let sent = server.dispatch(request(Command::Send { actor_id, message: MessageDto { kind: 0, value: 7 }, })); assert_eq!(sent.result, Ok(Reply::Acknowledged)); server.dispatch(request(Command::Tick { ticks: 100, instruction_budget: 10, })); let inspected = server.dispatch(request(Command::InspectActor { actor_id })); match inspected.result.unwrap() { Reply::Actor(snapshot) => assert_eq!(snapshot.state, 12), other => panic!("unexpected reply: {other:?}"), } let events = server.dispatch(request(Command::GetEvents)); match events.result.unwrap() { Reply::Events(events) => assert!(events.iter().any( |event| matches!(event, EventDto::Committed { actor_id: id } if *id == actor_id) )), other => panic!("unexpected reply: {other:?}"), } assert_eq!( server.dispatch(request(Command::GetEvents)).result, Ok(Reply::Events(Vec::new())) ); } #[test] fn dispatch_preserves_request_id_and_maps_validation_and_domain_errors() { let mut server = Server::new(); let mut invalid = request(Command::GetEvents); invalid.version += 1; let response = server.dispatch(invalid); assert_eq!(response.request_id, "test-request"); assert_eq!( response.result.unwrap_err().code, ErrorCode::UnsupportedVersion ); let missing = server.dispatch(request(Command::InspectActor { actor_id: 42 })); assert_eq!(missing.result.unwrap_err().code, ErrorCode::NotFound); publish(&mut server); let conflict = server.dispatch(request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: Some(99), source: "pub fn handle(state, kind, value) { (state, []) }".to_owned(), })); assert_eq!(conflict.result.unwrap_err().code, ErrorCode::Conflict); } #[test] fn dispatch_supports_behavior_history_source_and_safe_install() { let mut server = Server::new(); let first = publish(&mut server); let updated = server.dispatch(request(Command::PublishBehavior { name: "counter".to_owned(), expected_revision: Some(first.revision), source: "pub fn handle(state, kind, value) { (state * value, []) }".to_owned(), })); assert!(matches!(updated.result, Ok(Reply::Behavior(_)))); let source = server.dispatch(request(Command::GetBehaviorSource { name: "counter".to_owned(), revision: 1, })); assert_eq!( source.result, Ok(Reply::BehaviorSource { source: "pub fn handle(state, kind, value) { (state + value, []) }".to_owned() }) ); let actor_id = match server .dispatch(request(Command::Spawn { behavior_name: "counter".to_owned(), state: 3, mailbox_capacity: 1, })) .result .unwrap() { Reply::ActorSpawned { actor_id } => actor_id, other => panic!("unexpected reply: {other:?}"), }; let installed = server.dispatch(request(Command::InstallBehavior { actor_id, behavior_name: "counter".to_owned(), })); assert!(matches!( installed.result, Ok(Reply::BehaviorInstalled { .. }) )); } }