From 7ef43e28c0dc67006519e6d2a45d09d71b935138 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:20:39 +0000 Subject: [PATCH 1/6] Prepare cold replacement checkpoint acceptance --- tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md | 52 +++ tests/activity_checkpoint_server.rs | 507 ++++++++++++++++++++++++ 2 files changed, 559 insertions(+) create mode 100644 tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md create mode 100644 tests/activity_checkpoint_server.rs diff --git a/tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md b/tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md new file mode 100644 index 0000000..10d1d9f --- /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 two 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..7207944 --- /dev/null +++ b/tests/activity_checkpoint_server.rs @@ -0,0 +1,507 @@ +//! 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); + 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:?}"), + } +} + +#[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); + first.child.kill().unwrap();let killed=first.child.wait().unwrap();assert!(!killed.success()); + 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());} + 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!(recorded.iter().all(|e|e["payload"]["details"]["cursor"]!=99)); + if case=="acknowledged" {assert!(recorded.iter().any(|e|e["payload"]["details"]==details()));} + 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"); + } +} From 17db8a2528486b66d5c5a58d9ddfb95a63cd4e70 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:28:44 +0000 Subject: [PATCH 2/6] Reap the killed qualifier child outside the async runtime --- tests/activity_checkpoint_server.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/tests/activity_checkpoint_server.rs b/tests/activity_checkpoint_server.rs index 7207944..d6aaa5e 100644 --- a/tests/activity_checkpoint_server.rs +++ b/tests/activity_checkpoint_server.rs @@ -454,7 +454,11 @@ async fn published_cold_replacement_preserves_acknowledged_activity_checkpoint() 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); - first.child.kill().unwrap();let killed=first.child.wait().unwrap();assert!(!killed.success()); + 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"]); From 0133bacb1af1c1fdc15b295da95d68a921b9a5d3 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:37:52 +0000 Subject: [PATCH 3/6] Bind checkpoint history proof to producer progress payloads --- tests/activity_checkpoint_server.rs | 72 ++++++++++++++++++++++++++++- 1 file changed, 70 insertions(+), 2 deletions(-) diff --git a/tests/activity_checkpoint_server.rs b/tests/activity_checkpoint_server.rs index d6aaa5e..edaf52c 100644 --- a/tests/activity_checkpoint_server.rs +++ b/tests/activity_checkpoint_server.rs @@ -153,6 +153,9 @@ fn worker( 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})) } }); @@ -395,6 +398,65 @@ fn assert_stale_refusal( } } +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(); @@ -485,6 +547,9 @@ async fn published_cold_replacement_preserves_acknowledged_activity_checkpoint() 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(), @@ -499,8 +564,11 @@ async fn published_cold_replacement_preserves_acknowledged_activity_checkpoint() 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!(recorded.iter().all(|e|e["payload"]["details"]["cursor"]!=99)); - if case=="acknowledged" {assert!(recorded.iter().any(|e|e["payload"]["details"]==details()));} + 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, From fe9cbe38ffd92e9232d94b17a37c0a3e296a03d4 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:38:38 +0000 Subject: [PATCH 4/6] Keep checkpoint acceptance guidance independent of observer count --- tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md b/tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md index 10d1d9f..69c6296 100644 --- a/tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md +++ b/tests/ACTIVITY_CHECKPOINT_ACCEPTANCE.md @@ -1,7 +1,7 @@ # Activity checkpoint replacement acceptance `activity_checkpoint_server.rs` is an ignored integration qualifier for the -`activity_heartbeat_resume` v1 accessor. Compilation and its two inert observer +`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. From bfd890e67160d69500a2c982802df80caeb33097 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 12:20:11 +0000 Subject: [PATCH 5/6] Add a bounded checkpointed batch example with commit-boundary controls --- README.md | 5 + examples/CHECKPOINTED_BATCH.md | 67 +++++ examples/checkpointed_batch.rs | 452 +++++++++++++++++++++++++++++++++ 3 files changed, 524 insertions(+) create mode 100644 examples/CHECKPOINTED_BATCH.md create mode 100644 examples/checkpointed_batch.rs 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..3f8547b --- /dev/null +++ b/examples/CHECKPOINTED_BATCH.md @@ -0,0 +1,67 @@ +# 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. A repeated key must contain the identical receipt. 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..8c2d4e9 --- /dev/null +++ b/examples/checkpointed_batch.rs @@ -0,0 +1,452 @@ +//! 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, business_id: &str, index: usize, item: &str) -> 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 key = serde_json::to_vec(&(business_id, index))?; + let path = self.0.join(format!("{:x}.json", Sha256::digest(&key))); + let receipt = serde_json::to_vec(&json!({"key":[business_id,index],"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()); + } + store.commit_or_reconcile(&batch.business_id, index, &batch.items[index])?; + 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. + std::fs::create_dir(&directory).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, item) in batch.items.iter().enumerate() { + assert!(!fixture + .store + .commit_or_reconcile(&batch.business_id, index, item) + .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(); + assert!(fixture + .store + .commit_or_reconcile("batch", 0, "first") + .unwrap()); + assert!(fixture + .store + .commit_or_reconcile("batch", 0, "changed") + .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, "first") + .is_err()); + } +} From 80cb25b645d30e2bfd8827f9526383c63a49abf4 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 12:28:12 +0000 Subject: [PATCH 6/6] Bind provider receipts to complete batch input --- examples/CHECKPOINTED_BATCH.md | 4 +- examples/checkpointed_batch.rs | 119 +++++++++++++++++++++++++++------ 2 files changed, 100 insertions(+), 23 deletions(-) diff --git a/examples/CHECKPOINTED_BATCH.md b/examples/CHECKPOINTED_BATCH.md index 3f8547b..2dc1fd7 100644 --- a/examples/CHECKPOINTED_BATCH.md +++ b/examples/CHECKPOINTED_BATCH.md @@ -38,7 +38,9 @@ batch or outside this batch is refused. 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. A repeated key must contain the identical receipt. Changed +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 diff --git a/examples/checkpointed_batch.rs b/examples/checkpointed_batch.rs index 8c2d4e9..7a6f5e4 100644 --- a/examples/checkpointed_batch.rs +++ b/examples/checkpointed_batch.rs @@ -77,12 +77,19 @@ impl Batch { struct ReceiptStore(PathBuf); impl ReceiptStore { - fn commit_or_reconcile(&self, business_id: &str, index: usize, item: &str) -> Result { + 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 key = serde_json::to_vec(&(business_id, index))?; + 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":[business_id,index],"item":item}))?; + 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)?; @@ -148,7 +155,11 @@ where if fail(FailurePoint::BeforeCommit) { return Err(planned_failure()); } - store.commit_or_reconcile(&batch.business_id, index, &batch.items[index])?; + 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()); } @@ -186,7 +197,11 @@ async fn main() -> Result<()> { )); } // Refuse an existing directory so this demo cannot mix with application data. - std::fs::create_dir(&directory).map_err(provider_io)?; + 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") @@ -354,11 +369,8 @@ mod tests { ); assert_eq!(result["completed"], 3); assert_eq!(std::fs::read_dir(&fixture.store.0).unwrap().count(), 3); - for (index, item) in batch.items.iter().enumerate() { - assert!(!fixture - .store - .commit_or_reconcile(&batch.business_id, index, item) - .unwrap()); + for index in 0..batch.items.len() { + assert!(!fixture.store.commit_or_reconcile(&batch, index).unwrap()); } } } @@ -429,14 +441,11 @@ mod tests { #[test] fn provider_refuses_conflicting_or_partial_receipts() { let fixture = Fixture::new(); - assert!(fixture - .store - .commit_or_reconcile("batch", 0, "first") - .unwrap()); - assert!(fixture - .store - .commit_or_reconcile("batch", 0, "changed") - .is_err()); + 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() @@ -444,9 +453,75 @@ mod tests { .unwrap() .path(); std::fs::write(path, b"").unwrap(); - assert!(fixture - .store - .commit_or_reconcile("batch", 0, "first") - .is_err()); + 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; } }