Extend versioning's replay tester (temporalio#8124)

## What changed?
- Make the replay tester, used by versioning entity workflows, better.
- The versioning entity workflows CAN on every state change. Previously,
the replay tester was simply fetching the last execution of a versioning
entity workflow. Thus, we were skipping over a lot of NDE's that could
happen if one were to make changes in our workflows.

To fix this, the replay tester:
1. Fetches all the workflow ID's present in the CAN chain. Then, it
tests if for each individual run, are there code changes that have
caused NDE's.
2. It is important to remember that while we are fetching all the
workflows in CAN chain, SDK's replay tester does not test the chaining
itself. In other words, it tests the code path from the input that is
given at the start of each workflow execution. Thus, there could be a
chance that a code change introduced doesn't necessarily add new events
into a single workflow run. However, this code change can increase the
overall runs of the entity workflows and now the replay tester will see
this and understand that this as a cause for NDE.
3. Neatly makes a folder for that run and saves the required workflow
histories over there.
4. *Bonus*: I have also made the change to store the raw json file of
each workflow history. This would make debugging a whole lotta simpler.

It shall now look like:
<img width="1064" height="861" alt="image"
src="https://github.com/user-attachments/assets/67daf5c5-04c3-4236-86a0-f81470661d72"
/>



## Why?
- Robustness.

## How did you test it?
- [ ] built
- [ ] run locally and tested manually
- [ ] covered by existing tests
- [ ] added new unit test(s)
- [ ] added new functional test(s)

## Potential risks
- None

Author: Shivs11

service/worker/workerdeployment/replaytester/generate_history.sh

@@ -15,16 +15,111 @@
 deploymentName="foo"
 version="1.0"
 
+# Expected workflow counts - users can override these if their changes are expected to generate more workflows which will be true when a breaking change to 
+# these worfklows is introduced.
+# These values are used by the replay tester to validate that your workflow changes haven't accidentally created additional executions.
+EXPECTED_DEPLOYMENT_WORKFLOWS=${EXPECTED_DEPLOYMENT_WORKFLOWS:-8}
+EXPECTED_VERSION_WORKFLOWS=${EXPECTED_VERSION_WORKFLOWS:-10}
+
+echo "📋 Expected workflow counts:"
+echo "   Deployment workflows: $EXPECTED_DEPLOYMENT_WORKFLOWS"
+echo "   Version workflows: $EXPECTED_VERSION_WORKFLOWS"
+echo "   (Override with EXPECTED_DEPLOYMENT_WORKFLOWS=X EXPECTED_VERSION_WORKFLOWS=Y if you expect different counts)"
+echo ""
+
 # Create the default namespace
 temporal operator namespace create default
 
 # Run the worker which shall start the deployment entity workflows....
 echo "Running the Go program..."
 go run "$(dirname "$0")/worker/worker.go"
 
-# Download the history for the worker deployment workflow...
+echo "Waiting 5 seconds for all workflows to show up in visibility..."
+sleep 5
+
+# Function to download all workflow runs in CAN chain
+download_workflow_chain() {
+    local workflow_id=$1
+    local workflow_name=$2
+    local workflow_type=$3
+    local run_dir=$4
+    
+    echo "📥 Downloading all executions for: $workflow_id"
+    
+    # Use the working query method with TemporalNamespaceDivision
+    echo "   Getting the chain of CAN runs for this workflow using the TemporalNamespaceDivision query..."
+    run_ids=$(temporal workflow list \
+        --query "TemporalNamespaceDivision = \"TemporalWorkerDeployment\" AND WorkflowType = \"$workflow_type\"" \
+        --output json | \
+        jq -r '.[] | .execution.runId')
+    
+    # Count how many we found
+    if [ -z "$run_ids" ]; then
+        run_count=0
+    else
+        run_count=$(echo "$run_ids" | wc -l | tr -d ' ')
+    fi
+    echo "   Found $run_count executions"
+    
+    if [ "$run_count" -eq 0 ]; then
+        echo "   No executions found for $workflow_id"
+        return
+    fi
+    
+    # Download each execution
+    run_index=0
+    for run_id in $run_ids; do
+        if [ -n "$run_id" ]; then
+            echo "   Downloading run $((run_index + 1))/$run_count: $run_id"
+            
+            temporal workflow show \
+                -w "$workflow_id" \
+                -r "$run_id" \
+                --output json | \
+                gzip -9c > "$run_dir/replay_${workflow_name}_run_${run_id}.json.gz"
+            
+            ((run_index++))
+        fi
+    done
+    
+    # # Save run IDs for reference
+    # echo "$run_ids" > "$run_dir/${workflow_name}_all_runs.txt"
+    
+    echo "   ✅ Downloaded $run_index executions for $workflow_name"
+}
+
+# Create timestamped run directory 
 now=$(date +%s)
