Feature: Serialization context tests

features/features.go

@@ -45,6 +45,15 @@ import (
 	schedule_duplicate_error "github.com/temporalio/features/features/schedule/duplicate_error"
 	schedule_pause "github.com/temporalio/features/features/schedule/pause"
 	schedule_trigger "github.com/temporalio/features/features/schedule/trigger"
+	serialization_context_activity_payloads "github.com/temporalio/features/features/serialization_context/activity_payloads"
+	serialization_context_async_activity_completion "github.com/temporalio/features/features/serialization_context/async_activity_completion"
+	serialization_context_child_workflow_payloads "github.com/temporalio/features/features/serialization_context/child_workflow_payloads"
+	serialization_context_child_workflow_payloads_default_id "github.com/temporalio/features/features/serialization_context/child_workflow_payloads_default_id"
+	serialization_context_continue_as_new "github.com/temporalio/features/features/serialization_context/continue_as_new"
+	serialization_context_external_signal "github.com/temporalio/features/features/serialization_context/external_signal"
+	serialization_context_failure "github.com/temporalio/features/features/serialization_context/failure"
+	serialization_context_local_activity_payloads "github.com/temporalio/features/features/serialization_context/local_activity_payloads"
+	serialization_context_workflow_payloads "github.com/temporalio/features/features/serialization_context/workflow_payloads"
 	signal_external "github.com/temporalio/features/features/signal/external"
 	telemetry_metrics "github.com/temporalio/features/features/telemetry/metrics"
 	update_activities "github.com/temporalio/features/features/update/activities"
@@ -108,6 +117,15 @@ func init() {
 		schedule_duplicate_error.Feature,
 		schedule_pause.Feature,
 		schedule_trigger.Feature,
+		serialization_context_activity_payloads.Feature,
+		serialization_context_async_activity_completion.Feature,
+		serialization_context_child_workflow_payloads.Feature,
+		serialization_context_child_workflow_payloads_default_id.Feature,
+		serialization_context_continue_as_new.Feature,
+		serialization_context_external_signal.Feature,
+		serialization_context_failure.Feature,
+		serialization_context_local_activity_payloads.Feature,
+		serialization_context_workflow_payloads.Feature,
 		signal_external.Feature,
 		telemetry_metrics.Feature,
 		update_activities.Feature,

features/serialization_context/README.md

@@ -0,0 +1,42 @@
+# Serialization context
+
+A `DataConverter`, `PayloadCodec` or `FailureConverter` can opt into receiving
+the context a payload is being converted in, so that it can, for example, derive
+an encryption key from the namespace or use the workflow ID as associated data.
+
+The features in this directory share a `sercontext` helper per language, which
+provides:
+
+- a payload codec that stamps the signature of its serialization context onto
+  every payload it encodes and refuses to decode a payload encoded under a
+  different context, so any asymmetry between the encoding and the decoding side
+  fails the feature wherever it happens
+- a failure converter that records the signature of its serialization context in
+  `Failure.source`
+
+Each feature then asserts the exact signature recorded in history, which pins
+down the context values themselves rather than only their symmetry. Contexts
+that never reach history are asserted against the set of signatures the codec was
+actually asked to convert with.
+
+The signature format is per language, because the SDKs expose different context
+fields. Signatures are only ever compared within a single run, so they do not
+need to agree across languages.
+
+## History replay
+
+Go and Java disable the harness history check. The replayer runs histories under
+a placeholder namespace and workflow ID, so payloads recorded by a real execution
+can never decode under a context derived from them.
+
+## Language notes
+
+- **Go** — `local_activity_payloads` fails: the SDK encodes the local activity
+  result with the plain worker converter and decodes it with the workflow
+  context. See that feature's README.
+- **Python** — the workflow side of an activity context only carries an activity
+  ID when the workflow sets one explicitly, so the features that schedule
+  activities pass an explicit `activity_id`.
+- **TypeScript** — no `local_activity_payloads`: the SDK has no local
+  activities. The activity context carries no workflow or activity type.
+- **Java** — the activity context carries no activity ID.

features/serialization_context/activity_payloads/README.md

@@ -0,0 +1,22 @@
+# Serialization context: activity payloads
+
+Activity payloads are converted with an `ActivitySerializationContext` carrying
+the namespace, workflow ID, workflow type, activity type, task queue, and
+`IsLocal = false`.
+
+Steps:
+
+- register a payload codec that stamps the signature of its serialization
+  context onto every payload it encodes, and rejects payloads that were encoded
+  under a different context
+- run an activity that heartbeats and fails its first attempt, so the second
+  attempt has to decode the heartbeat details recorded by the first one
+- verify the client result
+- verify that the `ActivityTaskScheduled` input payload and the
+  `ActivityTaskCompleted` result payload carry the activity signature
+- verify that the `WorkflowExecutionCompleted` result payload carries the
+  workflow signature, not the activity one
+
+Python only puts an activity ID in the workflow side context when the
+workflow sets one explicitly, so the workflow schedules the activity with an
+explicit activity ID.

features/serialization_context/activity_payloads/config.json

@@ -0,0 +1,5 @@
+{
+  "go": {
+    "minVersion": "v1.42.0"
+  }
+}

features/serialization_context/activity_payloads/feature.go

@@ -0,0 +1,111 @@
+package activity_payloads
+
+import (
+	"context"
+	"time"
+
+	"github.com/temporalio/features/features/serialization_context/sercontext"
+	"github.com/temporalio/features/harness/go/harness"
+	historypb "go.temporal.io/api/history/v1"
+	"go.temporal.io/sdk/activity"
+	"go.temporal.io/sdk/client"
+	"go.temporal.io/sdk/temporal"
+	"go.temporal.io/sdk/workflow"
+)
+
+const (
+	workflowInput = "hello"
+	heartbeatData = "beat"
+)
+
+var Feature = harness.Feature{
+	Workflows:     Workflow,
+	Activities:    Activity,
+	ClientOptions: sercontext.ClientOptions(),
+	Execute:       harness.ExecuteWithArgs(Workflow, workflowInput),
+	CheckResult:   CheckResult,
+	CheckHistory:  harness.NoHistoryCheck,
+}
+
+func Workflow(ctx workflow.Context, input string) (string, error) {
+	opts := workflow.ActivityOptions{
+		StartToCloseTimeout: 10 * time.Second,
+		HeartbeatTimeout:    5 * time.Second,
+		RetryPolicy:         &temporal.RetryPolicy{InitialInterval: time.Millisecond, MaximumAttempts: 2},
+	}
+	var result string
+	err := workflow.ExecuteActivity(workflow.WithActivityOptions(ctx, opts), Activity, input).Get(ctx, &result)
+	return result, err
+}
+
+// Activity heartbeats and fails on its first attempt so that its second attempt
+// has to decode the heartbeat details recorded by the first one.
+func Activity(ctx context.Context, input string) (string, error) {
+	if activity.GetInfo(ctx).Attempt == 1 {
+		activity.RecordHeartbeat(ctx, heartbeatData)
+		return "", harness.AppErrorf("retrying to read back heartbeat details")
+	}
+	var details string
+	if err := activity.GetHeartbeatDetails(ctx, &details); err != nil {
+		return "", err
+	}
+	return input + "|" + details, nil
+}
+
+func CheckResult(ctx context.Context, runner *harness.Runner, run client.WorkflowRun) error {
+	var result string
+	if err := run.Get(ctx, &result); err != nil {
+		return err
+	}
+	runner.Require.Equal(workflowInput+"|"+heartbeatData, result)
+
+	events, err := sercontext.Events(ctx, runner.Client, run.GetID(), run.GetRunID())
+	if err != nil {
+		return err
+	}
+
+	started, err := sercontext.FindEvent(events, "WorkflowExecutionStarted", func(e *historypb.HistoryEvent) bool {
+		return e.GetWorkflowExecutionStartedEventAttributes() != nil
+	})
+	if err != nil {
+		return err
+	}
+	scheduled, err := sercontext.FindEvent(events, "ActivityTaskScheduled", func(e *historypb.HistoryEvent) bool {
+		return e.GetActivityTaskScheduledEventAttributes() != nil
+	})
+	if err != nil {
+		return err
+	}
+	scheduledAttrs := scheduled.GetActivityTaskScheduledEventAttributes()
+	expected := sercontext.ActivitySignature(
+		runner.Namespace,
+		run.GetID(),
+		started.GetWorkflowExecutionStartedEventAttributes().GetWorkflowType().GetName(),
+		scheduledAttrs.GetActivityType().GetName(),
+		scheduledAttrs.GetTaskQueue().GetName(),
+		false,
+	)
+	runner.Require.Equal(expected, sercontext.FirstSignature(scheduledAttrs.GetInput()))
+
+	completed, err := sercontext.FindEvent(events, "ActivityTaskCompleted", func(e *historypb.HistoryEvent) bool {
+		return e.GetActivityTaskCompletedEventAttributes() != nil
+	})
+	if err != nil {
+		return err
+	}
+	runner.Require.Equal(expected,
+		sercontext.FirstSignature(completed.GetActivityTaskCompletedEventAttributes().GetResult()))
+
+	workflowCompleted, err := sercontext.FindEvent(events, "WorkflowExecutionCompleted", func(e *historypb.HistoryEvent) bool {
+		return e.GetWorkflowExecutionCompletedEventAttributes() != nil
+	})
+	if err != nil {
+		return err
+	}
+	runner.Require.Equal(
+		sercontext.WorkflowSignature(runner.Namespace, run.GetID()),
+		sercontext.FirstSignature(workflowCompleted.GetWorkflowExecutionCompletedEventAttributes().GetResult()),
+	)
+
+	return nil
+}

features/serialization_context/activity_payloads/feature.java

@@ -0,0 +1,135 @@
+package serialization_context.activity_payloads;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import io.temporal.activity.Activity;
+import io.temporal.activity.ActivityInterface;
+import io.temporal.activity.ActivityMethod;
+import io.temporal.activity.ActivityOptions;
+import io.temporal.client.WorkflowClientOptions;
+import io.temporal.common.RetryOptions;
+import io.temporal.failure.ApplicationFailure;
+import io.temporal.sdkfeatures.Feature;
+import io.temporal.sdkfeatures.Run;
+import io.temporal.sdkfeatures.Runner;
+import io.temporal.worker.Worker;
+import io.temporal.workflow.Workflow;
+import io.temporal.workflow.WorkflowInterface;
+import io.temporal.workflow.WorkflowMethod;
+import java.time.Duration;
+import serialization_context.sercontext.SerContext;
+
+@WorkflowInterface
+public interface feature extends Feature {
+
+  String WORKFLOW_INPUT = "hello";
+  String HEARTBEAT_DATA = "beat";
+
+  @WorkflowMethod
+  String workflow(String input);
+
+  @ActivityInterface
+  interface Activities {
+    @ActivityMethod
+    String activityWithHeartbeat(String input);
+
+    /** Fails its first attempt so the second one has to decode the heartbeat details. */
+    class Impl implements Activities {
+      @Override
+      public String activityWithHeartbeat(String input) {
+        var context = Activity.getExecutionContext();
+        if (context.getInfo().getAttempt() == 1) {
+          context.heartbeat(HEARTBEAT_DATA);
+          throw ApplicationFailure.newFailure(
+              "retrying to read back heartbeat details", "RetryError");
+        }
+        return input + "|" + context.getHeartbeatDetails(String.class).orElse("");
+      }
+    }
+  }
+
+  class Impl implements feature {
+
+    @Override
+    public void prepareWorker(Worker worker) {
+      worker.registerActivitiesImplementations(new Activities.Impl());
+    }
+
+    @Override
+    public String workflow(String input) {
+      var activities =
+          Workflow.newActivityStub(
+              Activities.class,
+              ActivityOptions.newBuilder()
+                  .setStartToCloseTimeout(Duration.ofSeconds(10))
+                  .setHeartbeatTimeout(Duration.ofSeconds(5))
+                  .setRetryOptions(
+                      RetryOptions.newBuilder()
+                          .setInitialInterval(Duration.ofMillis(1))
+                          .setMaximumAttempts(2)
+                          .build())
+                  .build());
+      return activities.activityWithHeartbeat(input);
+    }
+
+    @Override
+    public void workflowClientOptions(WorkflowClientOptions.Builder builder) {
+      builder.setDataConverter(SerContext.dataConverter());
+    }
+
+    @Override
+    public Run execute(Runner runner) throws Exception {
+      return runner.executeSingleWorkflow(null, WORKFLOW_INPUT);
+    }
+
+    @Override
+    public void checkResult(Runner runner, Run run) throws Exception {
+      assertEquals(
+          WORKFLOW_INPUT + "|" + HEARTBEAT_DATA, runner.waitForRunResult(run, String.class));
+
+      var history = runner.getWorkflowHistory(run);
+      var started =
+          SerContext.findEvent(
+                  history,
+                  "WorkflowExecutionStarted",
+                  e -> e.hasWorkflowExecutionStartedEventAttributes())
+              .getWorkflowExecutionStartedEventAttributes();
+      var scheduled =
+          SerContext.findEvent(
+                  history, "ActivityTaskScheduled", e -> e.hasActivityTaskScheduledEventAttributes())
+              .getActivityTaskScheduledEventAttributes();
+
+      var expected =
+          SerContext.activitySignature(
+              runner.config.namespace,
+              run.execution.getWorkflowId(),
+              started.getWorkflowType().getName(),
+              scheduled.getActivityType().getName(),
+              scheduled.getTaskQueue().getName(),
+              false);
+      assertEquals(expected, SerContext.firstSignature(scheduled.getInput()));
+
+      var completed =
+          SerContext.findEvent(
+                  history, "ActivityTaskCompleted", e -> e.hasActivityTaskCompletedEventAttributes())
+              .getActivityTaskCompletedEventAttributes();
+      assertEquals(expected, SerContext.firstSignature(completed.getResult()));
+
+      var workflowCompleted =
+          SerContext.findEvent(
+                  history,
+                  "WorkflowExecutionCompleted",
+                  e -> e.hasWorkflowExecutionCompletedEventAttributes())
+              .getWorkflowExecutionCompletedEventAttributes();
+      assertEquals(
+          SerContext.workflowSignature(runner.config.namespace, run.execution.getWorkflowId()),
+          SerContext.firstSignature(workflowCompleted.getResult()));
+    }
+
+    @Override
+    public void checkHistory(Runner runner, Run run) {
+      // The replayer runs histories under a placeholder namespace and workflow ID, which a context
+      // derived signature can never match.
+    }
+  }
+}