Replayer (#108)

Author: cretz

README.md

@@ -580,6 +580,22 @@ cancellation of all outstanding activities.
 The `shutdown()` invocation will wait on all activities to complete, so if a long-running activity does not at least
 respect cancellation, the shutdown may never complete.
 
+### Workflow Replay
+
+Given a workflow's history, it can be replayed locally to check for things like non-determinism errors. For example,
+assuming `history_json_str` is populated with a JSON string history either exported from the web UI or from `tctl`, the
+following function will replay it:
+
+```python
+from temporalio.worker import Replayer
+
+async def run_replayer(history_json_str: str):
+  replayer = Replayer(workflows=[SayHello])
+  await replayer.replay_workflow(history_json_str)
+```
+
+This will throw an error if any non-determinism is detected.
+
 ### OpenTelemetry Support
 
 OpenTelemetry support requires the optional `opentelemetry` dependencies which are part of the `opentelemetry` extra.

temporalio/bridge/src/lib.rs

@@ -1,4 +1,5 @@
 use pyo3::prelude::*;
+use pyo3::types::PyBytes;
 
 mod client;
 mod telemetry;
@@ -22,6 +23,7 @@ fn temporal_sdk_bridge(py: Python, m: &PyModule) -> PyResult<()> {
     )?;
     m.add_class::<worker::WorkerRef>()?;
     m.add_function(wrap_pyfunction!(new_worker, m)?)?;
+    m.add_function(wrap_pyfunction!(new_replay_worker, m)?)?;
     Ok(())
 }
 
@@ -42,3 +44,11 @@ fn new_worker(
 ) -> PyResult<worker::WorkerRef> {
     worker::new_worker(&client, config)
 }
+
+#[pyfunction]
+fn new_replay_worker(
+    history_proto: &PyBytes,
+    config: worker::WorkerConfig,
+) -> PyResult<worker::WorkerRef> {
+    worker::new_replay_worker(&history_proto, config)
+}

temporalio/bridge/src/worker.rs

@@ -9,6 +9,7 @@ use temporal_sdk_core::api::errors::{PollActivityError, PollWfError};
 use temporal_sdk_core_api::Worker;
 use temporal_sdk_core_protos::coresdk::workflow_completion::WorkflowActivationCompletion;
 use temporal_sdk_core_protos::coresdk::{ActivityHeartbeat, ActivityTaskCompletion};
+use temporal_sdk_core_protos::temporal::api::history::v1::History;
 
 use crate::client;
 
@@ -50,6 +51,21 @@ pub fn new_worker(client: &client::ClientRef, config: WorkerConfig) -> PyResult<
     })
 }
 
+pub fn new_replay_worker(history_proto: &PyBytes, config: WorkerConfig) -> PyResult<WorkerRef> {
+    // This must be run with the Tokio context available
+    let _guard = pyo3_asyncio::tokio::get_runtime().enter();
+    let config: temporal_sdk_core::WorkerConfig = config.try_into()?;
+    let history = History::decode(history_proto.as_bytes())
+        .map_err(|err| PyValueError::new_err(format!("Invalid proto: {}", err)))?;
+    Ok(WorkerRef {
+        worker: Some(Arc::new(
+            temporal_sdk_core::init_replay_worker(config, &history).map_err(|err| {
+                PyValueError::new_err(format!("Failed creating replay worker: {}", err))
+            })?,
+        )),
+    })
+}
+
 #[pymethods]
 impl WorkerRef {
     fn poll_workflow_activation<'p>(&self, py: Python<'p>) -> PyResult<&'p PyAny> {

temporalio/bridge/worker.py

@@ -1,12 +1,15 @@
 """Worker using SDK Core."""
 
+from __future__ import annotations
+
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Awaitable, Callable, Iterable, List, Mapping, Optional
 
 import google.protobuf.internal.containers
 from typing_extensions import TypeAlias
 
 import temporalio.api.common.v1
+import temporalio.api.history.v1
 import temporalio.bridge.client
 import temporalio.bridge.proto
 import temporalio.bridge.proto.activity_task
@@ -42,14 +45,28 @@ class WorkerConfig:
 class Worker:
     """SDK Core worker."""
 
-    def __init__(
-        self, client: temporalio.bridge.client.Client, config: WorkerConfig
-    ) -> None:
-        """Create SDK core worker with a client and config."""
-        self._ref = temporalio.bridge.temporal_sdk_bridge.new_worker(
-            client._ref, config
+    @staticmethod
+    def create(client: temporalio.bridge.client.Client, config: WorkerConfig) -> Worker:
+        """Create a bridge worker from a bridge client."""
+        return Worker(
+            temporalio.bridge.temporal_sdk_bridge.new_worker(client._ref, config)
+        )
+
+    @staticmethod
+    def for_replay(
+        history: temporalio.api.history.v1.History, config: WorkerConfig
+    ) -> Worker:
+        """Create a bridge replay worker from history."""
+        return Worker(
+            temporalio.bridge.temporal_sdk_bridge.new_replay_worker(
+                history.SerializeToString(), config
+            )
         )
 
+    def __init__(self, ref: temporalio.bridge.temporal_sdk_bridge.WorkerRef) -> None:
+        """Create SDK core worker from a bridge worker."""
+        self._ref = ref
+
     async def poll_workflow_activation(
         self,
     ) -> temporalio.bridge.proto.workflow_activation.WorkflowActivation:

temporalio/worker/__init__.py

@@ -19,6 +19,7 @@
     WorkflowInterceptorClassInput,
     WorkflowOutboundInterceptor,
 )
+from .replayer import Replayer, ReplayerConfig
 from .worker import Worker, WorkerConfig
 from .workflow_instance import (
     UnsandboxedWorkflowRunner,
@@ -31,6 +32,8 @@
     # Primary types
     "Worker",
     "WorkerConfig",
+    "Replayer",
+    "ReplayerConfig",
     # Interceptor base classes
     "Interceptor",
     "ActivityInboundInterceptor",