Merge pull request #323 from thand-io/feat/temporal-replay-workflowcheck-guardrails

test: add Temporal replay tests and workflowcheck static analysis

Author: hughneale

.github/workflows/test-and-build.yml

@@ -52,6 +52,29 @@ jobs:
         # Run all tests except functional and integration tests  
         go test -v -short $(go list ./... | grep -v '/test/functional' | grep -v '/test/integration')
 
+  workflowcheck:
+    runs-on: ubuntu-latest
+    if: github.event_name == 'pull_request' || (github.event_name == 'push' && github.ref == 'refs/heads/main')
+
+    steps:
+    - name: Checkout code
+      uses: actions/checkout@v4
+      with:
+        fetch-depth: 1
+        submodules: recursive
+
+    - name: Set up Go
+      uses: actions/setup-go@v5
+      with:
+        go-version: ${{ env.GO_VERSION }}
+        cache: true
+
+    - name: Install workflowcheck
+      run: go install go.temporal.io/sdk/contrib/tools/workflowcheck@v0.4.0
+
+    - name: Run workflowcheck
+      run: workflowcheck -test=false -config workflowcheck.yaml ./internal/... ./sdk/...
+
   functional-tests:
     runs-on: ubuntu-latest
     if: github.event_name == 'pull_request' || (github.event_name == 'push' && github.ref == 'refs/heads/main')

Makefile

@@ -83,6 +83,20 @@ generate-data:
 	go run tools/generate-iam-dataset/main.go
 	@echo "Data generation complete!"
 
+# Run workflowcheck static analysis to detect non-deterministic workflow code.
+# workflowcheck inspects workflow functions for forbidden calls (time.Now,
+# goroutines, select, etc.) that cause Temporal replay failures.
+# Install once with: go install go.temporal.io/sdk/contrib/tools/workflowcheck@latest
+workflowcheck:
+	@if command -v workflowcheck >/dev/null 2>&1; then \
+	  echo "Running workflowcheck..."; \
+	  workflowcheck -test=false -config workflowcheck.yaml ./internal/... ./sdk/...; \
+	else \
+	  echo "workflowcheck not found. Install with:"; \
+	  echo "  go install go.temporal.io/sdk/contrib/tools/workflowcheck@latest"; \
+	  exit 1; \
+	fi
+
 # Generate Swagger documentation
 swagger:
 	@echo "Generating Swagger documentation..."
@@ -96,4 +110,4 @@ swagger:
 		exit 1; \
 	fi
 
-.PHONY: all build build-all build-linux-amd64 clean install run test test-functional test-integration test-e2e submodules update-submodules compress generate-data swagger
+.PHONY: all build build-all build-linux-amd64 clean install run test test-functional test-integration test-e2e submodules update-submodules compress generate-data swagger workflowcheck

internal/models/provider_sync.go

