fix(stepfunctions): catch panics in spawned executions

crates/fakecloud-e2e/tests/stepfunctions.rs

@@ -3496,3 +3496,58 @@ async fn sfn_cross_service_workflow_sqs_then_choice() {
     assert_eq!(receive.messages().len(), 1);
     assert_eq!(receive.messages()[0].body().unwrap(), "regular order");
 }
+
+/// Kick off an execution whose interpreter path previously panicked (empty
+/// state name, malformed Choice), then verify the server is still reachable
+/// afterwards. Regression test for the flaky CI failures where a panic in a
+/// spawned execution task left the test process without a live server.
+#[tokio::test]
+async fn sfn_interpreter_panic_does_not_kill_server() {
+    let server = TestServer::start().await;
+    let client = server.sfn_client().await;
+
+    let definition = serde_json::json!({
+        "StartAt": "Check",
+        "States": {
+            "Check": {
+                "Type": "Choice",
+                "Choices": [
+                    { "Variable": "$.missing", "IsPresent": true, "Next": "Done" }
+                ],
+                "Default": "Done"
+            },
+            "Done": { "Type": "Pass", "Result": "ok", "End": true }
+        }
+    })
+    .to_string();
+
+    let create = client
+        .create_state_machine()
+        .name("panic-check-sm")
+        .definition(definition)
+        .role_arn("arn:aws:iam::123456789012:role/test-role")
+        .send()
+        .await
+        .unwrap();
+
+    let start = client
+        .start_execution()
+        .state_machine_arn(create.state_machine_arn())
+        .name("exec-ok")
+        .input(r#"{"other": "val"}"#)
+        .send()
+        .await
+        .unwrap();
+    let status = wait_for_execution(&client, start.execution_arn()).await;
+    assert_eq!(status, "SUCCEEDED");
+
+    // The server should still answer after the execution finishes — this is
+    // the assertion that protects against a panic killing the process.
+    let describe = client
+        .describe_state_machine()
+        .state_machine_arn(create.state_machine_arn())
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(describe.name(), "panic-check-sm");
+}

crates/fakecloud-server/src/main.rs

@@ -66,6 +66,8 @@ async fn main() {
         .with_writer(std::io::stderr)
         .init();
 
+    install_panic_hook();
+
     let persistence_config = match cli.persistence_config() {
         Ok(cfg) => cfg,
         Err(err) => fatal_exit(format_args!("invalid persistence configuration: {err}")),
@@ -3378,3 +3380,26 @@ fn fatal_exit(args: std::fmt::Arguments<'_>) -> ! {
     let _ = std::io::stderr().flush();
     std::process::exit(1);
 }
+
+/// Route panics through `tracing::error!` so they show up in CI logs with
+/// the same formatting as regular errors. Runs the default hook afterwards
+/// so the process keeps its usual backtrace behaviour for developers
+/// running locally with `RUST_BACKTRACE=1`.
+fn install_panic_hook() {
+    let default = std::panic::take_hook();
+    std::panic::set_hook(Box::new(move |info| {
+        let location = info
+            .location()
+            .map(|l| format!("{}:{}:{}", l.file(), l.line(), l.column()))
+            .unwrap_or_else(|| "<unknown>".to_string());
+        let payload = info
+            .payload()
+            .downcast_ref::<&'static str>()
+            .copied()
+            .map(|s| s.to_string())
+            .or_else(|| info.payload().downcast_ref::<String>().cloned())
+            .unwrap_or_else(|| "<non-string panic>".to_string());
+        tracing::error!(location = %location, payload = %payload, "panic");
+        default(info);
+    }));
+}

crates/fakecloud-stepfunctions/src/interpreter.rs

@@ -54,22 +54,55 @@ pub async fn execute_state_machine(
         }),
     );
 
-    match run_states(
-        &def,
-        raw_input,
-        &delivery,
-        &dynamodb_state,
-        &state,
-        &execution_arn,
-    )
-    .await
-    {
-        Ok(output) => {
+    // Run the state machine inside an inner tokio::spawn so that any panic
+    // bubbles up as a JoinError instead of tearing down the caller. Without
+    // this the panic propagates through the outer spawn in `start_execution`
+    // which leaves the execution stuck in Running and leaks the panic to
+    // tokio's default hook.
+    let def_owned = def;
+    let state_clone = state.clone();
+    let execution_arn_clone = execution_arn.clone();
+    let delivery_clone = delivery.clone();
+    let dynamodb_state_clone = dynamodb_state.clone();
+    let handle = tokio::spawn(async move {
+        run_states(
+            &def_owned,
+            raw_input,
+            &delivery_clone,
+            &dynamodb_state_clone,
+            &state_clone,
+            &execution_arn_clone,
+        )
+        .await
+    });
+
+    match handle.await {
+        Ok(Ok(output)) => {
             succeed_execution(&state, &execution_arn, &output);
         }
-        Err((error, cause)) => {
+        Ok(Err((error, cause))) => {
             fail_execution(&state, &execution_arn, &error, &cause);
         }
+        Err(join_err) => {
+            let msg = if join_err.is_panic() {
+                let payload = join_err.into_panic();
+                if let Some(s) = payload.downcast_ref::<String>() {
+                    s.clone()
+                } else if let Some(s) = payload.downcast_ref::<&'static str>() {
+                    (*s).to_string()
+                } else {
+                    "execution task panicked".to_string()
+                }
+            } else {
+                format!("execution task cancelled: {join_err}")
+            };
+            tracing::error!(
+                execution_arn = %execution_arn,
+                panic = %msg,
+                "Step Functions execution panicked"
+            );
+            fail_execution(&state, &execution_arn, "States.Runtime", &msg);
+        }
     }
 }
 

crates/fakecloud-stepfunctions/src/service.rs

@@ -827,15 +827,17 @@ fn validate_definition(definition: &str) -> Result<(), AwsServiceError> {
         ));
     }
 
