use super::RuntimeController;
use crate::audit::AuditOutcome;
use crate::command::CommandRegistry;
use crate::context::RuntimeContext;
use crate::decision::DecisionRegistry;
use crate::envelope::{CommandEnvelope, EventEnvelope};
use crate::error::RuntimeResult;
use crate::event::EventRegistry;
use crate::handler::{CommandHandler, CommandResult};
use crate::ids::{
AppFamily, AppId, CommandName, CoreId, EventName, NodeId, RuntimeContractVersion, SyncGroup,
TenantId,
};
use crate::lifecycle::{RuntimeLifecycleEvent, RuntimeLifecycleState};
use crate::plugin::AppPlugin;
use crate::state::StateRegistry;
use crate::{RuntimeBuilder, RuntimeIdentity, TraceContext};
use appcore_contracts::{ApplicationId, ApplicationManifestV1, RuntimeRequirements, ServiceId};
use parking_lot::{Condvar, Mutex};
use std::sync::{Arc, Barrier};
use std::thread;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
fn assert_send_sync<T: Send + Sync>() {}
struct StaticContext {
app_id: AppId,
app_family: AppFamily,
sync_group: SyncGroup,
runtime_contract: RuntimeContractVersion,
node_id: NodeId,
}
impl RuntimeContext for StaticContext {
fn app_id(&self) -> &AppId {
&self.app_id
}
fn app_family(&self) -> &AppFamily {
&self.app_family
}
fn sync_group(&self) -> &SyncGroup {
&self.sync_group
}
fn runtime_contract(&self) -> RuntimeContractVersion {
self.runtime_contract
}
fn node_id(&self) -> &NodeId {
&self.node_id
}
}
struct StartHandler;
impl CommandHandler for StartHandler {
fn command_name(&self) -> CommandName {
CommandName::new("runtime.start".to_string()).unwrap()
}
fn handle(
&self,
_command: &CommandEnvelope,
_context: &dyn RuntimeContext,
) -> RuntimeResult<CommandResult> {
let event = EventEnvelope::new(
EventName::new("RuntimeStarted".to_string()).unwrap(),
"evt-1".to_string(),
AppId::new("example-app".to_string()).unwrap(),
NodeId::new("node-a".to_string()).unwrap(),
0,
vec![],
)?;
Ok(CommandResult::accepted(vec![event]))
}
}
struct ControllerPlugin;
impl AppPlugin for ControllerPlugin {
fn application_manifest(&self) -> ApplicationManifestV1 {
ApplicationManifestV1::new(
ApplicationId::new("example-app").unwrap(),
"1.0.0",
"Example App",
"Example Vendor",
ServiceId::new("example-service").unwrap(),
RuntimeRequirements::new("1.0.0", "1").unwrap(),
)
.unwrap()
}
fn identity(&self, node_id: NodeId) -> RuntimeIdentity {
RuntimeIdentity {
app_id: AppId::new("example-app").unwrap(),
app_family: AppFamily::new("example-family").unwrap(),
sync_group: SyncGroup::new("dev").unwrap(),
runtime_contract: RuntimeContractVersion::new(1),
node_id,
}
}
fn register_commands(&self, _registry: &mut CommandRegistry) -> RuntimeResult<()> {
Ok(())
}
fn register_events(&self, _registry: &mut EventRegistry) -> RuntimeResult<()> {
Ok(())
}
fn register_states(&self, _registry: &mut StateRegistry) -> RuntimeResult<()> {
Ok(())
}
fn register_decisions(&self, _registry: &mut DecisionRegistry) -> RuntimeResult<()> {
Ok(())
}
fn register_handlers(&self, bus: &mut crate::CommandBus) -> RuntimeResult<()> {
bus.register_handler(StartHandler)
}
}
fn build_instance() -> RuntimeResult<crate::runtime::RuntimeInstance> {
let plugin = ControllerPlugin;
let mut builder = RuntimeBuilder::new();
builder.with_plugin(&plugin, NodeId::new("node-a".to_string()).unwrap())?;
builder.build()
}
#[test]
fn new_keeps_instance() {
assert_send_sync::<RuntimeController>();
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let controller = RuntimeController::new(instance);
assert_eq!(
controller.instance().application_manifest().display_name(),
"Example App"
);
}
#[test]
fn exposes_lifecycle() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let controller = RuntimeController::new(instance);
assert_eq!(
controller.lifecycle().current(),
RuntimeLifecycleState::Booting
);
}
#[test]
fn apply_lifecycle_event_changes_state() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let controller = RuntimeController::new(instance);
let result = controller.apply_lifecycle_event(RuntimeLifecycleEvent::ConfigLoaded);
assert!(result.is_ok());
assert_eq!(
controller.lifecycle().current(),
RuntimeLifecycleState::CheckingSecurity
);
}
#[test]
fn invalid_transition_returns_error() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let controller = RuntimeController::new(instance);
let result = controller.apply_lifecycle_event(RuntimeLifecycleEvent::ApiStarted);
assert_eq!(result, Err(crate::RuntimeError::InvalidStateTransition));
assert_eq!(
controller.lifecycle().current(),
RuntimeLifecycleState::Booting
);
}
fn make_context() -> StaticContext {
StaticContext {
app_id: AppId::new("example-app".to_string()).unwrap(),
app_family: AppFamily::new("example-family".to_string()).unwrap(),
sync_group: SyncGroup::new("dev".to_string()).unwrap(),
runtime_contract: RuntimeContractVersion::new(1),
node_id: NodeId::new("node-a".to_string()).unwrap(),
}
}
fn make_command() -> RuntimeResult<CommandEnvelope> {
CommandEnvelope::new(
CommandName::new("runtime.start".to_string()).unwrap(),
"cmd-1".to_string(),
AppId::new("example-app".to_string()).unwrap(),
NodeId::new("node-a".to_string()).unwrap(),
0,
None,
vec![],
)
}
fn make_command_with_key(key: Option<&str>) -> RuntimeResult<CommandEnvelope> {
CommandEnvelope::new(
CommandName::new("runtime.start".to_string()).unwrap(),
"cmd-1".to_string(),
AppId::new("example-app".to_string()).unwrap(),
NodeId::new("node-a".to_string()).unwrap(),
0,
key.map(|value| value.to_string()),
vec![],
)
}
fn move_to_running(controller: &mut RuntimeController) {
let _ = controller.apply_lifecycle_event(RuntimeLifecycleEvent::ConfigLoaded);
let _ = controller.apply_lifecycle_event(RuntimeLifecycleEvent::SecurityChecked);
let _ = controller.apply_lifecycle_event(RuntimeLifecycleEvent::StorageOpened);
let _ = controller.apply_lifecycle_event(RuntimeLifecycleEvent::ApiStarted);
}
#[test]
fn controller_blocks_dispatch_in_booting() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let controller = RuntimeController::new(instance);
let context = make_context();
let command = make_command();
assert!(command.is_ok());
let command = match command {
Ok(command) => command,
Err(_) => return,
};
let result = controller.dispatch_command(&command, &context);
assert!(result.is_ok());
let result = match result {
Ok(result) => result,
Err(_) => return,
};
assert!(!result.is_accepted());
assert_eq!(result.message(), Some("runtime is not ready"));
}
#[test]
fn controller_allows_dispatch_in_running() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
let context = make_context();
let command = make_command();
assert!(command.is_ok());
let command = match command {
Ok(command) => command,
Err(_) => return,
};
let result = controller.dispatch_command(&command, &context);
assert!(result.is_ok());
let result = match result {
Ok(result) => result,
Err(_) => return,
};
assert!(result.is_accepted());
assert_eq!(controller.instance().event_bus().len(), 1);
assert_eq!(controller.instance().audit_log().len(), 1);
}
#[test]
fn controller_propagates_trace_to_events_and_audit() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
let context = make_context();
let trace = TraceContext::new(
"trace-1",
"span-1",
CoreId::new("core-a").unwrap(),
CoreId::new("core-a").unwrap(),
TenantId::new("tenant-a").unwrap(),
)
.unwrap()
.with_command_id("cmd-1")
.unwrap();
let command = make_command().unwrap().with_trace(trace.clone());
let result = controller.dispatch_command(&command, &context);
assert!(result.is_ok());
let events = controller.instance().event_bus().events();
let audit = controller.instance().audit_log().records();
assert_eq!(
events[0].trace.as_ref().map(|item| item.trace_id.as_str()),
Some("trace-1")
);
assert_eq!(
audit[0].trace.as_ref().map(|item| item.trace_id.as_str()),
Some("trace-1")
);
}
#[test]
fn controller_allows_dispatch_in_degraded() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
assert!(controller
.apply_lifecycle_event(RuntimeLifecycleEvent::DegradedDetected)
.is_ok());
let context = make_context();
let command = make_command();
assert!(command.is_ok());
let command = match command {
Ok(command) => command,
Err(_) => return,
};
let result = controller.dispatch_command(&command, &context);
assert!(result.is_ok());
let result = match result {
Ok(result) => result,
Err(_) => return,
};
assert!(result.is_accepted());
assert_eq!(controller.instance().event_bus().len(), 1);
assert_eq!(controller.instance().audit_log().len(), 1);
}
#[test]
fn controller_blocks_dispatch_in_restricted() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
assert!(controller
.apply_lifecycle_event(RuntimeLifecycleEvent::RestrictedDetected)
.is_ok());
let context = make_context();
let command = make_command();
assert!(command.is_ok());
let command = match command {
Ok(command) => command,
Err(_) => return,
};
let result = controller.dispatch_command(&command, &context);
assert!(result.is_ok());
let result = match result {
Ok(result) => result,
Err(_) => return,
};
assert!(!result.is_accepted());
assert_eq!(result.message(), Some("runtime is restricted"));
assert!(controller.instance().event_bus().is_empty());
assert_eq!(controller.instance().audit_log().len(), 1);
assert_eq!(
controller.instance().audit_log().records()[0].outcome,
AuditOutcome::Rejected
);
}
#[test]
fn controller_blocks_dispatch_in_shutting_down() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
assert!(controller
.apply_lifecycle_event(RuntimeLifecycleEvent::ShutdownRequested)
.is_ok());
let context = make_context();
let command = make_command();
assert!(command.is_ok());
let command = match command {
Ok(command) => command,
Err(_) => return,
};
let result = controller.dispatch_command(&command, &context);
assert!(result.is_ok());
let result = match result {
Ok(result) => result,
Err(_) => return,
};
assert!(!result.is_accepted());
assert_eq!(result.message(), Some("runtime is not ready"));
assert!(controller.instance().event_bus().is_empty());
assert_eq!(controller.instance().audit_log().len(), 1);
}
#[test]
fn runtime_instance_still_dispatches_without_lifecycle_check() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let context = make_context();
let command = make_command();
assert!(command.is_ok());
let command = match command {
Ok(command) => command,
Err(_) => return,
};
let result = instance.dispatch_command(&command, &context);
assert!(result.is_ok());
let result = match result {
Ok(result) => result,
Err(_) => return,
};
assert!(result.is_accepted());
}
#[test]
fn dispatch_rejected_does_not_emit_events() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
assert!(controller
.apply_lifecycle_event(RuntimeLifecycleEvent::RestrictedDetected)
.is_ok());
let context = make_context();
let command = make_command();
assert!(command.is_ok());
let command = match command {
Ok(command) => command,
Err(_) => return,
};
let result = controller.dispatch_command(&command, &context);
assert!(result.is_ok());
assert!(controller.instance().event_bus().is_empty());
assert_eq!(controller.instance().audit_log().len(), 1);
}
#[test]
fn runtime_instance_direct_dispatch_does_not_emit_to_event_bus() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let context = make_context();
let command = make_command();
assert!(command.is_ok());
let command = match command {
Ok(command) => command,
Err(_) => return,
};
let result = instance.dispatch_command(&command, &context);
assert!(result.is_ok());
assert!(instance.event_bus().is_empty());
assert!(instance.audit_log().is_empty());
}
#[test]
fn controller_audits_error_when_handler_missing() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
let context = make_context();
let command = CommandEnvelope::new(
CommandName::new("runtime.unknown".to_string()).unwrap(),
"cmd-missing".to_string(),
AppId::new("example-app".to_string()).unwrap(),
NodeId::new("node-a".to_string()).unwrap(),
10,
None,
vec![],
);
assert!(command.is_ok());
let command = match command {
Ok(command) => command,
Err(_) => return,
};
let result = controller.dispatch_command(&command, &context);
assert!(result.is_err());
assert!(controller.instance().event_bus().is_empty());
assert_eq!(controller.instance().audit_log().len(), 1);
assert_eq!(
controller.instance().audit_log().records()[0].outcome,
AuditOutcome::Error
);
assert_eq!(controller.idempotency_len(), 0);
}
#[test]
fn handler_error_does_not_poison_idempotency_retries() {
let instance = build_instance().unwrap();
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
let context = make_context();
let command = CommandEnvelope::new(
CommandName::new("runtime.unknown").unwrap(),
"cmd-retryable-error".to_string(),
AppId::new("example-app").unwrap(),
NodeId::new("node-a").unwrap(),
10,
Some("error-retry-key".to_string()),
Vec::new(),
)
.unwrap();
assert!(controller.dispatch_command(&command, &context).is_err());
assert_eq!(controller.idempotency_len(), 0);
assert!(controller.dispatch_command(&command, &context).is_err());
assert_eq!(controller.idempotency_len(), 0);
assert_eq!(controller.instance().audit_log().len(), 2);
}
#[test]
fn lifecycle_gate_runs_before_idempotency_check() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
assert!(controller
.apply_lifecycle_event(RuntimeLifecycleEvent::RestrictedDetected)
.is_ok());
let context = make_context();
let command = CommandEnvelope {
command_name: CommandName::new("runtime.start").unwrap(),
command_id: "cmd-1".to_string(),
app_id: AppId::new("example-app").unwrap(),
node_id: NodeId::new("node-a").unwrap(),
issued_at_ms: 0,
idempotency_key: Some("../bad".to_string()),
payload: vec![],
trace: None,
};
let result = controller.dispatch_command(&command, &context);
assert!(result.is_ok());
let result = match result {
Ok(result) => result,
Err(_) => return,
};
assert_eq!(result.message(), Some("runtime is restricted"));
assert_eq!(controller.idempotency_len(), 0);
}
#[test]
fn command_without_idempotency_key_executes_every_time() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
let context = make_context();
let first = make_command_with_key(None);
let second = make_command_with_key(None);
assert!(first.is_ok());
assert!(second.is_ok());
let first = match first {
Ok(command) => command,
Err(_) => return,
};
let second = match second {
Ok(command) => command,
Err(_) => return,
};
assert!(controller.dispatch_command(&first, &context).is_ok());
assert!(controller.dispatch_command(&second, &context).is_ok());
assert_eq!(controller.instance().event_bus().len(), 2);
assert_eq!(controller.idempotency_len(), 0);
}
#[test]
fn command_with_idempotency_key_executes_once_and_rejects_duplicate() {
let instance = build_instance();
assert!(instance.is_ok());
let instance = match instance {
Ok(instance) => instance,
Err(_) => return,
};
let mut controller = RuntimeController::new(instance);
move_to_running(&mut controller);
let context = make_context();
let first = make_command_with_key(Some("k-1"));
let second = make_command_with_key(Some("k-1"));
assert!(first.is_ok());
assert!(second.is_ok());
let first = match first {
Ok(command) => command,
Err(_) => return,
};
let second = match second {
Ok(command) => command,
Err(_) => return,
};
let first_result = controller.dispatch_command(&first, &context);
assert!(first_result.is_ok());
let second_result = controller.dispatch_command(&second, &context);
assert!(second_result.is_ok());
let second_result = match second_result {
Ok(result) => result,
Err(_) => return,
};
assert!(second_result.is_accepted());
assert_eq!(second_result.message(), None);
assert_eq!(controller.instance().event_bus().len(), 1);
assert_eq!(controller.instance().audit_log().len(), 2);
assert_eq!(
controller.instance().audit_log().records()[1].outcome,
AuditOutcome::Accepted
);
assert_eq!(controller.idempotency_len(), 1);
assert!(controller.idempotency_contains("k-1").unwrap_or(false));
}
#[test]
fn controller_deduplicates_across_reloads_via_file_idempotency() {
let file = temp_idempotency_file("reloads");
let first_instance = match build_instance() {
Ok(instance) => instance,
Err(_) => return,
};
let first_store = crate::FileIdempotencyStore::new(&file);
assert!(first_store.is_ok());
let mut first_controller = RuntimeController::with_idempotency_store(
first_instance,
Box::new(match first_store {
Ok(store) => store,
Err(_) => return,
}),
);
move_to_running(&mut first_controller);
let context = make_context();
let first_cmd = match make_command_with_key(Some("k-1")) {
Ok(cmd) => cmd,
Err(_) => return,
};
let second_cmd = first_cmd.clone();
let first_result = first_controller.dispatch_command(&first_cmd, &context);
assert!(first_result.is_ok());
drop(first_controller);
let second_instance = match build_instance() {
Ok(instance) => instance,
Err(_) => return,
};
let second_store = crate::FileIdempotencyStore::new(&file);
assert!(second_store.is_ok());
let mut second_controller = RuntimeController::with_idempotency_store(
second_instance,
Box::new(match second_store {
Ok(store) => store,
Err(_) => return,
}),
);
move_to_running(&mut second_controller);
let second_result = second_controller.dispatch_command(&second_cmd, &context);
assert!(second_result.is_ok());
let second_result = match second_result {
Ok(result) => result,
Err(_) => return,
};
assert!(second_result.is_accepted());
assert_eq!(second_result.message(), None);
assert_eq!(second_controller.instance().event_bus().len(), 0);
let _ = std::fs::remove_file(file);
}
#[test]
fn duplicate_idempotency_key_is_rejected_after_restart_simulated() {
let file = temp_idempotency_file("restart");
let context = make_context();
let first_cmd = match make_command_with_key(Some("k-restart")) {
Ok(command) => command,
Err(_) => return,
};
let second_cmd = match make_command_with_key(Some("k-restart")) {
Ok(command) => command,
Err(_) => return,
};
let first_instance = match build_instance() {
Ok(instance) => instance,
Err(_) => return,
};
let first_store = crate::FileIdempotencyStore::new(&file);
assert!(first_store.is_ok());
let mut first_controller = RuntimeController::with_idempotency_store(
first_instance,
Box::new(match first_store {
Ok(store) => store,
Err(_) => return,
}),
);
move_to_running(&mut first_controller);
let first_result = first_controller.dispatch_command(&first_cmd, &context);
assert!(first_result.is_ok());
assert_eq!(first_controller.instance().event_bus().len(), 1);
drop(first_controller);
let second_instance = match build_instance() {
Ok(instance) => instance,
Err(_) => return,
};
let second_store = crate::FileIdempotencyStore::new(&file);
assert!(second_store.is_ok());
let mut second_controller = RuntimeController::with_idempotency_store(
second_instance,
Box::new(match second_store {
Ok(store) => store,
Err(_) => return,
}),
);
move_to_running(&mut second_controller);
let second_result = second_controller.dispatch_command(&second_cmd, &context);
assert!(second_result.is_ok());
let second_result = match second_result {
Ok(result) => result,
Err(_) => return,
};
assert!(second_result.is_accepted());
assert_eq!(second_result.message(), None);
assert_eq!(second_controller.instance().event_bus().len(), 0);
let _ = std::fs::remove_file(file);
}
#[derive(Default)]
struct DispatchProbe {
state: Mutex<DispatchProbeState>,
changed: Condvar,
}
#[derive(Default)]
struct DispatchProbeState {
entered: usize,
active: usize,
peak: usize,
released: bool,
}
impl DispatchProbe {
fn enter(&self) {
let started = Instant::now();
let mut state = self.state.lock();
state.entered += 1;
state.active += 1;
state.peak = state.peak.max(state.active);
self.changed.notify_all();
while !state.released {
let remaining = Duration::from_secs(5).saturating_sub(started.elapsed());
if remaining.is_zero() {
break;
}
self.changed.wait_for(&mut state, remaining);
}
state.active -= 1;
}
fn wait_for_entered(&self, expected: usize, timeout: Duration) -> bool {
let started = Instant::now();
let mut state = self.state.lock();
while state.entered < expected {
let remaining = timeout.saturating_sub(started.elapsed());
if remaining.is_zero() {
return false;
}
self.changed.wait_for(&mut state, remaining);
}
true
}
fn release(&self) {
let mut state = self.state.lock();
state.released = true;
self.changed.notify_all();
}
fn entered(&self) -> usize {
self.state.lock().entered
}
fn peak(&self) -> usize {
self.state.lock().peak
}
}
struct ConcurrentHandler {
probe: Arc<DispatchProbe>,
}
impl CommandHandler for ConcurrentHandler {
fn command_name(&self) -> CommandName {
CommandName::new("runtime.concurrent").unwrap()
}
fn handle(
&self,
_command: &CommandEnvelope,
_context: &dyn RuntimeContext,
) -> RuntimeResult<CommandResult> {
self.probe.enter();
Ok(CommandResult::accepted(Vec::new()))
}
}
struct ConcurrentPlugin {
probe: Arc<DispatchProbe>,
}
impl AppPlugin for ConcurrentPlugin {
fn application_manifest(&self) -> ApplicationManifestV1 {
ControllerPlugin.application_manifest()
}
fn identity(&self, node_id: NodeId) -> RuntimeIdentity {
ControllerPlugin.identity(node_id)
}
fn register_commands(&self, _registry: &mut CommandRegistry) -> RuntimeResult<()> {
Ok(())
}
fn register_events(&self, _registry: &mut EventRegistry) -> RuntimeResult<()> {
Ok(())
}
fn register_states(&self, _registry: &mut StateRegistry) -> RuntimeResult<()> {
Ok(())
}
fn register_decisions(&self, _registry: &mut DecisionRegistry) -> RuntimeResult<()> {
Ok(())
}
fn register_handlers(&self, bus: &mut crate::CommandBus) -> RuntimeResult<()> {
bus.register_handler(ConcurrentHandler {
probe: Arc::clone(&self.probe),
})
}
}
fn concurrent_controller(probe: Arc<DispatchProbe>) -> RuntimeController {
let plugin = ConcurrentPlugin { probe };
let mut builder = RuntimeBuilder::new();
builder
.with_plugin(&plugin, NodeId::new("node-a").unwrap())
.unwrap();
let mut controller = RuntimeController::new(builder.build().unwrap());
move_to_running(&mut controller);
controller
}
fn concurrent_command(id: usize, key: Option<&str>) -> CommandEnvelope {
CommandEnvelope::new(
CommandName::new("runtime.concurrent").unwrap(),
format!("cmd-concurrent-{id}"),
AppId::new("example-app").unwrap(),
NodeId::new("node-a").unwrap(),
0,
key.map(str::to_string),
vec![id as u8],
)
.unwrap()
}
#[test]
fn independent_commands_execute_concurrently() {
const WORKERS: usize = 8;
let probe = Arc::new(DispatchProbe::default());
let controller = concurrent_controller(Arc::clone(&probe));
let barrier = Arc::new(Barrier::new(WORKERS + 1));
let mut workers = Vec::new();
for id in 0..WORKERS {
let controller = controller.clone();
let barrier = Arc::clone(&barrier);
workers.push(thread::spawn(move || {
barrier.wait();
controller.dispatch_command(&concurrent_command(id, None), &make_context())
}));
}
barrier.wait();
assert!(probe.wait_for_entered(WORKERS, Duration::from_secs(2)));
assert_eq!(probe.peak(), WORKERS);
assert_eq!(controller.inflight_commands(), WORKERS);
probe.release();
for worker in workers {
assert!(worker.join().unwrap().unwrap().is_accepted());
}
assert_eq!(controller.inflight_commands(), 0);
assert_eq!(controller.instance().audit_log().len(), WORKERS);
}
#[test]
fn matching_idempotency_keys_never_execute_twice() {
let probe = Arc::new(DispatchProbe::default());
let controller = concurrent_controller(Arc::clone(&probe));
let first_controller = controller.clone();
let first = thread::spawn(move || {
first_controller
.dispatch_command(&concurrent_command(1, Some("shared-key")), &make_context())
});
assert!(probe.wait_for_entered(1, Duration::from_secs(2)));
let duplicate =
controller.dispatch_command(&concurrent_command(1, Some("shared-key")), &make_context());
assert!(matches!(
duplicate,
Err(crate::RuntimeError::IdempotencyPending { .. })
));
assert_eq!(probe.entered(), 1);
probe.release();
assert!(first.join().unwrap().unwrap().is_accepted());
let replay = controller
.dispatch_command(&concurrent_command(1, Some("shared-key")), &make_context())
.unwrap();
assert!(replay.is_accepted());
assert_eq!(probe.entered(), 1);
assert_eq!(controller.inflight_commands(), 0);
}
#[test]
fn shutdown_rejects_new_commands_and_drains_admitted_work() {
let probe = Arc::new(DispatchProbe::default());
let controller = concurrent_controller(Arc::clone(&probe));
let dispatch_controller = controller.clone();
let active = thread::spawn(move || {
dispatch_controller.dispatch_command(&concurrent_command(1, None), &make_context())
});
assert!(probe.wait_for_entered(1, Duration::from_secs(2)));
assert_eq!(controller.inflight_commands(), 1);
assert_eq!(
controller
.apply_lifecycle_event(RuntimeLifecycleEvent::ShutdownRequested)
.unwrap(),
RuntimeLifecycleState::ShuttingDown
);
let rejected = controller
.dispatch_command(&concurrent_command(2, None), &make_context())
.unwrap();
assert!(!rejected.is_accepted());
assert_eq!(rejected.message(), Some("runtime is not ready"));
assert_eq!(probe.entered(), 1);
assert!(!controller.wait_for_inflight(Duration::from_millis(10)));
assert_eq!(
controller.apply_lifecycle_event(RuntimeLifecycleEvent::ShutdownCompleted),
Err(crate::RuntimeError::CommandRejected)
);
probe.release();
assert!(active.join().unwrap().unwrap().is_accepted());
assert!(controller.wait_for_inflight(Duration::from_secs(1)));
assert_eq!(
controller
.apply_lifecycle_event(RuntimeLifecycleEvent::ShutdownCompleted)
.unwrap(),
RuntimeLifecycleState::Stopped
);
}
fn temp_idempotency_file(name: &str) -> std::path::PathBuf {
let nanos = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_nanos())
.unwrap_or(0);
std::env::temp_dir().join(format!("appcore-controller-idemp-{name}-{nanos}.txt"))
}