From 8ac93adae31800212461798e445d64a4da3fe268 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:01:56 +0000 Subject: [PATCH 1/3] Expose negotiated acknowledged activity checkpoints --- CHANGELOG.md | 7 + README.md | 45 ++++ src/activity_checkpoint.rs | 142 +++++++++++++ src/cooperative_cancellation.rs | 2 + src/lib.rs | 53 ++++- src/local_activity.rs | 1 + src/testing/runner.rs | 1 + src/worker_session.rs | 5 + tests/activity_checkpoint.rs | 366 ++++++++++++++++++++++++++++++++ 9 files changed, 621 insertions(+), 1 deletion(-) create mode 100644 src/activity_checkpoint.rs create mode 100644 tests/activity_checkpoint.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index 192e810e..13be16f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,12 @@ # Changelog +## Unreleased + +- Service activity contexts expose negotiated, immutable acknowledged-heartbeat + checkpoint details and a typed acknowledgement time. Unsupported claims stay + unavailable, separately from a supported claim with no checkpoint. Heartbeat + writes do not optimistically advance the received snapshot. + ## 4.17.0 - Workflow code can use `ctx.logger()` to send new-execution observations to diff --git a/README.md b/README.md index 90e8bb85..0c738673 100644 --- a/README.md +++ b/README.md @@ -711,6 +711,51 @@ deadlines can end execution earlier. Local activities and direct `run_activity` tests have no received Server snapshot and report unknown policy; fresh `run_workflow` tests expose the modeled snapshot recorded in their history. +### Acknowledged activity checkpoints + +A service activity claim can carry a read-only checkpoint from an earlier attempt +of the same logical execution. The Worker advertises `activity_heartbeat_resume` +and accepts its version-1 Server contract. `ctx.last_heartbeat_checkpoint()` returns +`Ok(Some(checkpoint))` for received details, `Ok(None)` when the supported claim +explicitly has no checkpoint, and `Err(ActivityHeartbeatResumeUnavailable)` when +support or the participating claim fields are unavailable. Older Servers, unknown +contract versions, inline local activities and Server-free tests are unavailable; +they do not invent an empty checkpoint. + +`last_heartbeat_details()` and `last_heartbeat_at()` expose the same snapshot. +`checkpoint.details_as::()` deserializes application JSON, while +`checkpoint.acknowledged_at()` retains a typed UTC timestamp and its fractional +precision. `heartbeat_resume_version()` and `checkpoint.contract_version()` refer +to the negotiated protocol; v1 supplies no checkpoint write-sequence number. + +```rust +# use durable_workflow::{json, ActivityError, Client, Worker}; +# use serde::Deserialize; +# #[derive(Deserialize)] +# struct Progress { next_item: u64 } +# let mut worker = Worker::new(Client::new("http://localhost:8080")?, "batch"); +worker.register_activity("batch", |ctx, _| async move { + let next_item = match ctx.last_heartbeat_checkpoint() { + Ok(Some(checkpoint)) => checkpoint.details_as::()?.next_item, + Ok(None) => 0, + Err(_) => return Err(ActivityError::new( + "CheckpointUnavailable", "this activity requires acknowledged resume support" + ).non_retryable(true).into()), + }; + Ok(json!({"next_item": next_item})) +}); +# Ok::<(), durable_workflow::Error>(()) +``` + +The received snapshot stays immutable throughout the invocation. Calling +`heartbeat(...)` does not advance it, whether the write is acknowledged, rejected, +stale, or its response is uncertain. A later attempt receives whatever checkpoint +the Server retained; a lost response may follow a durable commit. Server v1 +resumes only recorded non-null details. Receipt of a checkpoint does not atomically +commit an external business effect. Continue using stable application idempotency +keys, provider reconciliation and external fencing when effects can repeat. +Original retry policies and timeout budgets remain unchanged. + 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/src/activity_checkpoint.rs b/src/activity_checkpoint.rs new file mode 100644 index 00000000..1c3aa8ef --- /dev/null +++ b/src/activity_checkpoint.rs @@ -0,0 +1,142 @@ +//! Read-only acknowledged heartbeat snapshots supplied by a negotiated claim. + +use crate::{Error, Result, Value}; +use chrono::{DateTime, Utc}; +use serde::de::DeserializeOwned; +use std::fmt; + +pub(super) const CAPABILITY: &str = "activity_heartbeat_resume"; + +/// This claim does not provide the supported heartbeat-resume v1 contract. +/// +/// This is distinct from a supported claim that explicitly reports no checkpoint. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct ActivityHeartbeatResumeUnavailable; + +impl fmt::Display for ActivityHeartbeatResumeUnavailable { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str("activity_heartbeat_resume v1 is unavailable for this claim") + } +} + +impl std::error::Error for ActivityHeartbeatResumeUnavailable {} + +impl From for Error { + fn from(value: ActivityHeartbeatResumeUnavailable) -> Self { + Self::Codec(value.to_string()) + } +} + +/// An immutable checkpoint received with this activity attempt's claim. +/// +/// Heartbeat writes never change this snapshot, including when a response is +/// rejected or lost. A later claim may carry a newer acknowledged checkpoint. +/// A checkpoint does not atomically commit an external business effect. +#[derive(Clone, PartialEq)] +pub struct ActivityHeartbeatCheckpoint { + details: Value, + acknowledged_at: DateTime, +} + +impl fmt::Debug for ActivityHeartbeatCheckpoint { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("ActivityHeartbeatCheckpoint") + .field("details", &"") + .field("acknowledged_at", &self.acknowledged_at) + .finish() + } +} + +impl ActivityHeartbeatCheckpoint { + /// The acknowledged details, without defaults or mutation. + pub fn details(&self) -> &Value { + &self.details + } + + /// Deserialize the received JSON details into an application type. + pub fn details_as(&self) -> serde_json::Result { + serde_json::from_value(self.details.clone()) + } + + /// The producer's acknowledgement time, preserving fractional precision. + pub fn acknowledged_at(&self) -> &DateTime { + &self.acknowledged_at + } + + /// The negotiated contract version, not a checkpoint write sequence. + /// Server v1 does not supply a per-checkpoint sequence number. + pub fn contract_version(&self) -> u64 { + 1 + } +} + +#[derive(Clone, Debug, Default)] +pub(super) enum HeartbeatResume { + #[default] + Unavailable, + Available(Option), +} + +impl HeartbeatResume { + pub(super) fn from_poll(wire: &Value, namespace: &str) -> Result { + let capability = &wire["server_capabilities"][CAPABILITY]; + if capability["supported"] != true || capability["version"].as_u64() != Some(1) { + return Ok(Self::Unavailable); + } + let task = &wire["task"]; + let (details, at) = ( + task.get("last_heartbeat_details"), + task.get("last_heartbeat_at"), + ); + // Older/nonparticipating claims omit both fields. An advertisement alone + // does not invent a supported empty checkpoint for such a delivery. + if details.is_none() && at.is_none() { + return Ok(Self::Unavailable); + } + match (details, at) { + (Some(Value::Null), Some(Value::Null)) => Ok(Self::Available(None)), + (Some(details), Some(Value::String(at))) if !details.is_null() => { + let identity = + |field: &str| task[field].as_str().is_some_and(|value| !value.is_empty()); + if !identity("activity_execution_id") + || !identity("activity_attempt_id") + || !identity("workflow_id") + || !identity("run_id") + || task["namespace"].as_str() != Some(namespace) + || !task["attempt_number"].as_u64().is_some_and(|n| n > 1) + { + return Err(invalid()); + } + let acknowledged_at = DateTime::parse_from_rfc3339(at) + .map_err(|_| invalid())? + .with_timezone(&Utc); + Ok(Self::Available(Some(ActivityHeartbeatCheckpoint { + details: details.clone(), + acknowledged_at, + }))) + } + _ => Err(invalid()), + } + } + + pub(super) fn version(&self) -> Option { + match self { + Self::Unavailable => None, + Self::Available(_) => Some(1), + } + } + + pub(super) fn checkpoint( + &self, + ) -> std::result::Result, ActivityHeartbeatResumeUnavailable> + { + match self { + Self::Unavailable => Err(ActivityHeartbeatResumeUnavailable), + Self::Available(checkpoint) => Ok(checkpoint.as_ref()), + } + } +} + +fn invalid() -> Error { + Error::Codec("activity_heartbeat_resume v1 claim has an invalid checkpoint".into()) +} diff --git a/src/cooperative_cancellation.rs b/src/cooperative_cancellation.rs index 8ae67bfd..ecef0660 100644 --- a/src/cooperative_cancellation.rs +++ b/src/cooperative_cancellation.rs @@ -1972,6 +1972,7 @@ impl Worker { worker_session, hydration_claim, retry_info, + heartbeat_resume, } = invocation; validate_activity_task_payloads(task)?; let handler = self @@ -1987,6 +1988,7 @@ impl Worker { activity_type: task.activity_type.clone(), attempt_number: task.attempt_number, retry_info, + heartbeat_resume, activity_execution_id: task.activity_execution_id.clone(), workflow_id: task.workflow_id.clone(), run_id: task.run_id.clone(), diff --git a/src/lib.rs b/src/lib.rs index 7324402a..73227780 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,5 +1,6 @@ #![doc = include_str!("../README.md")] +mod activity_checkpoint; mod activity_error; mod activity_hydration; mod activity_lease; @@ -39,6 +40,7 @@ mod workflow_hydration; mod workflow_logging; pub use workflow_logging::{WorkflowLogger, WorkflowReplayState}; +pub use activity_checkpoint::{ActivityHeartbeatCheckpoint, ActivityHeartbeatResumeUnavailable}; pub use activity_error::ActivityError; pub use activity_retry::{ActivityAttemptLimit, ActivityRetryInfo}; pub use cancellation_context::{ @@ -5006,7 +5008,13 @@ impl Client { .authority .refresh_delivery_session(&mut wire["task"])?; } - Ok(serde_json::from_value(wire)?) + let heartbeat_resume = + activity_checkpoint::HeartbeatResume::from_poll(&wire, &self.namespace)?; + let mut data: worker_session::SessionPollResponse = serde_json::from_value(wire)?; + if let Some(task) = data.task.as_mut() { + task.heartbeat_resume = heartbeat_resume; + } + Ok(data) }; let mut data: worker_session::SessionPollResponse = match lease.as_ref() { Some(lease) => lease.authority.protect(loading).await?, @@ -8811,6 +8819,7 @@ impl Worker { Some(DURABLE_SELECTION_CAPABILITY.to_string()), Some(MEMO_UPSERTS_CAPABILITY.to_string()), Some(TYPED_SEARCH_ATTRIBUTES_CAPABILITY.to_string()), + Some(activity_checkpoint::CAPABILITY.to_string()), self.client .local_activities_enabled .then(|| "local_activities".to_string()), @@ -10564,6 +10573,7 @@ impl Worker { worker_session, hydration_claim, retry_info, + heartbeat_resume, } = invocation; validate_activity_task_payloads(&task)?; if worker_session.is_some() { @@ -10611,6 +10621,7 @@ impl Worker { activity_type: task.activity_type, attempt_number: task.attempt_number, retry_info, + heartbeat_resume, activity_execution_id: task.activity_execution_id, workflow_id: task.workflow_id, run_id: task.run_id, @@ -16950,6 +16961,7 @@ pub struct ActivityContext { pub activity_type: String, pub attempt_number: u64, retry_info: ActivityRetryInfo, + heartbeat_resume: activity_checkpoint::HeartbeatResume, activity_execution_id: Option, workflow_id: Option, run_id: Option, @@ -16961,6 +16973,45 @@ pub struct ActivityContext { } impl ActivityContext { + /// Negotiated resume contract version for this claim; unavailable is `None`. + /// This is not a checkpoint write sequence. + pub fn heartbeat_resume_version(&self) -> Option { + self.heartbeat_resume.version() + } + + /// The read-only acknowledged checkpoint supplied by this claim. + /// + /// `Ok(None)` means the supported claim explicitly has no checkpoint. + /// `Err(ActivityHeartbeatResumeUnavailable)` means the v1 contract was not + /// supplied. Inline local activities and Server-free tests are unavailable. + /// Heartbeat writes never change this received snapshot. Acknowledgement + /// does not atomically commit an external business effect. + pub fn last_heartbeat_checkpoint( + &self, + ) -> std::result::Result, ActivityHeartbeatResumeUnavailable> + { + self.heartbeat_resume.checkpoint() + } + + /// Last acknowledged details, with unavailable distinct from no checkpoint. + pub fn last_heartbeat_details( + &self, + ) -> std::result::Result, ActivityHeartbeatResumeUnavailable> { + Ok(self + .last_heartbeat_checkpoint()? + .map(|value| value.details())) + } + + /// Producer acknowledgement time, with full fractional precision. + pub fn last_heartbeat_at( + &self, + ) -> std::result::Result>, ActivityHeartbeatResumeUnavailable> + { + Ok(self + .last_heartbeat_checkpoint()? + .map(|value| value.acknowledged_at())) + } + /// Logical Server activity execution ID, stable across that activity's retries. /// /// This is the received claim identity, including for worker-session activities. diff --git a/src/local_activity.rs b/src/local_activity.rs index 61cfc612..ce63415d 100644 --- a/src/local_activity.rs +++ b/src/local_activity.rs @@ -461,6 +461,7 @@ impl Worker { activity_type: activity_type.clone(), attempt_number: number, retry_info: crate::ActivityRetryInfo::new(number, None), + heartbeat_resume: Default::default(), activity_execution_id: None, workflow_id: task.workflow_id.clone(), run_id: task.run_id.clone(), diff --git a/src/testing/runner.rs b/src/testing/runner.rs index 05394dd3..18bfeee7 100644 --- a/src/testing/runner.rs +++ b/src/testing/runner.rs @@ -424,6 +424,7 @@ async fn activity_attempt( activity_type: name.into(), attempt_number: attempt, retry_info, + heartbeat_resume: Default::default(), activity_execution_id: None, workflow_id: None, run_id: None, diff --git a/src/worker_session.rs b/src/worker_session.rs index 1d53942c..f085519b 100644 --- a/src/worker_session.rs +++ b/src/worker_session.rs @@ -269,12 +269,15 @@ pub(super) struct SessionActivityTask { pub(super) hydration_claim: Option, #[serde(skip)] pub(super) activity_lease: Option>, + #[serde(skip)] + pub(super) heartbeat_resume: crate::activity_checkpoint::HeartbeatResume, } pub(super) struct ActivityInvocation { pub(super) worker_session: Option, pub(super) hydration_claim: Option, pub(super) retry_info: crate::ActivityRetryInfo, + pub(super) heartbeat_resume: crate::activity_checkpoint::HeartbeatResume, } #[cfg(test)] @@ -284,6 +287,7 @@ impl ActivityInvocation { worker_session: None, hydration_claim: None, retry_info: crate::ActivityRetryInfo::new(attempt_number, None), + heartbeat_resume: Default::default(), } } } @@ -297,6 +301,7 @@ impl SessionActivityTask { self.task.attempt_number, self.retry_policy.clone(), ), + heartbeat_resume: self.heartbeat_resume.clone(), } } } diff --git a/tests/activity_checkpoint.rs b/tests/activity_checkpoint.rs new file mode 100644 index 00000000..794032d3 --- /dev/null +++ b/tests/activity_checkpoint.rs @@ -0,0 +1,366 @@ +//! Poll-to-handler controls for acknowledged checkpoint snapshots, not a real Server. +use durable_workflow::{ + json, testing::WorkflowEnvironment, ActivityHeartbeatResumeUnavailable, Client, Value, Worker, +}; +use serde::Deserialize; +use std::{ + collections::VecDeque, + sync::{Arc, Mutex}, + time::Duration, +}; +use tokio::{ + io::{AsyncReadExt, AsyncWriteExt}, + net::TcpListener, +}; + +const AT: &str = "2026-11-14T09:00:00.123456789Z"; +fn claim(attempt: u64, execution: &str, details: Value, at: Value) -> Value { + json!({"task":{"task_id":format!("task-{execution}-{attempt}"),"workflow_id":"workflow", + "run_id":"run","namespace":"default","activity_execution_id":execution, + "activity_attempt_id":format!("attempt-{execution}-{attempt}"),"attempt_number":attempt, + "activity_type":"inspect","payload_codec":"avro","lease_owner":"checkpoint-worker", + "retry_policy":{"max_attempts":3,"backoff_seconds":[1,2],"schedule_to_close_timeout":90}, + "last_heartbeat_details":details,"last_heartbeat_at":at}, + "server_capabilities":{"activity_heartbeat_resume":{"supported":true,"version":1}}}) +} +struct Server { + url: String, + task: tokio::task::JoinHandle<()>, + registrations: Arc>>, +} +impl Drop for Server { + fn drop(&mut self) { + self.task.abort(); + } +} +impl Server { + async fn start(envelopes: Vec, heartbeat_replies: Vec<(u16, Value)>) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let url = format!("http://{}", listener.local_addr().unwrap()); + let registrations = Arc::new(Mutex::new(Vec::new())); + let recorded = registrations.clone(); + let mut envelopes: VecDeque<_> = envelopes.into(); + let mut heartbeats: VecDeque<_> = heartbeat_replies.into(); + let task = tokio::spawn(async move { + loop { + let (mut stream, _) = listener.accept().await.unwrap(); + let mut bytes = Vec::new(); + let (end, length) = loop { + let mut buffer = [0; 4096]; + let count = stream.read(&mut buffer).await.unwrap(); + assert_ne!(count, 0); + bytes.extend_from_slice(&buffer[..count]); + assert!(bytes.len() < 65536); + if let Some(index) = bytes.windows(4).position(|x| x == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&bytes[..index]); + let length = headers + .lines() + .find_map(|line| { + let (k, v) = line.split_once(':')?; + k.eq_ignore_ascii_case("content-length") + .then(|| v.trim().parse::().unwrap()) + }) + .unwrap_or(0); + assert!(length < 32768); + break (index + 4, length); + } + }; + while bytes.len() < end + length { + let mut buffer = [0; 4096]; + let count = stream.read(&mut buffer).await.unwrap(); + assert_ne!(count, 0); + bytes.extend_from_slice(&buffer[..count]); + } + let headers = String::from_utf8_lossy(&bytes[..end]); + let path = headers.split_whitespace().nth(1).unwrap(); + let request: Value = + serde_json::from_slice(&bytes[end..end + length]).unwrap_or(Value::Null); + let (status, body) = if path.ends_with("/cluster/info") { + (200, json!({})) + } else if path.ends_with("/register") { + recorded.lock().unwrap().push(request.clone()); + ( + 200, + json!({"registered":true,"worker_id":"checkpoint-worker","namespace":"default","task_queue":"checkpoint-queue","protocol_version":"1.20","server_capabilities":{"cooperative_cancellation":true,"activity_heartbeat_resume":{"supported":true,"version":1}},"capabilities":request["capabilities"]}), + ) + } else if path.ends_with("/workflow-tasks/poll") { + (200, json!({"task":null})) + } else if path.ends_with("/activity-tasks/poll") { + (200, envelopes.pop_front().unwrap_or(json!({"task":null}))) + } else if path.ends_with("/status") { + ( + 200, + json!({"task_id":path.split('/').rev().nth(1).unwrap(),"activity_attempt_id":request["activity_attempt_id"],"lease_owner":"checkpoint-worker","can_continue":true,"cancel_requested":false,"heartbeat_recorded":false,"reason":null,"task_status":"leased","attempt_status":"running","activity_status":"running","lease_expires_at":"2099-01-01T00:00:00Z"}), + ) + } else if path.ends_with("/heartbeat") { + heartbeats.pop_front().expect("declared heartbeat reply") + } else if path.ends_with("/complete") || path.ends_with("/fail") { + (200, json!({"recorded":true})) + } else { + panic!("unexpected mock path {path}") + }; + let body = body.to_string(); + let reason = if status == 200 { "OK" } else { "Refused" }; + stream.write_all(format!("HTTP/1.1 {status} {reason}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",body.len()).as_bytes()).await.unwrap(); + stream.shutdown().await.unwrap(); + } + }); + Self { + url, + task, + registrations, + } + } + fn worker(&self, cooperative: bool) -> Worker { + Worker::new( + Client::builder(&self.url) + .timeout(Duration::from_secs(2)) + .build() + .unwrap(), + "checkpoint-queue", + ) + .worker_id("checkpoint-worker") + .cooperative_cancellation(cooperative) + } +} + +#[derive(Debug, Deserialize, PartialEq)] +struct Progress { + cursor: u64, + flags: Vec>, +} + +#[tokio::test] +async fn unavailable_and_supported_empty_are_distinct() { + let valid = claim(2, "execution", json!({"cursor":1}), json!(AT)); + let mut cases = Vec::new(); + for capability in [ + Value::Null, + json!(false), + json!({"supported":false,"version":1}), + json!({"supported":true,"version":2}), + json!({"supported":true,"version":"1"}), + ] { + let mut wire = valid.clone(); + wire["server_capabilities"]["activity_heartbeat_resume"] = capability; + cases.push(wire); + } + let mut missing = valid; + missing["task"] + .as_object_mut() + .unwrap() + .remove("last_heartbeat_details"); + missing["task"] + .as_object_mut() + .unwrap() + .remove("last_heartbeat_at"); + cases.push(missing); + cases.push(claim(1, "empty", Value::Null, Value::Null)); + let server = Server::start(cases, vec![]).await; + let mut worker = server.worker(false); + let seen = Arc::new(Mutex::new(0)); + let captured = seen.clone(); + worker.register_activity("inspect", move |ctx, _| { + let seen = captured.clone(); + async move { + let index = *seen.lock().unwrap(); + if index < 6 { + assert_eq!(ctx.heartbeat_resume_version(), None); + assert_eq!( + ctx.last_heartbeat_details(), + Err(ActivityHeartbeatResumeUnavailable) + ); + assert!(ctx.last_heartbeat_at().is_err()); + } else { + assert_eq!(ctx.heartbeat_resume_version(), Some(1)); + assert_eq!(ctx.last_heartbeat_details().unwrap(), None); + assert_eq!(ctx.last_heartbeat_at().unwrap(), None); + assert!(ctx.last_heartbeat_checkpoint().unwrap().is_none()); + } + *seen.lock().unwrap() += 1; + Ok(Value::Null) + } + }); + worker.register().await.unwrap(); + for _ in 0..7 { + assert_eq!(worker.run_once().await.unwrap(), 1); + } + assert_eq!(*seen.lock().unwrap(), 7); + assert!(server.registrations.lock().unwrap()[0]["capabilities"] + .as_array() + .unwrap() + .iter() + .any(|v| v == "activity_heartbeat_resume")); +} + +#[tokio::test] +async fn ordinary_and_cooperative_handlers_receive_typed_immutable_claims() { + for cooperative in [false, true] { + let details = + json!({"cursor":u64::MAX,"flags":[true,null,false],"secret":"private-checkpoint"}); + let server = Server::start( + vec![claim(2, "execution", details.clone(), json!(AT))], + vec![], + ) + .await; + let mut worker = server.worker(cooperative); + let expected = details.clone(); + let finished = Arc::new(Mutex::new(false)); + let completed = finished.clone(); + worker.register_activity("inspect",move |mut ctx,_|{let expected=expected.clone();let completed=completed.clone();async move { + assert_eq!(ctx.activity_execution_id(),Some("execution"));assert_eq!(ctx.activity_attempt_id,"attempt-execution-2"); + assert_eq!(ctx.workflow_id(),Some("workflow"));assert_eq!(ctx.run_id(),Some("run")); + let checkpoint=ctx.last_heartbeat_checkpoint()?.unwrap().clone(); + assert_eq!(checkpoint.details(),&expected);assert_eq!(checkpoint.contract_version(),1); + assert_eq!(checkpoint.details_as::().unwrap(),Progress{cursor:u64::MAX,flags:vec![Some(true),None,Some(false)]}); + assert_eq!(checkpoint.acknowledged_at().to_rfc3339_opts(chrono::SecondsFormat::Nanos,true),AT); + assert_eq!(ctx.last_heartbeat_at()?.unwrap(),checkpoint.acknowledged_at()); + assert!(!format!("{ctx:?}").contains("private-checkpoint")); + ctx.attempt_number=99;assert_eq!(ctx.retry_info().attempt_number(),2); + assert_eq!(ctx.retry_info().policy(),Some(&json!({"max_attempts":3,"backoff_seconds":[1,2],"schedule_to_close_timeout":90}))); + let mut detached=ctx.last_heartbeat_details()?.unwrap().clone();detached["cursor"]=json!(0); + assert_eq!(ctx.last_heartbeat_checkpoint()?.unwrap(),&checkpoint);*completed.lock().unwrap()=true;Ok(Value::Null) + }}); + worker.register().await.unwrap(); + assert_eq!(worker.run_once().await.unwrap(), 1); + assert!(*finished.lock().unwrap()); + } +} + +#[tokio::test] +async fn later_claims_do_not_leak_checkpoints_between_executions_or_runs() { + let mut last = claim(2, "other-execution", Value::Null, Value::Null); + last["task"]["workflow_id"] = json!("other-workflow"); + last["task"]["run_id"] = json!("other-run"); + let server = Server::start( + vec![ + claim(1, "same", Value::Null, Value::Null), + claim(2, "same", json!({"cursor":7}), json!(AT)), + claim( + 3, + "same", + json!({"cursor":8}), + json!("2026-11-14T09:00:01Z"), + ), + last, + ], + vec![], + ) + .await; + let mut worker = server.worker(false); + let seen = Arc::new(Mutex::new(Vec::new())); + let captured = seen.clone(); + worker.register_activity("inspect", move |ctx, _| { + let seen = captured.clone(); + async move { + seen.lock().unwrap().push(( + ctx.activity_execution_id().unwrap().to_string(), + ctx.activity_attempt_id.clone(), + ctx.last_heartbeat_details()?.cloned(), + ctx.run_id().unwrap().to_string(), + )); + Ok(Value::Null) + } + }); + worker.register().await.unwrap(); + for _ in 0..4 { + assert_eq!(worker.run_once().await.unwrap(), 1); + } + let rows = seen.lock().unwrap(); + assert_eq!(rows[0].2, None); + assert_eq!(rows[1].2, Some(json!({"cursor":7}))); + assert_eq!(rows[2].2, Some(json!({"cursor":8}))); + assert_ne!(rows[1].1, rows[2].1); + assert_eq!(rows[1].0, rows[2].0); + assert_eq!(rows[3].2, None); + assert_eq!(rows[3].3, "other-run"); +} + +#[tokio::test] +async fn accepted_rejected_and_uncertain_writes_never_mutate_the_received_snapshot() { + let server = Server::start( + vec![claim(2, "execution", json!({"cursor":7}), json!(AT))], + vec![ + ( + 200, + json!({"heartbeat_recorded":true,"last_heartbeat_at":"2026-11-14T09:01:00Z"}), + ), + ( + 200, + json!({"heartbeat_recorded":false,"can_continue":false,"reason":"stale_attempt"}), + ), + (409, json!({"message":"stale attempt","recorded":false})), + (503, json!({"message":"response uncertain"})), + ], + ) + .await; + let mut worker = server.worker(false); + let finished = Arc::new(Mutex::new(false)); + let completed = finished.clone(); + worker.register_activity("inspect", move |ctx, _| { + let completed = completed.clone(); + async move { + let checkpoint = ctx.last_heartbeat_checkpoint()?.unwrap().clone(); + assert!(ctx.heartbeat(json!({"cursor":8})).await?.heartbeat_recorded); + let rejected = ctx.heartbeat(json!({"cursor":9})).await?; + assert!(!rejected.heartbeat_recorded && rejected.should_stop()); + assert!(ctx.heartbeat(json!({"cursor":10})).await.is_err()); + assert!(ctx.heartbeat(json!({"cursor":11})).await.is_err()); + assert_eq!(ctx.last_heartbeat_checkpoint()?.unwrap(), &checkpoint); + *completed.lock().unwrap() = true; + Ok(Value::Null) + } + }); + worker.register().await.unwrap(); + assert_eq!(worker.run_once().await.unwrap(), 1); + assert!(*finished.lock().unwrap()); +} + +#[tokio::test] +async fn malformed_negotiated_checkpoints_refuse_before_the_handler() { + let original = claim(2, "execution", json!({"cursor":7}), json!(AT)); + let mut cases = Vec::new(); + for (field, value) in [ + ("last_heartbeat_at", Value::Null), + ("last_heartbeat_at", json!("no timestamp")), + ("last_heartbeat_at", json!("2026-11-14T09:00:00")), + ("last_heartbeat_details", Value::Null), + ("activity_execution_id", json!("")), + ("activity_attempt_id", json!("")), + ("workflow_id", json!("")), + ("run_id", json!("")), + ("namespace", json!("foreign")), + ("attempt_number", json!(1)), + ] { + let mut wire = original.clone(); + wire["task"][field] = value; + cases.push(wire); + } + for field in ["last_heartbeat_at", "last_heartbeat_details"] { + let mut wire = original.clone(); + wire["task"].as_object_mut().unwrap().remove(field); + cases.push(wire); + } + let server = Server::start(cases, vec![]).await; + let mut worker = server.worker(false); + worker.register_activity("inspect", |_, _| async { + panic!("malformed checkpoint reached handler"); + #[allow(unreachable_code)] + Ok(Value::Null) + }); + worker.register().await.unwrap(); + for _ in 0..12 { + assert!(worker.run_once().await.is_err()); + } +} + +#[tokio::test] +async fn server_free_activities_report_unavailable() { + let mut env = WorkflowEnvironment::new().unwrap(); + env.register_activity("inspect", |ctx, _| async move { + assert_eq!(ctx.heartbeat_resume_version(), None); + assert!(ctx.last_heartbeat_checkpoint().is_err()); + Ok(Value::Null) + }); + assert_eq!(env.run_activity("inspect", ()).await.unwrap(), Value::Null); +} From e6fe082af0635da8cdb717f01dd2218e45af0601 Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:07:32 +0000 Subject: [PATCH 2/3] Keep the registration capability snapshot complete --- src/lib.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/lib.rs b/src/lib.rs index 73227780..f32bb123 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -34389,6 +34389,7 @@ mod tests { DURABLE_SELECTION_CAPABILITY, MEMO_UPSERTS_CAPABILITY, TYPED_SEARCH_ATTRIBUTES_CAPABILITY, + activity_checkpoint::CAPABILITY, QUERY_TASKS_CAPABILITY, WORKFLOW_UPDATES_CAPABILITY, MESSAGE_STREAMS_CAPABILITY From 6348d41bb1fd4e27315b972d32afa1392fbac00c Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Sat, 14 Nov 2026 11:27:59 +0000 Subject: [PATCH 3/3] Validate supported-empty checkpoint claim identity --- src/activity_checkpoint.rs | 21 ++++++++++++--------- tests/activity_checkpoint.rs | 30 +++++++++++++++++++++++++++++- 2 files changed, 41 insertions(+), 10 deletions(-) diff --git a/src/activity_checkpoint.rs b/src/activity_checkpoint.rs index 1c3aa8ef..59b3b45c 100644 --- a/src/activity_checkpoint.rs +++ b/src/activity_checkpoint.rs @@ -93,18 +93,21 @@ impl HeartbeatResume { if details.is_none() && at.is_none() { return Ok(Self::Unavailable); } + let identity = |field: &str| task[field].as_str().is_some_and(|value| !value.is_empty()); + let attempt = task["attempt_number"].as_u64().filter(|number| *number > 0); + if !identity("activity_execution_id") + || !identity("activity_attempt_id") + || !identity("workflow_id") + || !identity("run_id") + || task["namespace"].as_str() != Some(namespace) + || attempt.is_none() + { + return Err(invalid()); + } match (details, at) { (Some(Value::Null), Some(Value::Null)) => Ok(Self::Available(None)), (Some(details), Some(Value::String(at))) if !details.is_null() => { - let identity = - |field: &str| task[field].as_str().is_some_and(|value| !value.is_empty()); - if !identity("activity_execution_id") - || !identity("activity_attempt_id") - || !identity("workflow_id") - || !identity("run_id") - || task["namespace"].as_str() != Some(namespace) - || !task["attempt_number"].as_u64().is_some_and(|n| n > 1) - { + if attempt == Some(1) { return Err(invalid()); } let acknowledged_at = DateTime::parse_from_rfc3339(at) diff --git a/tests/activity_checkpoint.rs b/tests/activity_checkpoint.rs index 794032d3..e3553da3 100644 --- a/tests/activity_checkpoint.rs +++ b/tests/activity_checkpoint.rs @@ -341,6 +341,34 @@ async fn malformed_negotiated_checkpoints_refuse_before_the_handler() { wire["task"].as_object_mut().unwrap().remove(field); cases.push(wire); } + let empty = claim(1, "execution", Value::Null, Value::Null); + for (field, value) in [ + ("activity_execution_id", json!("")), + ("activity_attempt_id", json!("")), + ("workflow_id", json!("")), + ("run_id", json!("")), + ("namespace", json!("foreign")), + ("attempt_number", json!(0)), + ("attempt_number", json!(-1)), + ("attempt_number", json!("1")), + ] { + let mut wire = empty.clone(); + wire["task"][field] = value; + cases.push(wire); + } + for field in [ + "activity_execution_id", + "activity_attempt_id", + "workflow_id", + "run_id", + "namespace", + "attempt_number", + ] { + let mut wire = empty.clone(); + wire["task"].as_object_mut().unwrap().remove(field); + cases.push(wire); + } + let refused = cases.len(); let server = Server::start(cases, vec![]).await; let mut worker = server.worker(false); worker.register_activity("inspect", |_, _| async { @@ -349,7 +377,7 @@ async fn malformed_negotiated_checkpoints_refuse_before_the_handler() { Ok(Value::Null) }); worker.register().await.unwrap(); - for _ in 0..12 { + for _ in 0..refused { assert!(worker.run_once().await.is_err()); } }