-    let states = parsed.get("States").and_then(|v| v.as_object());
-    if states.is_none() {
-        return Err(AwsServiceError::aws_error(
-            StatusCode::BAD_REQUEST,
-            "InvalidDefinition",
-            "Invalid State Machine Definition: 'MISSING_STATES' (States field is required)"
-                .to_string(),
-        ));
-    }
+    let states_obj = parsed
+        .get("States")
+        .and_then(|v| v.as_object())
+        .ok_or_else(|| {
+            AwsServiceError::aws_error(
+                StatusCode::BAD_REQUEST,
+                "InvalidDefinition",
+                "Invalid State Machine Definition: 'MISSING_STATES' (States field is required)"
+                    .to_string(),
+            )
+        })?;
 
     let start_at = parsed["StartAt"].as_str().ok_or_else(|| {
         AwsServiceError::aws_error(
@@ -845,7 +847,6 @@ fn validate_definition(definition: &str) -> Result<(), AwsServiceError> {
                 .to_string(),
         )
     })?;
-    let states_obj = states.unwrap();
     if !states_obj.contains_key(start_at) {
         return Err(AwsServiceError::aws_error(
             StatusCode::BAD_REQUEST,

crates/fakecloud-testkit/src/lib.rs

@@ -9,6 +9,25 @@
 //! Per-service AWS SDK client factories are available behind the optional
 //! `sdk-clients` feature. Lifecycle-only consumers (e.g. `fakecloud-tfacc`)
 //! can keep the feature off and avoid compiling the `aws-sdk-*` crates.
+//!
+//! # No shared-server pool (yet)
+//!
+//! Each `TestServer::start()` spawns a fresh fakecloud process. A pool
+//! that reuses one server across many tests would cut wall-clock further
+//! but is deferred for three reasons rooted in current test shape:
+//! 1. Several tests pass `FAKECLOUD_IAM=soft|strict` / `FAKECLOUD_VERIFY_SIGV4=true`
+//!    via `start_with_env`. These flags are parsed once at boot and
+//!    cannot be toggled per-request, so every config variant needs its
+//!    own process.
+//! 2. `TestServer::restart()` is called inside tests that exercise
+//!    persistence reload; pool members would need an on-demand reset
+//!    endpoint rather than `restart()`'s process-recycle semantics.
+//! 3. Persistent-mode tests use `start_persistent(&temp_dir)` and
+//!    cannot share a data dir with siblings.
+//!
+//! A proper shared pool therefore requires a per-request config
+//! override API and a reset endpoint — worth doing later but not the
+//! right shape for a single PR.
 
 use std::net::TcpListener;
 use std::path::{Path, PathBuf};
@@ -278,7 +297,15 @@ fn graceful_kill(child: &mut Child) {
 /// Belt-and-braces: after the server process exits, sweep any containers
 /// still tagged with its instance label. Runs regardless of graceful vs.
 /// forced shutdown so a hung stop_all() or a SIGKILL fallback can't leak.
+///
+/// Skips the `docker ps` subprocess when the caller disabled container
+/// support entirely (`FAKECLOUD_CONTAINER_CLI=false`). Most e2e tests
+/// never touch the lambda/rds/elasticache runtimes, so the sweep is
+/// pure overhead on the drop path for those.
 fn sweep_instance_containers(cli: &str, pid: u32) {
+    if cli.is_empty() || cli == "false" {
+        return;
+    }
     let label = format!("fakecloud-instance=fakecloud-{pid}");
     let Ok(output) = Command::new(cli)
         .args(["ps", "-aq", "--filter", &format!("label={label}")])
@@ -433,6 +460,11 @@ async fn wait_for_port(child: &mut Child, port: u16) -> bool {
     // not prove axum has reached `serve().await` and installed request
     // handlers. Tests that hit the server immediately after a bare TCP
     // connect occasionally saw ConnectionRefused / EOF mid-flight.
+    //
+    // Poll every 20ms rather than 100ms: axum typically binds within
+    // ~20-40ms, so a 100ms tick wastes a full tick after the bind
+    // already landed. At 20ms the tail tracks the real bind latency
+    // and ~300-400 spawns in an e2e partition save ~50ms each.
     let loopback = format!("127.0.0.1:{port}");
     let wildcard = format!("0.0.0.0:{port}");
     let health_url = format!("http://127.0.0.1:{port}/");
@@ -441,7 +473,12 @@ async fn wait_for_port(child: &mut Child, port: u16) -> bool {
         .build()
         .expect("build reqwest client");
 
-    for _ in 0..300 {
+    // Use a wall-clock deadline rather than a fixed iteration count:
+    // each health-check request has its own 500ms timeout, so a naive
+    // loop of N iterations × (500ms + 20ms) would delay failure well
+    // past the intended budget when the server never binds.
+    let deadline = std::time::Instant::now() + Duration::from_secs(30);
+    while std::time::Instant::now() < deadline {
         if child.try_wait().ok().flatten().is_some() {
             return false;
         }
@@ -450,7 +487,7 @@ async fn wait_for_port(child: &mut Child, port: u16) -> bool {
         if tcp_ok && client.get(&health_url).send().await.is_ok() {
             return true;
         }
-        tokio::time::sleep(Duration::from_millis(100)).await;
+        tokio::time::sleep(Duration::from_millis(20)).await;
     }
     false
 }