refactor: max executions

Author: hiimjako

README.md

@@ -56,6 +56,7 @@ Main entry point. Manages a pool of JS workers, each with loaded modules.
 - `Pool::builder()` — creates a new pool builder
 - `PoolBuilder::max_size(n)` — sets worker pool size
 - `PoolBuilder::with_evaluation_timeout(duration)` — module load timeout
+- `PoolBuilder::with_max_executions(limit)` — recycle workers after n executions (prevents memory leaks)
 - `PoolBuilder::add_module(name, code)` — loads a module into all workers
 - `exec(module_name, function_name, params, ExecOptions)` — executes a function
 

src/lib.rs

@@ -10,7 +10,7 @@ mod runtime;
 mod worker;
 
 pub use manager::{ModuleDefinition, WorkerManager};
-pub use options::{DomainPermission, ExecOptions, RecyclePolicy, WorkerOptions};
+pub use options::{DomainPermission, ExecOptions, MaxExecutions, RecyclePolicy, WorkerOptions};
 pub use orama_extension::{OutputChannel, SharedCache};
 pub use parameters::{FunctionParameters, TryIntoFunctionParameters};
 pub use pool::{Pool, PoolBuilder};

src/manager.rs

@@ -97,12 +97,9 @@ impl Manager for WorkerManager {
                 .with_cache(cache)
                 .with_domain_permission(worker_options.domain_permission)
                 .with_evaluation_timeout(worker_options.evaluation_timeout)
+                .with_max_executions(worker_options.max_executions)
                 .with_version(version);
 
-            if let Some(max_executions) = worker_options.max_executions {
-                builder = builder.with_max_executions(max_executions);
-            }
-
             for (name, def) in modules {
                 builder = builder.add_module(name, def.code);
             }

src/options.rs

@@ -5,6 +5,33 @@ use crate::OutputChannel;
 
 pub use crate::permission::DomainPermission;
 
+/// Controls how many times a worker can execute functions before being recycled.
+///
+/// Worker recycling helps prevent memory leaks from accumulated module cache and state.
+/// When a worker reaches its execution limit, it will be invalidated and recreated
+/// with a fresh runtime on the next use.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MaxExecutions {
+    Unlimited,
+    Limited(u64),
+}
+
+impl Default for MaxExecutions {
+    fn default() -> Self {
+        MaxExecutions::Limited(100)
+    }
+}
+
+impl MaxExecutions {
+    /// Check if the execution count has been exceeded
+    pub fn is_exceeded(&self, count: u64) -> bool {
+        match self {
+            MaxExecutions::Unlimited => false,
+            MaxExecutions::Limited(max) => count >= *max,
+        }
+    }
+}
+
 /// Policy for when to recycle/invalidate a runtime
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
 pub enum RecyclePolicy {
@@ -65,7 +92,7 @@ pub struct WorkerOptions {
     pub evaluation_timeout: Duration,
     pub domain_permission: DomainPermission,
     pub recycle_policy: RecyclePolicy,
-    pub max_executions: Option<u64>,
+    pub max_executions: MaxExecutions,
 }
 
 impl WorkerOptions {
@@ -88,8 +115,8 @@ impl WorkerOptions {
         self
     }
 
-    pub fn with_max_executions(mut self, max: u64) -> Self {
-        self.max_executions = Some(max);
+    pub fn with_max_executions(mut self, max: MaxExecutions) -> Self {
+        self.max_executions = max;
         self
     }
 }
@@ -100,7 +127,7 @@ impl Default for WorkerOptions {
             evaluation_timeout: Duration::from_secs(5),
             domain_permission: DomainPermission::DenyAll,
             recycle_policy: RecyclePolicy::default(),
-            max_executions: None,
+            max_executions: MaxExecutions::default(),
         }
     }
 }

src/pool.rs

@@ -9,7 +9,7 @@ use crate::{orama_extension::SharedCache, RecyclePolicy};
 
 use super::{
     manager::{ModuleDefinition, WorkerManager},
-    options::{DomainPermission, ExecOptions, WorkerOptions},
+    options::{DomainPermission, ExecOptions, MaxExecutions, WorkerOptions},
     parameters::TryIntoFunctionParameters,
     runtime::RuntimeError,
     worker::WorkerBuilder,
@@ -138,7 +138,7 @@ pub struct PoolBuilder {
     domain_permission: Option<DomainPermission>,
     evaluation_timeout: Option<std::time::Duration>,
     recycle_policy: Option<RecyclePolicy>,
-    max_executions: Option<u64>,
+    max_executions: Option<MaxExecutions>,
 }
 
 impl PoolBuilder {
@@ -150,7 +150,7 @@ impl PoolBuilder {
             domain_permission: None,
             evaluation_timeout: None,
             recycle_policy: None,
-            max_executions: Some(100),
+            max_executions: None,
         }
     }
 
@@ -178,9 +178,8 @@ impl PoolBuilder {
         self
     }
 
-    /// Set the maximum number of executions before recycling the runtime
-    /// to prevent memory leaks from accumulated module cache.
-    pub fn with_max_executions(mut self, max: u64) -> Self {
+    /// Set the maximum number of executions before recycling workers in the pool.
+    pub fn with_max_executions(mut self, max: MaxExecutions) -> Self {
         self.max_executions = Some(max);
         self
     }
@@ -209,7 +208,7 @@ impl PoolBuilder {
             evaluation_timeout: self.evaluation_timeout.unwrap_or(Duration::from_secs(5)),
             domain_permission: self.domain_permission.unwrap_or_default(),
             recycle_policy: self.recycle_policy.unwrap_or_default(),
-            max_executions: self.max_executions,
+            max_executions: self.max_executions.unwrap_or_default(),
         };
 
         // Validate that all modules can be loaded within the evaluation timeout
@@ -1468,4 +1467,57 @@ mod tests {
             RuntimeError::MissingModule(name) if name == "nonexistent"
         ));
     }
