diff --git a/crates/ragfs/src/cache_runtime/memory.rs b/crates/ragfs/src/cache_runtime/memory.rs index dbe9f4eb5c..49e864e1ec 100644 --- a/crates/ragfs/src/cache_runtime/memory.rs +++ b/crates/ragfs/src/cache_runtime/memory.rs @@ -3,16 +3,18 @@ use super::provider::{CacheOperation, CacheProvider}; use super::{ CacheError, CacheResult, Expiration, ListDirection, ListInsertPosition, ListInsertRequest, - ListMoveRequest, SetCondition, SetOptions, SetResult, + ListMoveRequest, ScriptRequest, ScriptResult, SetCondition, SetOptions, SetResult, }; use async_trait::async_trait; use bytes::Bytes; use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::Mutex; +use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use tokio::sync::RwLock; +type ScriptHandler = dyn Fn(ScriptRequest) -> CacheResult + Send + Sync + 'static; + /// Controllable in-memory provider for tests and smoke validation. pub struct MemoryMockProvider { values: RwLock>, @@ -33,6 +35,7 @@ pub struct MemoryMockProvider { active_sets: AtomicU64, max_active_sets: AtomicU64, next_set_failure: Mutex>, + script_handler: Option>, } struct ActiveSetGuard<'a> { @@ -67,9 +70,19 @@ impl MemoryMockProvider { active_sets: AtomicU64::new(0), max_active_sets: AtomicU64::new(0), next_set_failure: Mutex::new(None), + script_handler: None, } } + /// Install a deterministic script handler for tests that need to advertise script support. + pub fn with_script_handler(mut self, handler: F) -> Self + where + F: Fn(ScriptRequest) -> CacheResult + Send + Sync + 'static, + { + self.script_handler = Some(Arc::new(handler)); + self + } + /// Delay individual get calls to exercise inflight and concurrency behavior. pub fn with_get_delay(mut self, delay: Duration) -> Self { self.get_delay = delay; @@ -231,9 +244,9 @@ impl Default for MemoryMockProvider { #[async_trait] impl CacheProvider for MemoryMockProvider { - /// Rejects script execution because the in-memory provider has no Lua engine. + /// Reject script-dependent consumers unless a test explicitly installs a handler. fn validate_operations(&self, operations: &[CacheOperation]) -> CacheResult<()> { - if operations.contains(&CacheOperation::ExecuteScript) { + if operations.contains(&CacheOperation::ExecuteScript) && self.script_handler.is_none() { return Err(CacheError::UnsupportedOperation( CacheOperation::ExecuteScript.name().to_string(), )); @@ -618,6 +631,15 @@ impl CacheProvider for MemoryMockProvider { Ok(Some(value)) } + async fn execute_script(&self, request: ScriptRequest) -> CacheResult { + self.ensure_open()?; + let handler = self + .script_handler + .as_ref() + .ok_or_else(|| CacheError::UnsupportedScript(request.script_id.clone()))?; + handler(request) + } + async fn close(&self) -> CacheResult<()> { self.closed.store(true, Ordering::Release); self.values.write().await.clear(); @@ -677,3 +699,37 @@ fn list_range(list: &VecDeque, start: i64, stop: i64) -> Vec { .cloned() .collect() } + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::atomic::{AtomicUsize, Ordering}; + + #[tokio::test] + async fn configured_script_handler_is_advertised_and_invoked() { + let calls = Arc::new(AtomicUsize::new(0)); + let observed_calls = Arc::clone(&calls); + let provider = MemoryMockProvider::new().with_script_handler(move |request| { + assert_eq!(request.script_id, "test.script.v1"); + observed_calls.fetch_add(1, Ordering::Relaxed); + Ok(ScriptResult { + payload: Bytes::from_static(b"handled"), + }) + }); + + provider + .validate_operations(&[CacheOperation::ExecuteScript]) + .unwrap(); + let result = provider + .execute_script(ScriptRequest { + script_id: "test.script.v1".to_string(), + keys: Vec::new(), + args: Vec::new(), + }) + .await + .unwrap(); + + assert_eq!(result.payload, Bytes::from_static(b"handled")); + assert_eq!(calls.load(Ordering::Relaxed), 1); + } +} diff --git a/crates/ragfs/src/core/mountable.rs b/crates/ragfs/src/core/mountable.rs index 97565c0cfa..e1fb696e14 100644 --- a/crates/ragfs/src/core/mountable.rs +++ b/crates/ragfs/src/core/mountable.rs @@ -1755,7 +1755,12 @@ mod tests { use crate::cache_runtime::{CacheRuntime, MemoryMockProvider}; use crate::plugins::QueueFSPlugin; - let provider = Arc::new(MemoryMockProvider::new()); + let provider = Arc::new(MemoryMockProvider::new().with_script_handler(|request| { + panic!( + "script execution is outside this shutdown test's scope: {}", + request.script_id + ) + })); let runtime = CacheRuntime::memory_with_provider(provider.clone()); let mfs = MountableFS::new(); mfs.register_plugin(QueueFSPlugin::with_cache_runtime(runtime)) diff --git a/crates/ragfs/src/plugins/queuefs/cache_backend.rs b/crates/ragfs/src/plugins/queuefs/cache_backend.rs index f96759afd4..26d0005264 100644 --- a/crates/ragfs/src/plugins/queuefs/cache_backend.rs +++ b/crates/ragfs/src/plugins/queuefs/cache_backend.rs @@ -670,6 +670,32 @@ fn cache_error(operation: &str, error: CacheError) -> Error { mod tests { use super::*; + fn lifecycle_test_runtime() -> Arc { + let provider = Arc::new( + crate::cache_runtime::MemoryMockProvider::new().with_script_handler(|request| { + panic!( + "script execution is outside this lifecycle test's scope: {}", + request.script_id + ) + }), + ); + CacheRuntime::memory_with_provider(provider) + } + + #[tokio::test] + async fn open_rejects_provider_without_script_support() { + let error = CacheQueueStorage::open(CacheRuntime::memory(), "unsupported".to_string()) + .await + .err() + .expect("memory provider must be rejected"); + + assert!(matches!( + error, + Error::Internal(message) + if message.contains("unsupported cache operation: execute_script") + )); + } + #[test] fn startup_recovery_uses_original_zero_thirty_sixty_second_offsets() { assert_eq!( @@ -688,7 +714,7 @@ mod tests { #[test] fn drop_outside_tokio_runtime_removes_heartbeat() { - let runtime = CacheRuntime::memory(); + let runtime = lifecycle_test_runtime(); let tokio_runtime = tokio::runtime::Runtime::new().unwrap(); let storage = tokio_runtime .block_on(CacheQueueStorage::open( @@ -706,7 +732,7 @@ mod tests { #[tokio::test] async fn shutdown_removes_heartbeat_before_returning() { - let runtime = CacheRuntime::memory(); + let runtime = lifecycle_test_runtime(); let storage = CacheQueueStorage::open(Arc::clone(&runtime), "async-shutdown-test".to_string()) .await diff --git a/crates/ragfs/src/plugins/queuefs/mod.rs b/crates/ragfs/src/plugins/queuefs/mod.rs index cf074939e4..5cfb5244cb 100644 --- a/crates/ragfs/src/plugins/queuefs/mod.rs +++ b/crates/ragfs/src/plugins/queuefs/mod.rs @@ -1330,7 +1330,14 @@ mod tests { #[cfg(feature = "cache")] #[tokio::test] async fn cache_runtime_failure_is_propagated_by_stat_and_read_dir() { - let provider = std::sync::Arc::new(crate::cache_runtime::MemoryMockProvider::new()); + let provider = std::sync::Arc::new( + crate::cache_runtime::MemoryMockProvider::new().with_script_handler(|request| { + panic!( + "script execution is outside this failure propagation test's scope: {}", + request.script_id + ) + }), + ); let runtime = crate::cache_runtime::CacheRuntime::memory_with_provider(provider.clone()); let fs = QueueFileSystem::with_cache_runtime(runtime, "queuefs-runtime-errors".to_string()) .await