-temporal workflow show -w "temporal-sys-worker-deployment:$deploymentName" --output json | gzip -9c > "$(dirname "$0")/testdata/replay_worker_deployment_wf_$now.json.gz"
+run_dir="$(dirname "$0")/testdata/run_$now"
+mkdir -p "$run_dir"
+
+echo "📁 Creating run directory: $run_dir"
+
+# Download all executions for both workflow types
+download_workflow_chain "temporal-sys-worker-deployment:$deploymentName" "worker_deployment_wf" "temporal-sys-worker-deployment-workflow" "$run_dir"
+download_workflow_chain "temporal-sys-worker-deployment-version:$deploymentName:$version" "worker_deployment_version_wf" "temporal-sys-worker-deployment-version-workflow" "$run_dir"
+
+echo ""
+echo "🎉 Complete! All workflow execution histories downloaded to $run_dir"
+echo ""
+echo "📊 Summary for this run:"
+echo "   📂 Run directory: $run_dir"
+
+# Count files by workflow type
+deployment_files=$(find "$run_dir" -name "replay_worker_deployment_wf_*.json.gz" 2>/dev/null | wc -l | tr -d ' ')
+version_files=$(find "$run_dir" -name "replay_worker_deployment_version_wf_*.json.gz" 2>/dev/null | wc -l | tr -d ' ')
+
+echo "   Worker Deployment workflows: $deployment_files executions"
+echo "   Worker Version workflows: $version_files executions"
+
+# Save expected counts to a file for the replay tester to read
+cat > "$run_dir/expected_counts.txt" << EOF
+# Expected workflow counts for replay testing
+# Generated by generate_history.sh on $(date)
+EXPECTED_DEPLOYMENT_WORKFLOWS=$EXPECTED_DEPLOYMENT_WORKFLOWS
+EXPECTED_VERSION_WORKFLOWS=$EXPECTED_VERSION_WORKFLOWS
+ACTUAL_DEPLOYMENT_WORKFLOWS=$deployment_files
+ACTUAL_VERSION_WORKFLOWS=$version_files
+EOF
 
-# Download the history for the worker deployment version workflow...
-temporal workflow show -w "temporal-sys-worker-deployment-version:$deploymentName.$version" --output json | gzip -9c > "$(dirname "$0")/testdata/replay_worker_deployment_version_wf_$now.json.gz"
+echo "   📝 Expected counts saved to: $run_dir/expected_counts.txt"

service/worker/workerdeployment/replaytester/replay_test.go

