diff --git a/README.md b/README.md index 0c73867..4f08310 100644 --- a/README.md +++ b/README.md @@ -756,6 +756,11 @@ commit an external business effect. Continue using stable application idempotenc keys, provider reconciliation and external fencing when effects can repeat. Original retry policies and timeout budgets remain unchanged. +The [bounded batch example](examples/CHECKPOINTED_BATCH.md) uses these accessors +and a synthetic provider receipt store to explain failures around external commit +and checkpoint acknowledgement. Its source tests are separate from published +worker-loss qualification. + Activity handlers can return an application-defined failure type with `ActivityError`. The same type, message and retry marker are reported for remote and local activities, including cooperative workers: diff --git a/examples/CHECKPOINTED_BATCH.md b/examples/CHECKPOINTED_BATCH.md new file mode 100644 index 0000000..2dc1fd7 --- /dev/null +++ b/examples/CHECKPOINTED_BATCH.md @@ -0,0 +1,69 @@ +# Resume a bounded batch + +`checkpointed_batch.rs` uses the new acknowledged-checkpoint accessor in the +current source. It needs a native Server advertising `activity_heartbeat_resume` +v1. Source compilation and the tests below do not establish a published-package +or worker-loss result. Use the separately reviewed replacement qualifier for +that evidence. + +The example processes three items, with a maximum of sixteen. Its activity has +two attempts, one one-second retry delay, a ten-second heartbeat timeout, a +thirty-second start-to-close timeout and a sixty-second schedule-to-close timeout. +Checkpoint recovery does not replace or extend those budgets. + +Run against your isolated development Server on Linux. Supply a **new** directory +for each invocation. The program leaves its synthetic provider receipts there +for inspection. It refuses an existing directory. + +```bash +export DURABLE_WORKFLOW_SERVER_URL=http://127.0.0.1:8080 +export DURABLE_WORKFLOW_TOKEN=your-development-token +cargo run --example checkpointed_batch -- ./receipts-before before-commit +cargo run --example checkpointed_batch -- ./receipts-commit after-commit +cargo run --example checkpointed_batch -- ./receipts-checkpoint after-checkpoint +``` + +The first attempt fails once at the selected boundary. The replacement uses +`last_heartbeat_checkpoint()` and typed details to obtain its next item. Missing +resume support fails explicitly. A supported empty checkpoint starts at zero. +The checkpoint also binds the complete business input, so a cursor from another +batch or outside this batch is refused. + +| Failure boundary | Provider state | Replacement action | +| --- | --- | --- | +| Before the first commit | No receipt | Commit item zero, then acknowledge progress | +| After commit, before acknowledgement | Item zero committed, no checkpoint | Reconcile its stable business key, then acknowledge progress | +| After acknowledgement | Item zero committed and checkpoint says next item one | Continue at item one | +| Heartbeat response lost | Checkpoint may or may not have committed | Use the next claim's checkpoint and reconcile any repeated operation | + +The receipt directory models a provider's durable operation lookup. Each item +uses the same application business ID and item index across retries, rather +than the attempt ID. Every receipt binds the complete batch fingerprint as well as the item. A repeated +key must contain the identical receipt, even when no checkpoint was received. +Appending, truncating or reordering input under that business ID refuses. Changed +input or a partial receipt refuses and requires reconciliation; the example +never overwrites it or assumes an uncertain effect failed. Files and directory +entries are synchronized before a new receipt is considered committed. This is +a synthetic provider, intended for a private local directory, not an integration +with a payment or other real external service. + +The application acknowledges only after the provider reports commit or a matching +receipt. It stops if heartbeat acknowledgement is missing or the attempt must +stop. The received SDK snapshot stays immutable. A lost heartbeat response can +follow a durable Server commit, so a later attempt must trust its received +checkpoint and still reconcile repeated business keys. + +In a real integration, replace `ReceiptStore` with the provider's idempotency, +operation lookup and fencing contract. A checkpoint does not atomically commit +an external effect, and this example does not promise exactly-once effects. +Retain the business ID when restarting the same business operation in another +workflow. Separate new business operations need new IDs. + +The following tests run without a Server. They cover the three failure boundaries, +both outcomes of a lost acknowledgement response, conflicting/partial receipts +and invalid checkpoint/input bounds. They simulate activity failures rather than +proving real process loss or Server checkpoint persistence. + +```bash +cargo test --locked --example checkpointed_batch +``` diff --git a/examples/checkpointed_batch.rs b/examples/checkpointed_batch.rs new file mode 100644 index 0000000..7a6f5e4 --- /dev/null +++ b/examples/checkpointed_batch.rs @@ -0,0 +1,527 @@ +//! Synthetic receipt provider and bounded service activity. See CHECKPOINTED_BATCH.md. +use durable_workflow::{ + json, ActivityError, ActivityOptions, ActivityRetryPolicy, Client, Error, Result, Uuid, Value, + Worker, WorkflowResultOptions, +}; +use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; +use std::{ + fs::{File, OpenOptions}, + future::Future, + io::{Read, Write}, + path::PathBuf, + time::Duration, +}; + +#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)] +#[serde(rename_all = "kebab-case")] +enum FailurePoint { + BeforeCommit, + AfterCommit, + AfterCheckpoint, +} + +#[derive(Clone, Deserialize, Serialize)] +struct Batch { + business_id: String, + items: Vec, + failure: FailurePoint, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +struct Checkpoint { + batch_sha256: String, + next_item: usize, +} + +fn invalid(message: &str) -> Error { + ActivityError::new("BatchRefused", message) + .non_retryable(true) + .into() +} + +fn planned_failure() -> Error { + ActivityError::new( + "PlannedFailure", + "the first attempt stops at the chosen boundary", + ) + .into() +} + +fn provider_io(error: std::io::Error) -> Error { + ActivityError::new("ProviderUnavailable", error.to_string()).into() +} + +impl Batch { + fn fingerprint(&self) -> Result { + if self.business_id.is_empty() + || self.business_id.len() > 128 + || self.items.is_empty() + || self.items.len() > 16 + || self + .items + .iter() + .any(|item| item.is_empty() || item.len() > 128) + { + return Err(invalid( + "use a business ID and between one and sixteen bounded items", + )); + } + let bytes = serde_json::to_vec(&(&self.business_id, &self.items))?; + Ok(format!("{:x}", Sha256::digest(bytes))) + } +} + +/// This directory is a synthetic provider's durable state, separate from checkpoints. +#[derive(Clone)] +struct ReceiptStore(PathBuf); + +impl ReceiptStore { + fn commit_or_reconcile(&self, batch: &Batch, index: usize) -> Result { + // A business key survives activity retries and workflow restarts. An attempt + // ID would create a different operation instead of reconciling the old one. + let fingerprint = batch.fingerprint()?; + let item = batch + .items + .get(index) + .ok_or_else(|| invalid("item index is outside the batch"))?; + let key = serde_json::to_vec(&(&batch.business_id, index))?; + let path = self.0.join(format!("{:x}.json", Sha256::digest(&key))); + let receipt = serde_json::to_vec( + &json!({"key":[batch.business_id,index],"batch_sha256":fingerprint,"item":item}), + )?; + match OpenOptions::new().write(true).create_new(true).open(&path) { + Ok(mut file) => { + file.write_all(&receipt).map_err(provider_io)?; + file.sync_all().map_err(provider_io)?; + File::open(&self.0) + .and_then(|directory| directory.sync_all()) + .map_err(provider_io)?; + Ok(true) + } + Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { + let mut existing = Vec::new(); + let mut file = File::open(path).map_err(provider_io)?; + (&mut file) + .take(2049) + .read_to_end(&mut existing) + .map_err(provider_io)?; + if existing != receipt { + return Err(invalid( + "provider receipt is incomplete or conflicts; reconcile before retrying", + )); + } + // A previous writer may have lost its sync response. Re-establish + // durability before acknowledging a reconciled receipt as well. + file.sync_all().map_err(provider_io)?; + File::open(&self.0) + .and_then(|directory| directory.sync_all()) + .map_err(provider_io)?; + Ok(false) + } + Err(error) => Err(provider_io(error)), + } + } +} + +async fn process_batch( + store: &ReceiptStore, + batch: &Batch, + received: Option, + attempt: u64, + mut acknowledge: A, +) -> Result +where + A: FnMut(Checkpoint) -> F, + F: Future>, +{ + let fingerprint = batch.fingerprint()?; + let next = match received { + Some(checkpoint) + if checkpoint.batch_sha256 == fingerprint + && checkpoint.next_item <= batch.items.len() => + { + checkpoint.next_item + } + Some(_) => { + return Err(invalid( + "checkpoint belongs to different input or is out of range", + )) + } + None => 0, + }; + for index in next..batch.items.len() { + let fail = |point| attempt == 1 && index == 0 && batch.failure == point; + if fail(FailurePoint::BeforeCommit) { + return Err(planned_failure()); + } + let provider = store.clone(); + let request = Batch::clone(batch); + tokio::task::spawn_blocking(move || provider.commit_or_reconcile(&request, index)) + .await + .map_err(|_| invalid("provider outcome is uncertain after its task stopped"))??; + if fail(FailurePoint::AfterCommit) { + return Err(planned_failure()); + } + acknowledge(Checkpoint { + batch_sha256: fingerprint.clone(), + next_item: index + 1, + }) + .await?; + if fail(FailurePoint::AfterCheckpoint) { + return Err(planned_failure()); + } + } + Ok(json!({"business_id":batch.business_id,"completed":batch.items.len()})) +} + +#[tokio::main] +async fn main() -> Result<()> { + let mut arguments = std::env::args().skip(1); + let directory = arguments + .next() + .ok_or_else(|| invalid("supply a new receipt directory"))?; + let failure = match arguments.next().as_deref() { + Some("before-commit") => FailurePoint::BeforeCommit, + Some("after-commit") => FailurePoint::AfterCommit, + Some("after-checkpoint") => FailurePoint::AfterCheckpoint, + _ => { + return Err(invalid( + "choose before-commit, after-commit or after-checkpoint", + )) + } + }; + if arguments.next().is_some() { + return Err(invalid( + "expected exactly a receipt directory and failure point", + )); + } + // Refuse an existing directory so this demo cannot mix with application data. + let new_directory = directory.clone(); + tokio::task::spawn_blocking(move || std::fs::create_dir(new_directory)) + .await + .map_err(|_| invalid("provider directory initialization stopped"))? + .map_err(provider_io)?; + let store = ReceiptStore(PathBuf::from(directory)); + let client = Client::builder( + std::env::var("DURABLE_WORKFLOW_SERVER_URL") + .unwrap_or_else(|_| "http://127.0.0.1:8080".into()), + ) + .token(std::env::var("DURABLE_WORKFLOW_TOKEN").ok()) + .namespace("default") + .build()?; + let queue = format!("checkpointed-batch-{}", Uuid::new_v4()); + let mut worker = Worker::new(client.clone(), &queue).poll_timeout(Duration::from_millis(250)); + worker.register_typed_activity("checkpointed-batch", move |ctx, batch: Batch| { + let store = store.clone(); + async move { + let received = ctx + .last_heartbeat_checkpoint() + .map_err(|_| invalid("this activity requires activity_heartbeat_resume v1"))? + .map(|checkpoint| { + checkpoint + .details_as::() + .map_err(|_| invalid("checkpoint details do not match this batch schema")) + }) + .transpose()?; + process_batch(&store, &batch, received, ctx.attempt_number, |checkpoint| { + let ctx = ctx.clone(); + async move { + let response = ctx.heartbeat(checkpoint).await?; + if !response.heartbeat_recorded || response.should_stop() { + return Err(invalid( + "checkpoint was not acknowledged or the attempt must stop", + )); + } + Ok(()) + } + }) + .await + } + }); + worker.register_typed_workflow("checkpointed-batch", |ctx, batch: Batch| async move { + ctx.activity_typed_with_options::<_, Value>( + "checkpointed-batch", + ActivityOptions::new() + .retry_policy( + ActivityRetryPolicy::new(2).backoff_intervals([Duration::from_secs(1)]), + ) + .heartbeat_timeout(Duration::from_secs(10)) + .start_to_close_timeout(Duration::from_secs(30)) + .schedule_to_close_timeout(Duration::from_secs(60)), + batch, + ) + .await + }); + worker.register().await?; + let handle = client + .start_workflow( + "checkpointed-batch", + &queue, + &queue, + Batch { + business_id: format!("demo-{}", Uuid::new_v4()), + items: vec!["first".into(), "second".into(), "third".into()], + failure, + }, + ) + .await?; + let watcher = handle.clone(); + worker + .run_until(async move { + let _ = tokio::time::timeout(Duration::from_secs(70), async move { + loop { + if watcher.describe().await.is_ok_and(|run| run.is_terminal()) { + break; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + }) + .await; + }) + .await?; + let result = handle + .result(WorkflowResultOptions { + poll_interval: Duration::from_millis(250), + timeout: Duration::from_secs(3), + }) + .await?; + println!("{result}"); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + struct Fixture { + store: ReceiptStore, + } + impl Fixture { + fn new() -> Self { + let directory = + std::env::temp_dir().join(format!("checkpoint-batch-{}", Uuid::new_v4())); + std::fs::create_dir(&directory).unwrap(); + Self { + store: ReceiptStore(directory), + } + } + } + impl Drop for Fixture { + fn drop(&mut self) { + std::fs::remove_dir_all(&self.store.0).unwrap(); + } + } + fn batch(failure: FailurePoint) -> Batch { + Batch { + business_id: "business-batch".into(), + items: vec!["first".into(), "second".into(), "third".into()], + failure, + } + } + + #[tokio::test] + async fn retries_reconcile_commit_and_resume_only_acknowledged_items() { + for point in [ + FailurePoint::BeforeCommit, + FailurePoint::AfterCommit, + FailurePoint::AfterCheckpoint, + ] { + let fixture = Fixture::new(); + let batch = batch(point); + let mut persisted = None; + let first = process_batch(&fixture.store, &batch, None, 1, |checkpoint| { + persisted = Some(checkpoint); + std::future::ready(Ok(())) + }) + .await; + assert!(first.is_err()); + assert_eq!( + persisted.as_ref().map(|value| value.next_item), + if point == FailurePoint::AfterCheckpoint { + Some(1) + } else { + None + } + ); + assert_eq!( + std::fs::read_dir(&fixture.store.0).unwrap().count(), + if point == FailurePoint::BeforeCommit { + 0 + } else { + 1 + } + ); + let mut resumed = Vec::new(); + let result = process_batch(&fixture.store, &batch, persisted, 2, |checkpoint| { + resumed.push(checkpoint.next_item); + std::future::ready(Ok(())) + }) + .await + .unwrap(); + assert_eq!( + resumed, + if point == FailurePoint::AfterCheckpoint { + vec![2, 3] + } else { + vec![1, 2, 3] + } + ); + assert_eq!(result["completed"], 3); + assert_eq!(std::fs::read_dir(&fixture.store.0).unwrap().count(), 3); + for index in 0..batch.items.len() { + assert!(!fixture.store.commit_or_reconcile(&batch, index).unwrap()); + } + } + } + + #[tokio::test] + async fn lost_ack_response_reconciles_with_or_without_a_committed_checkpoint() { + for committed in [false, true] { + let fixture = Fixture::new(); + let batch = batch(FailurePoint::AfterCheckpoint); + let mut persisted = None; + assert!( + process_batch(&fixture.store, &batch, None, 1, |checkpoint| { + if committed { + persisted = Some(checkpoint); + } + std::future::ready(Err(Error::Timeout)) + }) + .await + .is_err() + ); + let result = process_batch(&fixture.store, &batch, persisted, 2, |_| { + std::future::ready(Ok(())) + }) + .await + .unwrap(); + assert_eq!(result["completed"], 3); + assert_eq!(std::fs::read_dir(&fixture.store.0).unwrap().count(), 3); + } + } + + #[tokio::test] + async fn mismatched_checkpoint_or_unbounded_batch_is_refused_before_provider_access() { + let fixture = Fixture::new(); + let batch = batch(FailurePoint::AfterCommit); + for checkpoint in [ + Checkpoint { + batch_sha256: "foreign".into(), + next_item: 1, + }, + Checkpoint { + batch_sha256: batch.fingerprint().unwrap(), + next_item: 4, + }, + ] { + let acknowledged = std::cell::Cell::new(false); + assert!( + process_batch(&fixture.store, &batch, Some(checkpoint), 2, |_| { + acknowledged.set(true); + std::future::ready(Ok(())) + }) + .await + .is_err() + ); + assert!(!acknowledged.get()); + } + let mut changed = batch; + changed.items = vec!["item".into(); 17]; + assert!( + process_batch(&fixture.store, &changed, None, 2, |_| std::future::ready( + Ok(()) + )) + .await + .is_err() + ); + assert_eq!(std::fs::read_dir(&fixture.store.0).unwrap().count(), 0); + } + + #[test] + fn provider_refuses_conflicting_or_partial_receipts() { + let fixture = Fixture::new(); + let batch = batch(FailurePoint::AfterCommit); + assert!(fixture.store.commit_or_reconcile(&batch, 0).unwrap()); + let mut changed = batch.clone(); + changed.items[0] = "changed".into(); + assert!(fixture.store.commit_or_reconcile(&changed, 0).is_err()); + let path = std::fs::read_dir(&fixture.store.0) + .unwrap() + .next() + .unwrap() + .unwrap() + .path(); + std::fs::write(path, b"").unwrap(); + assert!(fixture.store.commit_or_reconcile(&batch, 0).is_err()); + } + + async fn assert_changed_business_input_refuses(changed_items: Vec) { + let fixture = Fixture::new(); + let original = batch(FailurePoint::AfterCommit); + process_batch(&fixture.store, &original, None, 2, |_| { + std::future::ready(Ok(())) + }) + .await + .unwrap(); + let inventory = || { + let mut files = std::fs::read_dir(&fixture.store.0) + .unwrap() + .map(|entry| { + let path = entry.unwrap().path(); + ( + path.file_name().unwrap().to_owned(), + std::fs::read(&path).unwrap(), + ) + }) + .collect::>(); + files.sort(); + files + }; + let before = inventory(); + let mut changed = original; + changed.items = changed_items; + let acknowledged = std::cell::Cell::new(false); + let result = process_batch(&fixture.store, &changed, None, 2, |_| { + acknowledged.set(true); + std::future::ready(Ok(())) + }) + .await; + assert!( + result.is_err(), + "changed business input must refuse without a checkpoint" + ); + assert!(!acknowledged.get()); + assert_eq!( + inventory(), + before, + "refusal must preserve all provider receipts" + ); + } + + #[tokio::test] + async fn appended_business_input_without_checkpoint_is_refused() { + assert_changed_business_input_refuses(vec![ + "first".into(), + "second".into(), + "third".into(), + "fourth".into(), + ]) + .await; + } + + #[tokio::test] + async fn truncated_business_input_without_checkpoint_is_refused() { + assert_changed_business_input_refuses(vec!["first".into(), "second".into()]).await; + } + + #[tokio::test] + async fn reordered_business_input_without_checkpoint_is_refused() { + assert_changed_business_input_refuses(vec![ + "second".into(), + "first".into(), + "third".into(), + ]) + .await; + } +} diff --git a/tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md b/tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md new file mode 100644 index 0000000..69c6296 --- /dev/null +++ b/tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md @@ -0,0 +1,52 @@ +# Activity checkpoint replacement acceptance + +`activity_checkpoint_server.rs` is an ignored integration qualifier for the +`activity_heartbeat_resume` v1 accessor. Compilation and its inert observer +tests do not establish an installed result. Run the ignored parent only after +the exact test, package, producer, compiled executable and isolated caller have +passed their required reviews. + +Use an ordinary registry-installed SDK containing the accessor, without a path, +Git or patch override. Copy this test unchanged into that consumer. Retain its +lockfile, downloaded crate checksum, test hash and compiled parent hash. The +qualifier executes children from the running Linux parent image and verifies +their SHA-256, length, device and inode before constructing a client. + +The reviewed caller supplies `DURABLE_WORKFLOW_LOCAL_ISOLATED=1`, a numeric +loopback `DURABLE_WORKFLOW_SERVER_URL`, a synthetic fixture token in +`DURABLE_WORKFLOW_AUTH_TOKEN`, and the selected published native version in +`DURABLE_WORKFLOW_EXPECT_SERVER_VERSION`. It owns the exact immutable runtime, +disposable database, bounded resources, evidence and cleanup. This test does not +start a Server or select an image. + +```bash +cargo test --locked --test activity_checkpoint_server \ + published_cold_replacement_preserves_acknowledged_activity_checkpoint \ + -- --ignored --exact --nocapture --test-threads=1 +``` + +The parent runs acknowledged and empty-checkpoint cases in separate workflow +runs. For each it creates three cold Worker processes with the same compiled +definition. Attempt 1 reports its claim, acknowledges typed progress when that +case calls for it, then stays pending until the parent kills and reaps it. +Attempt 2 receives the retry claim. While it remains live, the parent tries the +dead attempt's heartbeat and requires a producer refusal rather than accepting +a transport error as proof. Attempt 2 then reports one retryable failure. +Attempt 3 must receive the same original checkpoint, proving the stale write +did not replace it, and completes the activity and workflow. + +Assertions bind workflow/run/execution identity, distinct attempt IDs and PIDs, +typed details, exact parsed acknowledgement time, the complete received retry +policy and its original timeout values, one authored activity, three starts, +two retries, one completion and the retained committed-history prefix. The +empty case also checks that the earlier run's checkpoint does not leak into a +new execution. Child output is bounded while being read. Timeouts and failed +assertions kill and reap only children returned by this parent's `spawn`. + +Retain the structured `ACTIVITY_CHECKPOINT` observations and exact package, +runtime, configuration, process and independent cleanup evidence. A failure or +runner refusal remains NONPASS. This qualifier does not test unavailable older +producers, a committed write with a lost response, checkpoint pruning, embedded +PHP parity or external business-effect reconciliation. Those claims need +their own evidence under shared #218. A checkpoint never atomically commits an +external business effect. diff --git a/tests/activity_checkpoint_server.rs b/tests/activity_checkpoint_server.rs new file mode 100644 index 0000000..edaf52c --- /dev/null +++ b/tests/activity_checkpoint_server.rs @@ -0,0 +1,579 @@ +//! Source-only qualifier until an exact package, producer and caller are reviewed. +//! This test starts only its owned Worker children, never a Server. +use durable_workflow::{ + json, ActivityError, ActivityOptions, ActivityRetryPolicy, Client, Error, Value, Worker, + WorkflowResultOptions, +}; +use serde::Deserialize; +use sha2::{Digest, Sha256}; +use std::{ + io::{BufRead, Read, Write}, + process::{Child, Command, Stdio}, + sync::{mpsc, Arc, Mutex}, + thread, + time::{Duration, Instant}, +}; + +const CHILD: &str = "DURABLE_WORKFLOW_CHECKPOINT_CHILD"; +const QUEUE: &str = "DURABLE_WORKFLOW_CHECKPOINT_QUEUE"; +const PHASE: &str = "DURABLE_WORKFLOW_CHECKPOINT_PHASE"; +const CASE: &str = "DURABLE_WORKFLOW_CHECKPOINT_CASE"; +const IMAGE: &str = "DURABLE_WORKFLOW_CHECKPOINT_IMAGE"; +const REPORT: &str = "CHECKPOINT_WORKER "; +const STREAM_LIMIT: usize = 32 * 1024; + +#[derive(Debug, Deserialize, PartialEq)] +struct Progress { + cursor: u64, + checked: bool, + label: String, + optional: Option, +} + +fn details() -> Value { + json!({"cursor":7,"checked":true,"label":"acknowledged","optional":null}) +} + +fn executable_identity() -> Value { + #[cfg(target_os = "linux")] + { + use std::os::unix::fs::MetadataExt; + let mut file = std::fs::File::open("/proc/self/exe").unwrap(); + let metadata = file.metadata().unwrap(); + let mut digest = Sha256::new(); + let mut length = 0_u64; + let mut buffer = [0; 65536]; + loop { + let count = file.read(&mut buffer).unwrap(); + if count == 0 { + break; + } + length += count as u64; + digest.update(&buffer[..count]); + } + assert_eq!(length, metadata.len()); + json!({"sha256":format!("{:x}",digest.finalize()),"length":length, + "device":metadata.dev(),"inode":metadata.ino()}) + } + #[cfg(not(target_os = "linux"))] + panic!("the isolated qualifier requires Linux parent-image execution"); +} + +fn client() -> Client { + assert_eq!( + std::env::var("DURABLE_WORKFLOW_LOCAL_ISOLATED").as_deref(), + Ok("1") + ); + let origin = std::env::var("DURABLE_WORKFLOW_SERVER_URL").unwrap(); + assert!(origin + .strip_prefix("http://127.0.0.1:") + .and_then(|port| port.parse::().ok()) + .is_some_and(|port| port > 0)); + Client::builder(origin) + .namespace("default") + .timeout(Duration::from_secs(3)) + .token(Some(std::env::var("DURABLE_WORKFLOW_AUTH_TOKEN").unwrap())) + .build() + .unwrap() +} + +fn emit(value: Value) { + let text = value.to_string(); + assert!(text.len() <= 8192); + println!("{REPORT}{text}"); + std::io::stdout().flush().unwrap(); +} + +fn worker( + client: Client, + queue: &str, + phase: u64, + case: &str, + observations: Arc>>, +) -> Worker { + let mut worker = Worker::new(client, queue) + .worker_id(format!("{queue}-{phase}")) + .poll_timeout(Duration::from_millis(250)); + worker.register_workflow(queue, |ctx, _| async move { + ctx.activity_with_options( + "resume-batch", + ActivityOptions::new() + .retry_policy(ActivityRetryPolicy::new(3).backoff_intervals([Duration::ZERO; 2])) + .heartbeat_timeout(Duration::from_secs(5)) + .start_to_close_timeout(Duration::from_secs(30)) + .schedule_to_close_timeout(Duration::from_secs(90)), + json!([]), + ) + .await + }); + let checkpoint = case == "acknowledged"; + worker.register_activity("resume-batch", move |ctx, _| { + let observations = observations.clone(); + async move { + assert_eq!(ctx.attempt_number, phase); + assert_eq!(ctx.heartbeat_resume_version(), Some(1)); + let received = ctx.last_heartbeat_checkpoint()?.cloned(); + if phase == 1 || !checkpoint { + assert!(received.is_none()); + } else { + let received = received.as_ref().unwrap(); + assert_eq!(received.details(), &details()); + assert_eq!(received.contract_version(), 1); + assert_eq!(received.details_as::().unwrap(), Progress { + cursor:7,checked:true,label:"acknowledged".into(),optional:None, + }); + } + let mut observation = json!({"event":"claimed","pid":std::process::id(), + "phase":phase,"execution":ctx.activity_execution_id(),"attempt":ctx.activity_attempt_id, + "task":ctx.task_id,"owner":ctx.lease_owner,"workflow":ctx.workflow_id(),"run":ctx.run_id(), + "details":received.as_ref().map(|v|v.details()), + "at":received.as_ref().map(|v|v.acknowledged_at().to_rfc3339()), + "retry_policy":ctx.retry_info().policy()}); + if phase == 1 && checkpoint { + let ack = ctx.heartbeat(details()).await?; + assert!(ack.heartbeat_recorded && !ack.should_stop()); + observation["acknowledged_at"] = json!(ack.last_heartbeat_at.unwrap()); + assert!(ctx.last_heartbeat_checkpoint()?.is_none()); + } + observations.lock().unwrap().push(observation.clone()); + emit(observation); + if phase == 1 { + // The parent kills this process only after the flushed claim/ack. + std::future::pending::<()>().await; + } + if phase == 2 { + // Keep attempt 2 live while the parent tries attempt 1's stale write. + let mut command = Vec::new(); + std::io::stdin().lock().take(16).read_until(b'\n', &mut command).unwrap(); + assert_eq!(command, b"retry\n"); + assert_eq!(ctx.last_heartbeat_checkpoint()?.cloned(), received); + return Err(ActivityError::new("CheckpointProbe", "one declared retry").into()); + } + assert_eq!(phase, 3); + let ack = ctx.heartbeat(json!({"cursor":8})).await?; + assert!(ack.heartbeat_recorded && !ack.should_stop()); + assert_eq!(ctx.last_heartbeat_checkpoint()?.cloned(), received); + emit(json!({"event":"acknowledged","pid":std::process::id(),"phase":phase, + "execution":ctx.activity_execution_id(),"attempt":ctx.activity_attempt_id, + "at":ack.last_heartbeat_at.unwrap()})); + Ok(json!({"resumed":received.as_ref().map(|v|v.details()),"completed_through":8})) + } + }); + worker +} + +#[tokio::test] +#[ignore = "entrypoint only for the isolated checkpoint qualifier parent"] +async fn checkpoint_worker_entrypoint() { + assert_eq!(std::env::var(CHILD).as_deref(), Ok("1")); + let image = executable_identity(); + assert_eq!( + image, + serde_json::from_str::(&std::env::var(IMAGE).unwrap()).unwrap() + ); + // No client or credential access precedes the held running-image identity check. + let queue = std::env::var(QUEUE).unwrap(); + let phase: u64 = std::env::var(PHASE).unwrap().parse().unwrap(); + let case = std::env::var(CASE).unwrap(); + assert!(matches!(phase, 1..=3) && matches!(case.as_str(), "acknowledged" | "empty")); + let client = client(); + let observations = Arc::new(Mutex::new(Vec::new())); + let worker = worker(client.clone(), &queue, phase, &case, observations.clone()); + worker.register().await.unwrap(); + emit(json!({"event":"registered","pid":std::process::id(),"image":image,"phase":phase})); + tokio::time::timeout(Duration::from_secs(40), async { + for _ in 0..100 { + worker.run_once().await.unwrap(); + let callbacks = observations.lock().unwrap().len(); + if callbacks > 0 { + if phase == 2 + || client + .describe_workflow(&queue) + .await + .unwrap() + .is_terminal() + { + emit( + json!({"event":"finished","pid":std::process::id(),"phase":phase, + "callbacks":callbacks}), + ); + return; + } + } + } + panic!("bounded worker polling did not finish"); + }) + .await + .expect("checkpoint child exceeded its original bound"); + client + .deregister_worker_registration(&format!("{queue}-{phase}")) + .await + .unwrap(); +} + +struct OwnedChild { + child: Child, + lines: mpsc::Receiver, &'static str>>, + readers: Vec>, + registered: bool, + image: Value, +} +impl Drop for OwnedChild { + fn drop(&mut self) { + if self.child.try_wait().ok().flatten().is_none() { + let _ = self.child.kill(); + let _ = self.child.wait(); + } + // Reaping closes both pipes; readers never block on the channel. + for reader in self.readers.drain(..) { + let _ = reader.join(); + } + } +} + +fn read_stream(mut stream: impl Read, events: mpsc::Sender, &'static str>>) { + let mut total = 0; + let mut line = Vec::new(); + let mut buffer = [0; 4096]; + loop { + let count = match stream.read(&mut buffer) { + Ok(0) => break, + Ok(count) => count, + Err(_) => { + let _ = events.send(Err("owned child stream read failed")); + return; + } + }; + total += count; + if total > STREAM_LIMIT { + let _ = events.send(Err("owned child stream exceeded its byte bound")); + return; + } + for byte in &buffer[..count] { + line.push(*byte); + if *byte == b'\n' && events.send(Ok(std::mem::take(&mut line))).is_err() { + return; + } + } + } + if !line.is_empty() { + let _ = events.send(Ok(line)); + } +} + +impl OwnedChild { + fn spawn(queue: &str, phase: u64, case: &str, image: &Value) -> Self { + assert!(cfg!(target_os = "linux")); + let mut command = Command::new("/proc/self/exe"); + command + .args([ + "--ignored", + "--exact", + "checkpoint_worker_entrypoint", + "--nocapture", + "--test-threads=1", + "--quiet", + ]) + .env_clear() + .env(CHILD, "1") + .env(QUEUE, queue) + .env(PHASE, phase.to_string()) + .env(CASE, case) + .env(IMAGE, image.to_string()); + for name in [ + "DURABLE_WORKFLOW_LOCAL_ISOLATED", + "DURABLE_WORKFLOW_SERVER_URL", + "DURABLE_WORKFLOW_AUTH_TOKEN", + ] { + command.env(name, std::env::var(name).unwrap()); + } + let mut child = command + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + // An unbounded channel stores at most the two readers' fixed byte caps. + let (tx, lines) = mpsc::channel(); + let stdout = child.stdout.take().unwrap(); + let stderr = child.stderr.take().unwrap(); + let errors = tx.clone(); + let readers = vec![ + thread::spawn(move || read_stream(stdout, tx)), + thread::spawn(move || read_stream(stderr, errors)), + ]; + Self { + child, + lines, + readers, + registered: false, + image: image.clone(), + } + } + async fn event(&mut self, name: &str) -> Value { + let deadline = Instant::now() + Duration::from_secs(30); + loop { + match self.lines.try_recv() { + Ok(Ok(line)) => { + let text = String::from_utf8(line).unwrap(); + if let Some(at) = text.find(REPORT) { + let record: Value = + serde_json::from_str(text[at + REPORT.len()..].trim()).unwrap(); + assert_eq!(record["pid"], self.child.id()); + if record["event"] == "registered" { + assert_eq!(record["image"], self.image); + self.registered = true; + } + assert!(self.registered, "claim before bound registration"); + if record["event"] == name { + return record; + } + } + } + Ok(Err(error)) => panic!("{error}"), + Err(mpsc::TryRecvError::Disconnected) => { + panic!("child exited before required event") + } + Err(mpsc::TryRecvError::Empty) => { + assert!(Instant::now() < deadline, "owned child event timed out"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } + } + } + async fn finish(&mut self) { + let record = self.event("finished").await; + assert_eq!(record["callbacks"], 1); + let deadline = Instant::now() + Duration::from_secs(5); + loop { + if let Some(status) = self.child.try_wait().unwrap() { + assert!(status.success()); + for reader in self.readers.drain(..) { + reader.join().unwrap(); + } + for event in self.lines.try_iter() { + event.expect("bounded child output"); + } + return; + } + assert!(Instant::now() < deadline, "owned child did not exit"); + tokio::time::sleep(Duration::from_millis(10)).await; + } + } +} + +fn instant(value: &Value) -> chrono::DateTime { + chrono::DateTime::parse_from_rfc3339(value.as_str().unwrap()).unwrap() +} + +fn assert_stale_refusal( + result: durable_workflow::Result, + original: &Value, +) { + match result { + Err(Error::ActivityTaskRejected(rejection)) => { + assert_eq!(rejection.status, 409); + assert!(matches!( + rejection.reason.as_str(), + "stale_attempt" + | "attempt_closed" + | "activity_not_running" + | "task_not_leased" + | "attempt_not_found" + )); + assert_eq!(rejection.task_id, original["task"].as_str().unwrap()); + assert_eq!( + rejection.activity_attempt_id, + original["attempt"].as_str().unwrap() + ); + } + Ok(response) => { + assert!(!response.heartbeat_recorded && response.can_continue == Some(false)); + assert!(matches!( + response.reason.as_deref(), + Some("stale_attempt" | "attempt_closed" | "activity_not_running") + )); + } + other => panic!("expected a fenced stale-write refusal, got {other:?}"), + } +} + +fn assert_heartbeat_record( + event: &Value, + attempt: &Value, + expected_details: &Value, + acknowledged_at: &Value, +) { + assert_eq!(event["type"], "ActivityHeartbeatRecorded"); + let payload = &event["payload"]; + assert_eq!(payload["activity_execution_id"], attempt["execution"]); + assert_eq!(payload["activity_attempt_id"], attempt["attempt"]); + assert_eq!(payload["attempt_number"], attempt["phase"]); + assert_eq!(payload["activity_attempt"]["id"], attempt["attempt"]); + assert_eq!(payload["activity_attempt"]["task_id"], attempt["task"]); + assert_eq!(payload["activity_attempt"]["lease_owner"], attempt["owner"]); + assert_eq!(payload["progress"]["details"], *expected_details); + assert_eq!(instant(&payload["heartbeat_at"]), instant(acknowledged_at)); + assert_eq!( + instant(&payload["activity_attempt"]["last_heartbeat_at"]), + instant(acknowledged_at) + ); +} + +#[test] +fn heartbeat_observer_binds_nested_progress_identity_and_full_clock() { + let at = json!("2026-11-14T11:00:00.123456Z"); + let attempt = json!({"execution":"execution","attempt":"attempt-1","phase":1,"task":"task-1","owner":"worker"}); + // Tagged native 3.10 activity_heartbeats.rs retains progress.details and + // heartbeat_at; history export preserves that complete payload nesting. + let event = json!({"type":"ActivityHeartbeatRecorded","payload":{ + "activity_execution_id":"execution","activity_attempt_id":"attempt-1","attempt_number":1, + "progress":{"details":details()},"heartbeat_at":at, + "activity_attempt":{"id":"attempt-1","task_id":"task-1","lease_owner":"worker","last_heartbeat_at":at}}}); + assert_heartbeat_record(&event, &attempt, &details(), &at); + let predecessor = || assert_eq!(event["payload"]["details"], details()); + assert!( + std::panic::catch_unwind(predecessor).is_err(), + "old top-level-only lookup must fail on producer-shaped data" + ); + for (pointer, value) in [ + ("/payload/activity_execution_id", json!("foreign")), + ("/payload/activity_attempt_id", json!("other-attempt")), + ("/payload/progress/details/cursor", json!(99)), + ( + "/payload/heartbeat_at", + json!("2026-11-14T11:00:00.123457Z"), + ), + ] { + let mut changed = event.clone(); + *changed.pointer_mut(pointer).unwrap() = value; + assert!(std::panic::catch_unwind(|| assert_heartbeat_record( + &changed, + &attempt, + &details(), + &at + )) + .is_err()); + } +} + +#[test] +fn child_stream_reader_refuses_oversized_output_during_reads() { + let (send, received) = mpsc::channel(); + read_stream(std::io::Cursor::new(vec![b'x'; STREAM_LIMIT + 1]), send); + assert_eq!( + received.into_iter().collect::>(), + vec![Err("owned child stream exceeded its byte bound")] + ); + let (send, received) = mpsc::channel(); + read_stream(std::io::Cursor::new(b"one\ntwo"), send); + assert_eq!( + received.into_iter().collect::>(), + vec![Ok(b"one\n".to_vec()), Ok(b"two".to_vec())] + ); +} + +#[test] +fn stale_write_observer_refuses_transport_and_unrelated_errors() { + let original = json!({"task":"owned-task","attempt":"old-attempt"}); + let accepted = || { + serde_json::from_value( + json!({"heartbeat_recorded":false,"can_continue":false,"reason":"stale_attempt"}), + ) + .unwrap() + }; + assert_stale_refusal(Ok(accepted()), &original); + assert!(std::panic::catch_unwind(|| assert_stale_refusal( + Err(Error::Codec("not a producer refusal".into())), + &original + )) + .is_err()); + let wrong = serde_json::from_value( + json!({"heartbeat_recorded":true,"can_continue":false,"reason":"stale_attempt"}), + ) + .unwrap(); + assert!(std::panic::catch_unwind(|| assert_stale_refusal(Ok(wrong), &original)).is_err()); +} + +#[tokio::test] +#[ignore = "requires reviewed exact published SDK/Server/caller; starts no Server"] +async fn published_cold_replacement_preserves_acknowledged_activity_checkpoint() { + let client = client(); + let cluster = client.cluster_info().await.unwrap(); + assert_eq!(cluster["implementation"], "rust"); + assert_eq!( + cluster["version"], + std::env::var("DURABLE_WORKFLOW_EXPECT_SERVER_VERSION").unwrap() + ); + let image = executable_identity(); + for case in ["acknowledged", "empty"] { + tokio::time::timeout(Duration::from_secs(100),async { + let queue=format!("rust-checkpoint-{}",durable_workflow::Uuid::new_v4()); + let registration=worker(client.clone(),&queue,0,case,Arc::new(Mutex::new(Vec::new()))); + registration.register().await.unwrap(); + let handle=client.start_workflow(&queue,&queue,&queue,json!([])).await.unwrap(); + let run=handle.run_id.as_deref().unwrap().to_string(); + let mut first=OwnedChild::spawn(&queue,1,case,&image); + let original=first.event("claimed").await; + assert_eq!(original["workflow"],queue);assert_eq!(original["run"],run); + tokio::task::spawn_blocking(move || { + first.child.kill().unwrap(); + let killed=first.child.wait().unwrap(); + assert!(!killed.success()); + }).await.unwrap(); + let mut second=OwnedChild::spawn(&queue,2,case,&image); + let replacement=second.event("claimed").await; + assert_ne!(original["pid"],replacement["pid"]); + for key in ["workflow","run","execution","retry_policy"] {assert_eq!(original[key],replacement[key]);} + assert!(original["execution"].as_str().is_some_and(|s|!s.is_empty())); + assert_ne!(original["attempt"],replacement["attempt"]); + assert_eq!(original["retry_policy"]["max_attempts"],3); + assert_eq!(original["retry_policy"]["backoff_seconds"],json!([0,0])); + assert_eq!(original["retry_policy"]["heartbeat_timeout"],5); + assert_eq!(original["retry_policy"]["start_to_close_timeout"],30); + assert_eq!(original["retry_policy"]["schedule_to_close_timeout"],90); + let prefix=client.export_workflow_history(&queue,&run).await.unwrap(); + // A dead attempt cannot replace the checkpoint while attempt 2 owns it. + let stale=client.heartbeat_activity_task(original["task"].as_str().unwrap(), + original["attempt"].as_str().unwrap(),original["owner"].as_str().unwrap(),json!({"cursor":99})).await; + assert_stale_refusal(stale,&original); + second.child.stdin.as_mut().unwrap().write_all(b"retry\n").unwrap();second.finish().await; + let mut third=OwnedChild::spawn(&queue,3,case,&image); + let retained=third.event("claimed").await; + for key in ["workflow","run","execution","retry_policy","details","at"] {assert_eq!(retained[key],replacement[key]);} + assert_ne!(retained["pid"],replacement["pid"]);assert_ne!(retained["pid"],original["pid"]); + assert_ne!(retained["attempt"],replacement["attempt"]);assert_ne!(retained["attempt"],original["attempt"]); + if case=="acknowledged" { + assert_eq!(replacement["details"],details()); + assert_eq!(instant(&replacement["at"]),instant(&original["acknowledged_at"])); + } else {assert!(replacement["details"].is_null() && replacement["at"].is_null());} + let completed_ack=third.event("acknowledged").await; + assert_eq!(completed_ack["execution"],retained["execution"]); + assert_eq!(completed_ack["attempt"],retained["attempt"]); + third.finish().await; + assert_eq!(handle.describe_selected_run().await.unwrap().status.as_deref(),Some("completed")); + assert_eq!(handle.result_selected_run(WorkflowResultOptions::default()).await.unwrap(), + json!({"resumed":if case=="acknowledged" {details()}else{Value::Null},"completed_through":8})); + let history=client.export_workflow_history(&queue,&run).await.unwrap(); + assert_eq!(history["workflow"]["instance_id"],queue);assert_eq!(history["workflow"]["run_id"],run); + assert_eq!(history["workflow"]["details_state"],"retained"); + let events=history["history_events"].as_array().unwrap(); + let prefix=prefix["history_events"].as_array().unwrap();assert_eq!(&events[..prefix.len()],prefix); + let count=|kind:&str| events.iter().filter(|e|e["type"]==kind).count(); + assert_eq!(count("ActivityScheduled"),1);assert_eq!(count("ActivityStarted"),3); + assert_eq!(count("ActivityRetryScheduled"),2);assert_eq!(count("ActivityCompleted"),1); + assert_eq!(count("WorkflowCompleted"),1); + let recorded:Vec<_>=events.iter().filter(|e|e["type"]=="ActivityHeartbeatRecorded").collect(); + assert_eq!(recorded.len(),if case=="acknowledged" {2}else{1}); + if case=="acknowledged" { + assert_heartbeat_record(recorded[0],&original,&details(),&original["acknowledged_at"]); + } + assert_heartbeat_record(recorded.last().unwrap(),&retained,&json!({"cursor":8}),&completed_ack["at"]); + client.deregister_worker_registration(&format!("{queue}-0")).await.unwrap(); + client.deregister_worker_registration(&format!("{queue}-1")).await.unwrap(); + let report=json!({"case":case,"original":original,"replacement":replacement,"retained":retained, + "worker1_killed":true,"stale_write_refused":true,"history":history}); + let report=report.to_string();assert!(report.len()<=512*1024);println!("ACTIVITY_CHECKPOINT {report}"); + }).await.expect("checkpoint case exceeded its original budget"); + } +}