Expose acknowledged activity heartbeat checkpoints

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

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::<T>()` 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::<Progress>()?.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:

src/activity_checkpoint.rs

@@ -0,0 +1,145 @@
+//! 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<ActivityHeartbeatResumeUnavailable> 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<Utc>,
+}
+
+impl fmt::Debug for ActivityHeartbeatCheckpoint {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("ActivityHeartbeatCheckpoint")
+            .field("details", &"<checkpoint>")
+            .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<T: DeserializeOwned>(&self) -> serde_json::Result<T> {
+        serde_json::from_value(self.details.clone())
+    }
+
+    /// The producer's acknowledgement time, preserving fractional precision.
+    pub fn acknowledged_at(&self) -> &DateTime<Utc> {
+        &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<ActivityHeartbeatCheckpoint>),
+}
+
+impl HeartbeatResume {
+    pub(super) fn from_poll(wire: &Value, namespace: &str) -> Result<Self> {
+        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);
+        }
+        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() => {
+                if attempt == Some(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<u64> {
+        match self {
+            Self::Unavailable => None,
+            Self::Available(_) => Some(1),
+        }
+    }
+
+    pub(super) fn checkpoint(
+        &self,
+    ) -> std::result::Result<Option<&ActivityHeartbeatCheckpoint>, 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())
+}

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(),

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<String>,
     workflow_id: Option<String>,
     run_id: Option<String>,
@@ -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<u64> {
+        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<Option<&ActivityHeartbeatCheckpoint>, ActivityHeartbeatResumeUnavailable>
+    {
+        self.heartbeat_resume.checkpoint()
+    }
+
+    /// Last acknowledged details, with unavailable distinct from no checkpoint.
+    pub fn last_heartbeat_details(
+        &self,
+    ) -> std::result::Result<Option<&Value>, 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<Option<&DateTime<chrono::Utc>>, 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.
@@ -34338,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

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(),

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,

src/worker_session.rs

@@ -269,12 +269,15 @@ pub(super) struct SessionActivityTask {
     pub(super) hydration_claim: Option<crate::activity_hydration::HydrationClaim>,
     #[serde(skip)]
     pub(super) activity_lease: Option<Arc<crate::activity_lease::authority::Lease>>,
+    #[serde(skip)]
+    pub(super) heartbeat_resume: crate::activity_checkpoint::HeartbeatResume,
 }
 
 pub(super) struct ActivityInvocation {
     pub(super) worker_session: Option<WorkerSession>,
     pub(super) hydration_claim: Option<crate::activity_hydration::HydrationClaim>,
     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(),
         }
     }
 }