@@ -1,11 +1,15 @@
 package replaytester
 
 import (
+	"bufio"
 	"compress/gzip"
 	"fmt"
 	"os"
 	"path/filepath"
+	"strconv"
+	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 	"go.temporal.io/sdk/client"
@@ -19,33 +23,164 @@ import (
 // TestReplays tests workflow logic backwards compatibility from previous versions.
 func TestReplays(t *testing.T) {
 	replayer := worker.NewWorkflowReplayer()
+
+	// Create version workflow wrapper to match production registration
+	versionWorkflow := func(ctx workflow.Context, args *deploymentspb.WorkerDeploymentVersionWorkflowArgs) error {
+		refreshIntervalGetter := func() any {
+			return 5 * time.Minute // default value for testing
+		}
+		visibilityGracePeriodGetter := func() any {
+			return 3 * time.Minute // default value for testing
+		}
+		return workerdeployment.VersionWorkflow(ctx, refreshIntervalGetter, visibilityGracePeriodGetter, args)
+	}
+
+	// Create deployment workflow wrapper to match production registration
 	deploymentWorkflow := func(ctx workflow.Context, args *deploymentspb.WorkerDeploymentWorkflowArgs) error {
 		maxVersionsGetter := func() int {
 			return 100
 		}
 		return workerdeployment.Workflow(ctx, maxVersionsGetter, args)
 	}
-	replayer.RegisterWorkflowWithOptions(workerdeployment.VersionWorkflow, workflow.RegisterOptions{Name: workerdeployment.WorkerDeploymentVersionWorkflowType})
+
+	replayer.RegisterWorkflowWithOptions(versionWorkflow, workflow.RegisterOptions{Name: workerdeployment.WorkerDeploymentVersionWorkflowType})
 	replayer.RegisterWorkflowWithOptions(deploymentWorkflow, workflow.RegisterOptions{Name: workerdeployment.WorkerDeploymentWorkflowType})
 
-	files, err := filepath.Glob("testdata/replay_*.json.gz")
+	logger := log.NewSdkLogger(log.NewTestLogger())
+
+	// Test all run directories (default behavior for comprehensive replay testing)
+	testAllRunDirectories(t, replayer, logger)
+
+}
+
+// testAllRunDirectories tests all directories prepended with "run_" since they contain replay test data
+func testAllRunDirectories(t *testing.T, replayer worker.WorkflowReplayer, logger *log.SdkLogger) {
+	runDirs, err := filepath.Glob("testdata/run_*")
+	require.NoError(t, err)
+
+	if len(runDirs) == 0 {
+		t.Skip("No run directories found. Run generate_history.sh first.")
+	}
+
+	fmt.Printf("Testing %d run directories\n", len(runDirs))
+
+	for _, runDir := range runDirs {
+		t.Run(filepath.Base(runDir), func(t *testing.T) {
+			fmt.Printf("Testing run: %s\n", runDir)
+			testRunDirectory(t, replayer, logger, runDir)
+		})
+	}
+}
+
+// testRunDirectory tests all workflow histories in a specific run directory
+func testRunDirectory(t *testing.T, replayer worker.WorkflowReplayer, logger *log.SdkLogger, runDir string) {
+	files, err := filepath.Glob(filepath.Join(runDir, "replay_*.json.gz"))
 	require.NoError(t, err)
 
-	fmt.Println("Number of files to replay:", len(files))
+	fmt.Printf("  Found %d workflow histories to replay\n", len(files))
 
-	logger := log.NewSdkLogger(log.NewTestLogger())
+	// Validate that workflow counts match expected values
+	validateWorkflowCounts(t, files, runDir)
 
+	// Validate that histories replay successfully
 	for _, filename := range files {
-		logger.Info("Replaying", "file", filename)
-		f, err := os.Open(filename)
-		require.NoError(t, err)
-		r, err := gzip.NewReader(f)
-		require.NoError(t, err)
-		history, err := client.HistoryFromJSON(r, client.HistoryJSONOptions{})
-		require.NoError(t, err)
-		err = replayer.ReplayWorkflowHistory(logger, history)
-		require.NoError(t, err)
-		_ = r.Close()
-		_ = f.Close()
+		t.Run(filepath.Base(filename), func(t *testing.T) {
+			replayWorkflowHistory(t, replayer, logger, filename)
+		})
+	}
+}
+
+// replayWorkflowHistory replays a single workflow history file and validates it
+func replayWorkflowHistory(t *testing.T, replayer worker.WorkflowReplayer, logger *log.SdkLogger, filename string) {
+	logger.Info("Replaying", "file", filename)
+	f, err := os.Open(filename)
+	require.NoError(t, err)
+	r, err := gzip.NewReader(f)
+	require.NoError(t, err)
+	history, err := client.HistoryFromJSON(r, client.HistoryJSONOptions{})
+	require.NoError(t, err)
+	err = replayer.ReplayWorkflowHistory(logger, history)
+	require.NoError(t, err)
+	_ = r.Close()
+	_ = f.Close()
+}
+
+// readExpectedCounts reads expected workflow counts from the expected_counts.txt file
+func readExpectedCounts(runDir string) (deploymentCount, versionCount int, err error) {
+	expectedCountsFile := filepath.Join(runDir, "expected_counts.txt")
+
+	file, err := os.Open(expectedCountsFile)
+	if err != nil {
+		// File doesn't exist - this might be an older test data directory
+		return 0, 0, fmt.Errorf("expected_counts.txt not found in %s: %w", runDir, err)
+	}
+	defer file.Close()
+
+	scanner := bufio.NewScanner(file)
+	for scanner.Scan() {
+		line := strings.TrimSpace(scanner.Text())
+		if strings.HasPrefix(line, "#") || line == "" {
+			continue // Skip comments and empty lines
+		}
+
+		if strings.HasPrefix(line, "EXPECTED_DEPLOYMENT_WORKFLOWS=") {
+			value := strings.TrimPrefix(line, "EXPECTED_DEPLOYMENT_WORKFLOWS=")
+			deploymentCount, err = strconv.Atoi(value)
+			if err != nil {
+				return 0, 0, fmt.Errorf("invalid deployment count: %w", err)
+			}
+		} else if strings.HasPrefix(line, "EXPECTED_VERSION_WORKFLOWS=") {
+			value := strings.TrimPrefix(line, "EXPECTED_VERSION_WORKFLOWS=")
+			versionCount, err = strconv.Atoi(value)
+			if err != nil {
+				return 0, 0, fmt.Errorf("invalid version count: %w", err)
+			}
+		}
+	}
+
+	return deploymentCount, versionCount, scanner.Err()
+}
+
+// validateWorkflowCounts ensures the number of deployment and version workflows matches expectations
+func validateWorkflowCounts(t *testing.T, files []string, runDir string) {
+	// Read expected counts from file
+	expectedDeploymentCount, expectedVersionCount, err := readExpectedCounts(runDir)
+	if err != nil {
+		// For backwards compatibility, skip validation if expected_counts.txt doesn't exist
+		fmt.Printf("  ⚠️  Skipping workflow count validation since expected_counts.txt doesn't exist for this test data directory; this is expected for older test data directories: %v\n", err)
+		return
+	}
+
+	// Count actual workflows
+	actualDeploymentCount := 0
+	actualVersionCount := 0
+
+	for _, file := range files {
+		filename := filepath.Base(file)
+		// Only count .gz files since those are the ones used for replay testing
+		if !strings.HasSuffix(filename, ".json.gz") {
+			continue
+		}
+
+		if strings.Contains(filename, "replay_worker_deployment_wf_run_") {
+			actualDeploymentCount++
+		} else if strings.Contains(filename, "replay_worker_deployment_version_wf_run_") {
+			actualVersionCount++
+		}
 	}
+
+	fmt.Printf("  Workflow counts - Expected: Deployment=%d, Version=%d | Actual: Deployment=%d, Version=%d\n",
+		expectedDeploymentCount, expectedVersionCount, actualDeploymentCount, actualVersionCount)
+
+	require.Equal(t, expectedDeploymentCount, actualDeploymentCount,
+		"Deployment workflow count mismatch in %s. Expected %d, got %d. "+
+			"This could mean your changes caused additional workflow executions. "+
+			"If this is expected, regenerate test data with: EXPECTED_DEPLOYMENT_WORKFLOWS=%d ./generate_history.sh",
+		runDir, expectedDeploymentCount, actualDeploymentCount, actualDeploymentCount)
+
+	require.Equal(t, expectedVersionCount, actualVersionCount,
+		"Version workflow count mismatch in %s. Expected %d, got %d. "+
+			"This could mean your changes caused additional workflow executions. "+
+			"If this is expected, regenerate test data with: EXPECTED_VERSION_WORKFLOWS=%d ./generate_history.sh",
+		runDir, expectedVersionCount, actualVersionCount, actualVersionCount)
 }

…play_deployment-workflow_initial.json.gz …play_deployment-workflow_initial.json.gzservice/worker/workerdeployment/replaytester/testdata/replay_deployment-workflow_initial.json.gz renamed to service/worker/workerdeployment/replaytester/testdata/run_1753913370/replay_deployment-workflow_initial.json.gz service/worker/workerdeployment/replaytester/testdata/replay_deployment-workflow_initial.json.gz renamed to service/worker/workerdeployment/replaytester/testdata/run_1753913370/replay_deployment-workflow_initial.json.gz

@@ -0,0 +1,6 @@
+# Expected workflow counts for replay testing
+# Generated by generate_history.sh on Thu 31 Jul 2025 14:44:51 EDT
+EXPECTED_DEPLOYMENT_WORKFLOWS=8
+EXPECTED_VERSION_WORKFLOWS=10
+ACTUAL_DEPLOYMENT_WORKFLOWS=8
+ACTUAL_VERSION_WORKFLOWS=10