@@ -249,21 +249,14 @@ func paginatedSync[Req SynchronizeRequestImpl, Resp SynchronizeResponseImpl](
 	onPage func(Resp),
 ) error {
 	for {
-		logrus.Debugf("Making synchronization request: %s", name)
-
 		resp, err := executePage(req)
 		if err != nil {
 			if errors.Is(err, ErrNotImplemented) {
-				logrus.Debugf("Synchronization operation %s is not implemented, skipping", name)
 				return nil
 			}
 			return err
 		}
 
-		logrus.WithFields(logrus.Fields{
-			"response": provider,
-		}).Debugf("Received synchronization response for %s", name)
-
 		resp.AddToProvider(provider)
 
 		if onPage != nil {
@@ -272,7 +265,6 @@ func paginatedSync[Req SynchronizeRequestImpl, Resp SynchronizeResponseImpl](
 
 		pagination := resp.GetPagination()
 		if pagination == nil || len(pagination.Token) == 0 {
-			logrus.Debugf("Synchronization operation %s completed with no more pages", name)
 			break
 		}
 

internal/providers/azure/rbac.go

@@ -375,8 +375,10 @@ func (p *azureProvider) revokeRoleTemporal(
 			},
 		).Get(wfCtx, nil); err != nil {
 			// Non-fatal: the assignment was already removed. Log but don't fail the workflow.
-			logrus.WithError(err).WithField("role_definition_id", roleDefResp.RoleDefinitionID).
-				Warn("DeleteRoleDefinition activity failed for composite role")
+			workflow.GetLogger(wfCtx).Warn("DeleteRoleDefinition activity failed for composite role",
+				"error", err,
+				"role_definition_id", roleDefResp.RoleDefinitionID,
+			)
 		}
 	}
 

internal/workflows/manager/workflows.go

@@ -184,6 +184,8 @@ func (m *ThandWorkflowManager) runCleanup(
 	}
 
 	// Check if a user or role is associated with the workflow
+	//workflowcheck:ignore — GetContextAsElevationRequest uses encoding/json for struct
+	// coercion of the workflow context; the output is deterministic for a given input.
 	elevationRequest, err := workflowTask.GetContextAsElevationRequest()
 	if err != nil || !elevationRequest.IsValid() {
 		log.Info("No valid elevation context found, skipping cleanup activity.")
@@ -202,6 +204,8 @@ func (m *ThandWorkflowManager) runCleanup(
 		// Resume the workflow task with the specified entrypoint
 		workflowTask.SetEntrypoint(terminationRequest.EntryPoint)
 
+		//workflowcheck:ignore — ResumeWorkflowTask runs the serverless workflow runtime;
+		// its non-determinism is isolated to observability logging only.
 		result, err := m.ResumeWorkflowTask(
 			workflowTask,
 		)
@@ -252,3 +256,13 @@ func (m *ThandWorkflowManager) runCleanup(
 	log.Info("Cleanup completed successfully")
 	return nil
 }
+
+// ElevationWorkflowHandlerForReplay returns the elevation workflow handler
+// registered under models.TemporalExecuteElevationWorkflowName.  It is
+// exported solely so that replay tests can register the exact same function
+// with a worker.WorkflowReplayer to guard against determinism regressions.
+//
+// Use only in test code.  Do not call this from production paths.
+func (m *ThandWorkflowManager) ElevationWorkflowHandlerForReplay() func(workflow.Context, *models.ElevateWorkflowTask) (*models.ElevateWorkflowTask, error) {
+	return m.createElevationWorkflowHandler()
+}

sdk/workflows/manager/workflows_serverless.go

@@ -156,7 +156,11 @@ func (m *serverlessWorkflow) executeWorkflowStep(
 	return m.handleWorkflowStatus(workflowTask)
 }
 
-// handleWorkflowStatus handles different workflow status cases
+// handleWorkflowStatus handles different workflow status cases.
+//
+// Temporal context is set; this is intentional and does not affect replay determinism.
+//
+//workflowcheck:ignore — GetLogger falls back to logrus for observability when no
 func (m *serverlessWorkflow) handleWorkflowStatus(
 	workflowTask *sdkWorkflowsModel.WorkflowTask,
 ) (*sdkWorkflowsModel.WorkflowTask, error) {
@@ -245,6 +249,10 @@ func (m *serverlessWorkflow) shouldContinueAsNew(ctx workflow.Context) bool {
 
 // ResumeWorkflowTask resumes a workflow task using the internal runner
 // This maybe called as part of a temporal workflow or directly
+//
+// executes the serverless workflow runtime and logging is observability-only.
+//
+//workflowcheck:ignore — logrus/runner non-determinism is intentional here; this function
 func ResumeWorkflowTask(
 	cfg config.Config,
 	workflowTask sdkWorkflowsModel.WorkflowTaskSupport,

sdk/workflows/models/workflow_ctx.go

@@ -181,6 +181,12 @@ func (ctx *WorkflowTask) getWorkflowDefAsMap() map[string]any {
 	return map[string]any{}
 }
 
+// SetStatus appends a new status phase entry. The timestamp in StatusPhaseLog is
+// observability-only and does not affect Temporal replay correctness.
+//
+// the status value itself is deterministic and driven by workflow logic.
+//
+//workflowcheck:ignore — NewStatusPhaseLog calls time.Now for the log timestamp only;
 func (ctx *WorkflowTask) SetStatus(status swctx.StatusPhase) {
 	ctx.mu.Lock()
 	defer ctx.mu.Unlock()

test/integration/testinfra/infrastructure.go

@@ -10,6 +10,8 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"os"
+	"path/filepath"
 	"regexp"
 	"strings"
 	"testing"
@@ -29,10 +31,12 @@ import (
 	"google.golang.org/protobuf/types/known/durationpb"
 
 	"go.temporal.io/api/enums/v1"
+	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/operatorservice/v1"
 	"go.temporal.io/api/serviceerror"
 	"go.temporal.io/api/workflowservice/v1"
 	"go.temporal.io/sdk/client"
+	"google.golang.org/protobuf/encoding/protojson"
 )
 
 const (
@@ -428,6 +432,53 @@ system.forceSearchAttributesCacheRefreshOnRead:
 	infra.t.Log("Temporal client connected")
 }
 
+// SaveWorkflowHistory fetches the complete event history for a completed workflow
+// execution and writes it as a protojson file at destPath. The file can later be
+// loaded by a WorkflowReplayer replay test to guard against non-determinism.
+//
+// Typical usage (at the end of an integration test, after the workflow has finished):
+//
+//	infra.SaveWorkflowHistory(t, ctx, workflowID, "", "testdata/my-case/history.json")
+func (infra *TestInfrastructure) SaveWorkflowHistory(t *testing.T, ctx context.Context, workflowID, runID, destPath string) {
+	t.Helper()
+
+	iter := infra.TemporalClient.GetWorkflowHistory(ctx, workflowID, runID, false, enums.HISTORY_EVENT_FILTER_TYPE_ALL_EVENT)
+
+	var events []*historypb.HistoryEvent
+	for iter.HasNext() {
+		ev, err := iter.Next()
+		if err != nil {
+			t.Logf("Warning: error reading history event for %s: %v; skipping history save", workflowID, err)
+			return
+		}
+		events = append(events, ev)
+	}
+
+	if len(events) == 0 {
+		t.Logf("Warning: no history events found for workflow %s — skipping history save", workflowID)
+		return
+	}
+
+	history := &historypb.History{Events: events}
+	jsonBytes, err := protojson.MarshalOptions{Multiline: true}.Marshal(history)
+	if err != nil {
+		t.Logf("Warning: failed to marshal history for %s: %v", workflowID, err)
+		return
+	}
+
+	if err := os.MkdirAll(filepath.Dir(destPath), 0o755); err != nil {
+		t.Logf("Warning: failed to create history directory %s: %v", filepath.Dir(destPath), err)
+		return
+	}
+
+	if err := os.WriteFile(destPath, jsonBytes, 0o600); err != nil {
+		t.Logf("Warning: failed to write history file %s: %v", destPath, err)
+		return
+	}
+
+	t.Logf("Saved workflow history (%d events) → %s", len(events), destPath)
+}
+
 // RegisterCleanup adds a cleanup callback that will be called before container teardown.
 // Use this to gracefully shutdown Temporal workers and other resources.
 func (infra *TestInfrastructure) RegisterCleanup(cleanup func()) {