From 7534e13826d8567e251ae467e40521455e21a695 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 11:38:50 +0000 Subject: [PATCH] refactor(execution_engine): drop unnecessary Mutex around EngineLookup, fix JS runner error message EngineLookup's methods only take &self - there's no mutation - so the Mutex wrapping it in Engine (and the PoisonedLock error variant it required) was unneeded synchronization inherited from apid's own mutation needs, not the trait's contract. apid's OperationRepos is legitimately mutated elsewhere (background loader), so that lock is now encapsulated behind a new LockedLookup adapter instead of leaking into Engine's field type. Also fixes a copy-paste bug where Engine::run's JAVASCRIPT SimpleCode dispatch reported "Code runner not found for python" on a missing JS runner, by factoring the duplicated Python/JS dispatch blocks into one dispatch_code_runner helper parameterized on the language key. Closes #16 Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_019EKR96FrBygNE1sfscQ6CG --- binary/apid/src/main.rs | 29 +++- runners/workflow_runner/src/lib.rs | 14 +- usecases/execution_engine/src/error.rs | 5 - usecases/execution_engine/src/lib.rs | 225 ++++++++++++++----------- 4 files changed, 160 insertions(+), 113 deletions(-) diff --git a/binary/apid/src/main.rs b/binary/apid/src/main.rs index 813ffdf..2f0ad3d 100644 --- a/binary/apid/src/main.rs +++ b/binary/apid/src/main.rs @@ -494,6 +494,28 @@ fn panic_message(panic: &(dyn std::any::Any + Send)) -> String { } } +/// Adapts a shared, mutably-locked [`OperationRepos`] (also written to by +/// the background loader - see [`workers::start_background_watcher`]) to +/// [`EngineLookup`]'s read-only, unlocked contract, encapsulating the lock +/// so [`execution_engine::Engine`] itself never has to know one exists. +struct LockedLookup(Arc>); + +impl EngineLookup for LockedLookup { + fn get_service(&self, id: &str) -> Option { + self.0 + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get_service(id) + } + + fn get_credentials(&self, id: &str) -> Option { + self.0 + .lock() + .unwrap_or_else(PoisonError::into_inner) + .get_credentials(id) + } +} + /// Builds an [`execution_engine::Engine`] backed by `lookup`, and registers /// every adapter enabled by this build's Cargo features (the API-call /// connector is always registered; Python/JavaScript/Lua code runners, @@ -506,7 +528,7 @@ fn panic_message(panic: &(dyn std::any::Any + Send)) -> String { /// down from an already-running async context) — callers on the async /// main thread must invoke this through `tokio::task::spawn_blocking`. fn construct_execution_engine( - lookup: Arc>, + lookup: Arc, signals: Signals, workflow_path: &str, api_path: &str, @@ -644,12 +666,13 @@ async fn main() -> anyhow::Result<()> { let engine = { let repos = Arc::>::clone(&repos); + let lookup: Arc = Arc::new(LockedLookup(repos)); let signals = Arc::clone(&signals); let workflow_path = config.log.workflow_path.clone(); let api_path = config.log.api_path.clone(); tokio::task::spawn_blocking(move || { - construct_execution_engine(repos, signals, &workflow_path, &api_path) + construct_execution_engine(lookup, signals, &workflow_path, &api_path) }) .await?? }; @@ -703,7 +726,7 @@ mod tests { Box::new(InMemoryRepository::new()), Box::new(InMemoryRepository::new()), ); - let repos: Arc> = Arc::new(Mutex::new(repos)); + let repos: Arc = Arc::new(repos); let signals: Signals = Arc::new(Mutex::new(HashMap::new())); let log_dir = tempfile::tempdir().unwrap(); diff --git a/runners/workflow_runner/src/lib.rs b/runners/workflow_runner/src/lib.rs index b340e41..08e02a9 100644 --- a/runners/workflow_runner/src/lib.rs +++ b/runners/workflow_runner/src/lib.rs @@ -380,7 +380,7 @@ mod tests { fn empty_engine() -> Arc> { let (logger, _handle) = common_data_structures::log_writer::LogWriter::spawn(tempfile::tempfile().unwrap()); - let lookup: Arc> = Arc::new(Mutex::new(EmptyLookup)); + let lookup: Arc = Arc::new(EmptyLookup); Arc::new(RwLock::new(execution_engine::Engine::new(lookup, logger))) } @@ -529,8 +529,8 @@ mod tests { async fn workflow_adapter_api_call_bridges_into_the_registered_async_connector() { let (logger, _handle) = common_data_structures::log_writer::LogWriter::spawn(tempfile::tempfile().unwrap()); - let lookup: Arc> = - Arc::new(Mutex::new(SingleServiceLookup(swagger_service()))); + let lookup: Arc = + Arc::new(SingleServiceLookup(swagger_service())); let engine = Arc::new(RwLock::new(execution_engine::Engine::new( lookup, logger.clone(), @@ -567,8 +567,8 @@ mod tests { async fn workflow_adapter_api_call_errors_for_a_non_swagger_manifest() { let (logger, _handle) = common_data_structures::log_writer::LogWriter::spawn(tempfile::tempfile().unwrap()); - let lookup: Arc> = - Arc::new(Mutex::new(SingleServiceLookup(simple_code_service()))); + let lookup: Arc = + Arc::new(SingleServiceLookup(simple_code_service())); let engine = Arc::new(RwLock::new(execution_engine::Engine::new( lookup, logger.clone(), @@ -608,8 +608,8 @@ mod tests { async fn workflow_adapter_api_run_bridges_into_the_existing_sync_engine() { let (logger, _handle) = common_data_structures::log_writer::LogWriter::spawn(tempfile::tempfile().unwrap()); - let lookup: Arc> = - Arc::new(Mutex::new(SingleServiceLookup(simple_code_service()))); + let lookup: Arc = + Arc::new(SingleServiceLookup(simple_code_service())); let engine = Arc::new(RwLock::new(execution_engine::Engine::new( lookup, logger.clone(), diff --git a/usecases/execution_engine/src/error.rs b/usecases/execution_engine/src/error.rs index 135d42a..9f2221d 100644 --- a/usecases/execution_engine/src/error.rs +++ b/usecases/execution_engine/src/error.rs @@ -32,11 +32,6 @@ pub enum ExecutionEngine { source: io::Error, }, - /// The shared lookup's lock was poisoned by a panic in another thread - /// while holding it. - #[error("Get out of here! The Lock is poisoned: {0}")] - PoisonedLock(String), - /// TODO: Rename to OutputPort #[error(transparent)] Other { diff --git a/usecases/execution_engine/src/lib.rs b/usecases/execution_engine/src/lib.rs index bb53a1d..3203d2d 100644 --- a/usecases/execution_engine/src/lib.rs +++ b/usecases/execution_engine/src/lib.rs @@ -34,7 +34,7 @@ use services::{ AsyncDataConnectionRunner, CodeRunner, DataConnectionRunner, DataConnectorBundle, EngineInputContext, EngineLookup, FilteredRunner, InputPrompter, ScriptRunner, WorkflowRunner, }; -use std::{collections::HashMap, sync::Mutex}; +use std::collections::HashMap; use chrono::offset::Local; use core_entities::service::{code_resource::Language, service_manifest_latest}; @@ -60,8 +60,11 @@ pub fn wrap_result(result: Value, raw_response: bool) -> Value { /// manifest's type. pub struct Engine { /// The input port used to resolve a service/its credentials by ID at - /// execution time. - lookup: Arc>, + /// execution time. `EngineLookup`'s methods take `&self` only - no + /// mutation - so this holds no lock of its own; an implementation + /// backed by mutable state (e.g. `apid`'s background-loaded service + /// repository) is responsible for its own internal synchronization. + lookup: Arc, /// Where every dispatched run is logged. logger: LogWriter, @@ -105,7 +108,7 @@ impl Engine { /// Creates an [`Engine`] with no adapters registered yet; use the /// `register_*` methods to add them. #[inline] - pub fn new(lookup: Arc>, logger: LogWriter) -> Self { + pub fn new(lookup: Arc, logger: LogWriter) -> Self { Self { lookup, logger, @@ -216,19 +219,12 @@ impl Engine { let (service_name, operation_name) = Self::parse_identifier(identifier, context.parent.as_deref())?; - let (service, credentials) = { - let lookup = self - .lookup - .lock() - .map_err(|err| error::ExecutionEngine::PoisonedLock(err.to_string()))?; - let service = lookup - .get_service(service_name) - .ok_or_else(|| error::ExecutionEngine::NotFound(identifier.into()))?; - - let credentials = lookup.get_credentials(service_name); + let service = self + .lookup + .get_service(service_name) + .ok_or_else(|| error::ExecutionEngine::NotFound(identifier.into()))?; + let credentials = self.lookup.get_credentials(service_name); - (service, credentials) - }; let service = service.v1(); let manifest = service.manifest.v2(); @@ -320,44 +316,24 @@ impl Engine { } &Some(service_manifest_latest::Value::SimpleCode(ref simple_code)) => { match simple_code.code.language.enum_value() { - Ok(Language::PYTHON) => { - if let Some(code_runner) = self.code_runners.get("python") { - self.log(identifier, "SIMPLE_CODE", "STARTED")?; - let result = code_runner.run( - service_name, - operation_name, - simple_code.code.codeString(), - params, - context, - )?; - self.log(identifier, "SIMPLE_CODE", "COMPLETED")?; - - Ok(result) - } else { - Err(error::ExecutionEngine::NotFound( - "Code runner not found for python".into(), - )) - } - } - Ok(Language::JAVASCRIPT) => { - if let Some(code_runner) = self.code_runners.get("js") { - self.log(identifier, "SIMPLE_CODE", "STARTED")?; - let result = code_runner.run( - service_name, - operation_name, - simple_code.code.codeString(), - params, - context, - )?; - self.log(identifier, "SIMPLE_CODE", "COMPLETED")?; - - Ok(result) - } else { - Err(error::ExecutionEngine::NotFound( - "Code runner not found for python".into(), - )) - } - } + Ok(Language::PYTHON) => self.dispatch_code_runner( + identifier, + service_name, + operation_name, + "python", + simple_code.code.codeString(), + params, + context, + ), + Ok(Language::JAVASCRIPT) => self.dispatch_code_runner( + identifier, + service_name, + operation_name, + "js", + simple_code.code.codeString(), + params, + context, + ), // LUA is deliberately not dispatched to here - see #73: // `Workflow`-kind manifests (via `WorkflowRunner`) are // the replacement for Lua `SimpleCode` operations, not @@ -407,15 +383,10 @@ impl Engine { let (service_name, operation_name) = Self::parse_identifier(identifier, context.parent.as_deref())?; - let service = { - let lookup = self - .lookup - .lock() - .map_err(|err| error::ExecutionEngine::PoisonedLock(err.to_string()))?; - lookup - .get_service(service_name) - .ok_or_else(|| error::ExecutionEngine::NotFound(identifier.into()))? - }; + let service = self + .lookup + .get_service(service_name) + .ok_or_else(|| error::ExecutionEngine::NotFound(identifier.into()))?; let service = service.v1(); let manifest = service.manifest.v2(); @@ -491,13 +462,9 @@ impl Engine { return false; }; - let Ok(lookup) = self.lookup.lock() else { - return false; - }; - let Some(service) = lookup.get_service(service_name) else { + let Some(service) = self.lookup.get_service(service_name) else { return false; }; - drop(lookup); let service = service.v1(); let manifest = service.manifest.v2(); @@ -538,17 +505,12 @@ impl Engine { let (service_name, operation_name) = Self::parse_identifier(identifier, context.parent.as_deref())?; - let (service, credentials) = { - let lookup = self - .lookup - .lock() - .map_err(|err| error::ExecutionEngine::PoisonedLock(err.to_string()))?; - let service = lookup - .get_service(service_name) - .ok_or_else(|| error::ExecutionEngine::NotFound(identifier.into()))?; - let credentials = lookup.get_credentials(service_name); - (service, credentials) - }; + let service = self + .lookup + .get_service(service_name) + .ok_or_else(|| error::ExecutionEngine::NotFound(identifier.into()))?; + let credentials = self.lookup.get_credentials(service_name); + let service = service.v1(); let manifest = service.manifest.v2(); @@ -573,6 +535,37 @@ impl Engine { } } + /// Dispatches to the [`CodeRunner`] registered under `lang_key`, + /// logging `SIMPLE_CODE` start/completion - the shared body of + /// [`Engine::run`]'s `SimpleCode` dispatch, parameterized by language + /// so `python`/`javascript` (and any future language) share one + /// implementation instead of near-identical copies (see #16). + /// + /// # Errors + /// Returns an error if no `CodeRunner` is registered for `lang_key`, or + /// if the registered runner's own call fails. + #[allow(clippy::too_many_arguments)] + fn dispatch_code_runner( + &self, + identifier: &str, + service_name: &str, + operation_name: &str, + lang_key: &str, + source_code: &str, + params: Value, + context: &EngineInputContext, + ) -> error::Result { + let code_runner = self.code_runners.get(lang_key).ok_or_else(|| { + error::ExecutionEngine::NotFound(format!("Code runner not found for {lang_key}")) + })?; + + self.log(identifier, "SIMPLE_CODE", "STARTED")?; + let result = code_runner.run(service_name, operation_name, source_code, params, context)?; + self.log(identifier, "SIMPLE_CODE", "COMPLETED")?; + + Ok(result) + } + /// Queues a timestamped `(action_type) [status] id` line to the shared /// log writer. fn log(&self, id: &str, action_type: &str, status: &str) -> error::Result<()> { @@ -642,7 +635,7 @@ mod tests { let (writer, handle) = common_data_structures::log_writer::LogWriter::spawn(file.try_clone().unwrap()); - let lookup: Arc> = Arc::new(Mutex::new(FakeLookup)); + let lookup: Arc = Arc::new(FakeLookup); let engine = Engine::new(lookup, writer.clone()); engine.log("svc.op", "ACTION", "STARTED").unwrap(); @@ -720,10 +713,48 @@ mod tests { writer } + /// Builds a [`VersionedServiceTree`] wrapping a single `SimpleCode` + /// manifest in `language`. + fn simple_code_service( + language: core_entities::service::code_resource::Language, + ) -> core_entities::service::VersionedServiceTree { + let mut code = core_entities::service::CodeResource::new(); + code.set_codeString("return 1".to_owned()); + code.language = protobuf::EnumOrUnknown::new(language); + + let mut simple_code = core_entities::service::SimpleCodeService::new(); + simple_code.code = protobuf::MessageField::some(code); + + let mut manifest = core_entities::service::ServiceManifest::new(); + manifest.mut_v2().set_simpleCode(simple_code); + + let mut tree = core_entities::service::VersionedServiceTree::new(); + tree.mut_v1().manifest = protobuf::MessageField::some(manifest); + tree + } + + #[test] + fn run_reports_the_correct_language_when_no_javascript_code_runner_is_registered() { + let lookup: Arc = Arc::new(WorkflowLookup( + simple_code_service(core_entities::service::code_resource::Language::JAVASCRIPT), + )); + let engine = Engine::new(lookup, test_logger()); + + let ctx = EngineInputContext::new(None, "exec-1".into(), false); + let result = engine.run("svc.execute", Value::Null, Value::Null, &ctx); + + let err = result.expect_err("expected an error since no js runner is registered"); + assert_eq!( + err.to_string(), + "Not found: Code runner not found for js", + "the JAVASCRIPT branch's error message previously wrongly said 'python' - #16" + ); + } + #[tokio::test] async fn run_workflow_dispatches_to_the_registered_workflow_runner() { - let lookup: Arc> = - Arc::new(Mutex::new(WorkflowLookup(workflow_service("return 42")))); + let lookup: Arc = + Arc::new(WorkflowLookup(workflow_service("return 42"))); let mut engine = Engine::new(lookup, test_logger()); let calls = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); @@ -755,8 +786,8 @@ mod tests { #[tokio::test] async fn run_workflow_errors_when_no_workflow_runner_is_registered() { - let lookup: Arc> = - Arc::new(Mutex::new(WorkflowLookup(workflow_service("return 42")))); + let lookup: Arc = + Arc::new(WorkflowLookup(workflow_service("return 42"))); let engine = Engine::new(lookup, test_logger()); let ctx = EngineInputContext::new(None, "exec-1".into(), false); @@ -775,8 +806,7 @@ mod tests { let mut tree = core_entities::service::VersionedServiceTree::new(); tree.mut_v1().manifest = protobuf::MessageField::some(manifest); - let lookup: Arc> = - Arc::new(Mutex::new(WorkflowLookup(tree))); + let lookup: Arc = Arc::new(WorkflowLookup(tree)); let mut engine = Engine::new(lookup, test_logger()); let calls = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); engine.register_workflow_runner(Arc::new(FakeWorkflowRunner { @@ -798,8 +828,8 @@ mod tests { #[test] fn is_workflow_operation_is_true_for_a_workflow_manifest() { - let lookup: Arc> = - Arc::new(Mutex::new(WorkflowLookup(workflow_service("return 42")))); + let lookup: Arc = + Arc::new(WorkflowLookup(workflow_service("return 42"))); let engine = Engine::new(lookup, test_logger()); let ctx = EngineInputContext::new(None, "exec-1".into(), false); @@ -814,8 +844,7 @@ mod tests { let mut tree = core_entities::service::VersionedServiceTree::new(); tree.mut_v1().manifest = protobuf::MessageField::some(manifest); - let lookup: Arc> = - Arc::new(Mutex::new(WorkflowLookup(tree))); + let lookup: Arc = Arc::new(WorkflowLookup(tree)); let engine = Engine::new(lookup, test_logger()); let ctx = EngineInputContext::new(None, "exec-1".into(), false); @@ -825,7 +854,7 @@ mod tests { #[test] fn is_workflow_operation_is_false_when_the_service_is_not_found() { - let lookup: Arc> = Arc::new(Mutex::new(FakeLookup)); + let lookup: Arc = Arc::new(FakeLookup); let engine = Engine::new(lookup, test_logger()); let ctx = EngineInputContext::new(None, "exec-1".into(), false); @@ -835,7 +864,7 @@ mod tests { #[test] fn is_workflow_operation_is_false_for_an_unparseable_identifier() { - let lookup: Arc> = Arc::new(Mutex::new(FakeLookup)); + let lookup: Arc = Arc::new(FakeLookup); let engine = Engine::new(lookup, test_logger()); let ctx = EngineInputContext::new(None, "exec-1".into(), false); @@ -881,8 +910,8 @@ mod tests { #[test] fn resolve_data_connector_returns_owned_pieces_for_a_swagger_manifest() { - let lookup: Arc> = - Arc::new(Mutex::new(WorkflowLookup(swagger_service()))); + let lookup: Arc = + Arc::new(WorkflowLookup(swagger_service())); let mut engine = Engine::new(lookup, test_logger()); let calls = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); @@ -905,8 +934,8 @@ mod tests { #[test] fn resolve_data_connector_errors_when_no_async_connector_is_registered() { - let lookup: Arc> = - Arc::new(Mutex::new(WorkflowLookup(swagger_service()))); + let lookup: Arc = + Arc::new(WorkflowLookup(swagger_service())); let engine = Engine::new(lookup, test_logger()); let ctx = EngineInputContext::new(None, "exec-1".into(), false); @@ -921,8 +950,8 @@ mod tests { #[test] fn resolve_data_connector_errors_when_the_manifest_is_not_swagger() { - let lookup: Arc> = - Arc::new(Mutex::new(WorkflowLookup(workflow_service("return 42")))); + let lookup: Arc = + Arc::new(WorkflowLookup(workflow_service("return 42"))); let mut engine = Engine::new(lookup, test_logger()); let calls = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));