diff --git a/lib/apps/fabro-server/tests/it/api/mod.rs b/lib/apps/fabro-server/tests/it/api/mod.rs index 7fcdf254a..8833cfa21 100644 --- a/lib/apps/fabro-server/tests/it/api/mod.rs +++ b/lib/apps/fabro-server/tests/it/api/mod.rs @@ -8,6 +8,7 @@ mod events; mod install; mod install_openai_compatible; mod mcp_servers; +mod petri_store; mod routing; mod run_files; mod runs; diff --git a/lib/apps/fabro-server/tests/it/api/petri_store.rs b/lib/apps/fabro-server/tests/it/api/petri_store.rs new file mode 100644 index 000000000..9c8d6d157 --- /dev/null +++ b/lib/apps/fabro-server/tests/it/api/petri_store.rs @@ -0,0 +1,307 @@ +//! The Petri run store over HTTP: Petri's store conformance suite against +//! `HttpRunStore` talking to a loopback server, the operator release through +//! the server's store, a lost reply to an append, and two workers contending +//! for one run's lease. + +use std::collections::HashMap; +use std::future; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use axum::Router; +use axum::extract::{Request, State}; +use axum::http::{Method, StatusCode}; +use axum::middleware::{self, Next}; +use axum::response::Response; +use fabro_client::{Client, Credential, apply_bearer_token_auth}; +use fabro_petri::HttpRunStore; +use fabro_petri::petri::{Access, LogId, OwnerId, RunKey, RunLogs, RunStore, StoreError}; +use fabro_petri::test_support::run_store::{self, conformance, stale_owner_conformance}; +use fabro_server::server::{AppState, RouterOptions, build_router_with_options}; +use fabro_server::test_support::{test_app_state, test_auth_mode}; +use fabro_types::RunId; +use tokio::net::TcpListener; +use tokio::runtime::Handle; +use tokio::task; + +/// A loopback server over `state`, its router wrapped by `wrap`. +async fn serve(state: Arc, wrap: impl FnOnce(Router) -> Router) -> String { + let listener = TcpListener::bind("127.0.0.1:0") + .await + .expect("a loopback listener binds"); + let addr = listener.local_addr().expect("the listener has an address"); + let router = wrap(build_router_with_options( + state, + &test_auth_mode(), + RouterOptions::default(), + )); + tokio::spawn(async move { + let _ = axum::serve(listener, router).await; + }); + format!("http://{addr}") +} + +/// A worker's client: the run's worker token as its bearer, and a request +/// timeout after which a reply counts as lost. +async fn worker_client(base_url: &str, token: &str, timeout: Duration) -> Client { + let http = apply_bearer_token_auth(fabro_http::HttpClientBuilder::new().no_proxy(), token) + .expect("the bearer header builds") + .build() + .expect("the test HTTP client builds"); + Client::builder() + .transport(base_url, http) + .credential(Credential::Worker(token.to_string())) + .request_timeout(timeout) + .connect() + .await + .expect("the worker client connects") +} + +/// The Petri key of a Fabro run: its id. +fn petri_key(run_id: RunId) -> RunKey { + RunKey::new(run_id.to_string()) +} + +/// Wait until the server's store shows no lease holder on `key`: a dropped +/// handle's release travels to the server on its own, and only the store +/// that dropped it awaits that. Another worker starts after the first is +/// gone, which is what this waits for. +async fn wait_until_released(store: &fabro_petri::SqliteRunStore, key: &RunKey) { + for _ in 0..500 { + if store.owner(key).await.expect("reads the lease").is_none() { + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + panic!("the lease on {key} was not released"); +} + +/// One worker's view of one run: a store over a client whose token names +/// the run. +async fn worker_store(state: &AppState, base_url: &str, run_id: RunId) -> HttpRunStore { + let token = state.test_issue_worker_token(&run_id); + HttpRunStore::new(worker_client(base_url, &token, Duration::from_secs(5)).await) +} + +/// The suite opens runs under keys of its own (`lifecycle`, `drop`), while a +/// key over the API is a Fabro run id that the worker's token names. This +/// adapter gives each suite key a Fabro run of its own, with a token minted +/// for that run alone: the least a worker holds. +struct WorkerRuns { + state: Arc, + base_url: String, + runs: Mutex>>, +} + +struct WorkerRun { + run_id: RunId, + store: HttpRunStore, +} + +impl WorkerRuns { + fn new(state: Arc, base_url: String) -> Self { + Self { + state, + base_url, + runs: Mutex::default(), + } + } + + async fn run(&self, key: &RunKey) -> Arc { + if let Some(run) = self.runs.lock().expect("runs lock").get(key) { + return Arc::clone(run); + } + let run_id = RunId::new(); + let store = worker_store(&self.state, &self.base_url, run_id).await; + let run = Arc::new(WorkerRun { run_id, store }); + self.runs + .lock() + .expect("runs lock") + .insert(key.clone(), Arc::clone(&run)); + run + } + + /// The Fabro run a suite key was given, once opened. + fn run_id(&self, key: &RunKey) -> RunId { + self.runs + .lock() + .expect("runs lock") + .get(key) + .expect("the suite opens a key before it releases it") + .run_id + } +} + +#[async_trait::async_trait] +impl RunStore for WorkerRuns { + async fn open(&self, key: &RunKey, access: Access) -> Result, StoreError> { + let run = self.run(key).await; + run.store.open(&petri_key(run.run_id), access).await + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn the_http_store_passes_the_conformance_suite() { + let state = test_app_state(); + let base_url = serve(Arc::clone(&state), |router| router).await; + let runs: Arc = Arc::new(WorkerRuns::new(state, base_url)); + conformance(|| Arc::clone(&runs)).await; +} + +/// An operator release through the server's own store ends the worker's +/// lease from outside: the worker's handle turns stale and the next writer +/// takes the run. The release is async, so the suite's synchronous closure +/// blocks on it in place. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn an_operator_release_through_the_server_makes_the_worker_stale() { + let state = test_app_state(); + let base_url = serve(Arc::clone(&state), |router| router).await; + let runs = WorkerRuns::new(Arc::clone(&state), base_url); + let release = |key: &RunKey| { + let key = petri_key(runs.run_id(key)); + let store = state.test_petri_run_store(); + task::block_in_place(|| Handle::current().block_on(store.release_lease(&key))) + .expect("the operator releases the lease"); + }; + stale_owner_conformance(&runs, release).await; +} + +/// Swallow the reply of the first append the server commits: the handler +/// runs, its transaction commits, and the response never leaves. That is +/// a lost reply as the worker sees it. +async fn swallow_one_append_reply( + State(swallowed): State>, + request: Request, + next: Next, +) -> Response { + let is_append = request.method() == Method::POST && request.uri().path().ends_with("/records"); + let response = next.run(request).await; + if is_append + && response.status() == StatusCode::NO_CONTENT + && swallowed + .compare_exchange(0, 1, Ordering::SeqCst, Ordering::SeqCst) + .is_ok() + { + future::pending::<()>().await; + } + response +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_lost_reply_to_an_append_is_safe_to_retry() { + let state = test_app_state(); + let swallowed = Arc::new(AtomicUsize::new(0)); + let base_url = serve(Arc::clone(&state), { + let swallowed = Arc::clone(&swallowed); + move |router| { + router.layer(middleware::from_fn_with_state( + swallowed, + swallow_one_append_reply, + )) + } + }) + .await; + let run_id = RunId::new(); + let token = state.test_issue_worker_token(&run_id); + let timeout = Duration::from_millis(500); + let store = HttpRunStore::new(worker_client(&base_url, &token, timeout).await); + let key = petri_key(run_id); + let logs = store + .open(&key, Access::Create { + owner: OwnerId::new("worker"), + }) + .await + .expect("the worker creates the run"); + + let batch = [ + run_store::record(0, "execution.started"), + run_store::record(1, "step.started"), + ]; + let started = Instant::now(); + logs.append(&LogId::Coordinator, &batch) + .await + .expect("the append succeeds once the resend is answered"); + assert_eq!( + swallowed.load(Ordering::SeqCst), + 1, + "the server committed one append whose reply was swallowed" + ); + assert!( + started.elapsed() >= timeout, + "the first attempt waited out the request timeout" + ); + assert_eq!( + logs.read(&LogId::Coordinator).await.expect("reads"), + batch.to_vec(), + "the log holds each record once" + ); +} + +/// Two workers with owners of their own never hold one run's lease at the +/// same time, in either order, and the server's store reports who holds it. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn two_workers_cannot_both_hold_a_run_lease() { + let state = test_app_state(); + let base_url = serve(Arc::clone(&state), |router| router).await; + let run_id = RunId::new(); + let key = petri_key(run_id); + let first_worker = worker_store(&state, &base_url, run_id).await; + let second_worker = worker_store(&state, &base_url, run_id).await; + let first = OwnerId::new("first"); + let second = OwnerId::new("second"); + let server_store = state.test_petri_run_store(); + + let held = first_worker + .open(&key, Access::Create { + owner: first.clone(), + }) + .await + .expect("the first worker creates"); + assert_eq!( + server_store.owner(&key).await.expect("reads"), + Some(first.clone()) + ); + let refused = second_worker + .open(&key, Access::Write { + owner: second.clone(), + }) + .await + .err() + .expect("the second worker is refused while the first holds the lease"); + assert!( + matches!(&refused, StoreError::Leased { owner, .. } if *owner == first), + "{refused}" + ); + assert!( + refused.to_string().contains(&run_id.to_string()), + "the message names the run: {refused}" + ); + + drop(held); + wait_until_released(server_store, &key).await; + let taken = second_worker + .open(&key, Access::Write { + owner: second.clone(), + }) + .await + .expect("the second worker takes the run once the first releases"); + assert_eq!( + server_store.owner(&key).await.expect("reads"), + Some(second.clone()) + ); + let refused = first_worker + .open(&key, Access::Write { + owner: first.clone(), + }) + .await + .err() + .expect("the first worker is refused in turn"); + assert!( + matches!(&refused, StoreError::Leased { owner, .. } if *owner == second), + "{refused}" + ); + + drop(taken); + wait_until_released(server_store, &key).await; +}