+
+    #[tokio::test]
+    async fn test_pool_max_executions() {
+        let _ = tracing_subscriber::fmt::try_init();
+
+        let counter_code = r#"
+            let callCount = 0;
+            function increment() {
+                callCount++;
+                return callCount;
+            }
+            export default { increment };
+        "#;
+
+        let pool = Pool::builder()
+            .max_size(1)
+            .add_module("counter", counter_code.to_string())
+            .with_max_executions(MaxExecutions::Limited(2))
+            .build()
+            .await
+            .unwrap();
+
+        // First 3 executions should increment the counter
+        let result1: i32 = pool
+            .exec("counter", "increment", (), ExecOptions::new())
+            .await
+            .unwrap();
+        assert_eq!(result1, 1);
+
+        let result2: i32 = pool
+            .exec("counter", "increment", (), ExecOptions::new())
+            .await
+            .unwrap();
+        assert_eq!(result2, 2);
+
+        // After 2 executions, the worker should be recycled
+        // and the counter should reset to 1
+        let result4: i32 = pool
+            .exec("counter", "increment", (), ExecOptions::new())
+            .await
+            .unwrap();
+        assert_eq!(
+            result4, 1,
+            "Counter should reset after max_executions is reached"
+        );
+
+        // Verify it continues to work correctly after recycling
+        let result5: i32 = pool
+            .exec("counter", "increment", (), ExecOptions::new())
+            .await
+            .unwrap();
+        assert_eq!(result5, 2);
+    }
 }

src/worker.rs

@@ -8,7 +8,7 @@ use crate::orama_extension::SharedCache;
 use crate::runtime::ModuleName;
 
 use super::{
-    options::{DomainPermission, ExecOptions},
+    options::{DomainPermission, ExecOptions, MaxExecutions},
     parameters::TryIntoFunctionParameters,
     runtime::{Runtime, RuntimeError},
 };
@@ -27,7 +27,7 @@ pub struct Worker {
     cache: SharedCache,
     domain_permission: DomainPermission,
     evaluation_timeout: std::time::Duration,
-    max_executions: Option<u64>,
+    max_executions: MaxExecutions,
     execution_count: u64,
     version: u64,
 }
@@ -38,7 +38,7 @@ impl Worker {
         cache: SharedCache,
         domain_permission: DomainPermission,
         evaluation_timeout: std::time::Duration,
-        max_executions: Option<u64>,
+        max_executions: MaxExecutions,
         version: u64,
     ) -> Self {
         Self {
@@ -121,15 +121,10 @@ impl Worker {
         }
 
         // Check if we need to invalidate the runtime due to execution limit
-        if let Some(max_executions) = self.max_executions {
-            if self.execution_count >= max_executions {
-                warn!(
-                    "Worker reached max executions limit ({}), invalidating runtime",
-                    max_executions
-                );
-                self.runtime = None;
-                self.execution_count = 0;
-            }
+        if self.max_executions.is_exceeded(self.execution_count) {
+            warn!("Worker reached max executions limit, invalidating runtime");
+            self.runtime = None;
+            self.execution_count = 0;
         }
 
         let domain_permission = exec_options
@@ -208,7 +203,7 @@ pub struct WorkerBuilder {
     cache: Option<SharedCache>,
     domain_permission: Option<DomainPermission>,
     evaluation_timeout: Option<std::time::Duration>,
-    max_executions: Option<u64>,
+    max_executions: MaxExecutions,
     version: u64,
 }
 
@@ -220,7 +215,7 @@ impl WorkerBuilder {
             cache: None,
             domain_permission: None,
             evaluation_timeout: None,
-            max_executions: None,
+            max_executions: MaxExecutions::default(),
             version: 0,
         }
     }
@@ -254,10 +249,9 @@ impl WorkerBuilder {
         self
     }
 
-    /// Set the maximum number of executions before recycling the runtime
-    /// to prevent memory leaks from accumulated module cache.
-    pub fn with_max_executions(mut self, max: u64) -> Self {
-        self.max_executions = Some(max);
+    /// Set the maximum number of executions before recycling the runtime.
+    pub fn with_max_executions(mut self, max: MaxExecutions) -> Self {
+        self.max_executions = max;
         self
     }
 
@@ -440,7 +434,7 @@ mod tests {
 
         let mut worker = Worker::builder()
             .add_module("counter", counter_code.to_string())
-            .with_max_executions(3)
+            .with_max_executions(MaxExecutions::Limited(3))
             .build()
             .await
             .unwrap();