From ea6d2897b458aa58ec0bd7d811d64298c15d9f32 Mon Sep 17 00:00:00 2001 From: Emmanuel Maduwuba Date: Thu, 8 Oct 2026 11:08:27 -0400 Subject: [PATCH 1/2] Bump Temporal Go SDK to v1.16.0 and add a workflow replay test The deploy and PR workflows run for a long time and never continue as new, so their histories grow without bound. Fixing that needs workflow.GetInfo(ctx).GetCurrentHistoryLength(), which the SDK added in v1.16.0. This bumps go.temporal.io/sdk v1.15.0 -> v1.16.0 (and go.temporal.io/api v1.8.0 -> v1.11.0, gogo/status v1.1.0 -> v1.1.1) with no workflow code changes. An SDK bump or workflow change that alters the commands a workflow emits breaks workflows that are already running. TestReplayHistories replays exported histories against the current code to catch that. Histories hold production workflow inputs, so the test reads them from the directory in ATLANTIS_REPLAY_HISTORY_DIR and skips when it isn't set. The test registers workflows the way the worker does. The Deploy workflow and its plugins are moved into deployWorkflow() so the worker and the test share one definition. Current Temporal CLI exports name enums like EVENT_TYPE_WORKFLOW_EXECUTION_STARTED and include fields newer than this API version, so the test rewrites enum names and ignores unknown fields, as the worker does when it receives history as protobuf. --- go.mod | 8 +- go.sum | 18 +- server/neptune/temporalworker/replay_test.go | 185 +++++++++++++++++++ server/neptune/temporalworker/server.go | 19 +- 4 files changed, 210 insertions(+), 20 deletions(-) create mode 100644 server/neptune/temporalworker/replay_test.go diff --git a/go.mod b/go.mod index b71ab57d8..be28ca801 100644 --- a/go.mod +++ b/go.mod @@ -118,7 +118,7 @@ require ( github.com/twmb/murmur3 v1.1.6 // indirect ) -require go.temporal.io/sdk v1.15.0 +require go.temporal.io/sdk v1.16.0 require ( github.com/graymeta/stow v0.2.7 @@ -150,8 +150,8 @@ require ( github.com/asaskevich/govalidator v0.0.0-20210307081110-f21760c49a8d // indirect github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect github.com/gogo/googleapis v1.4.1 // indirect - github.com/gogo/protobuf v1.3.2 // indirect - github.com/gogo/status v1.1.0 // indirect + github.com/gogo/protobuf v1.3.2 + github.com/gogo/status v1.1.1 // indirect github.com/golang/mock v1.6.0 // indirect github.com/grpc-ecosystem/go-grpc-middleware v1.3.0 // indirect github.com/hashicorp/terraform-json v0.22.1 @@ -161,6 +161,6 @@ require ( github.com/robfig/cron v1.2.0 // indirect github.com/slack-go/slack v0.12.2 github.com/stretchr/objx v0.5.0 // indirect - go.temporal.io/api v1.8.0 + go.temporal.io/api v1.11.0 golang.org/x/sync v0.1.0 ) diff --git a/go.sum b/go.sum index 68638ad87..73a809a92 100644 --- a/go.sum +++ b/go.sum @@ -199,7 +199,6 @@ github.com/Masterminds/semver/v3 v3.1.1 h1:hLg3sBzpNErnxhQtUy/mmLR2I9foDujNK030I github.com/Masterminds/semver/v3 v3.1.1/go.mod h1:VPu/7SZ7ePZ3QOrcuXROw5FAcLl4a0cBrbBpGY/8hQs= github.com/Masterminds/sprig/v3 v3.2.2 h1:17jRggJu518dr3QaafizSXOjKYp94wKfABxUmyxvxX8= github.com/Masterminds/sprig/v3 v3.2.2/go.mod h1:UoaO7Yp8KlPnJIYWTFkMaqPUYKTfGFPhxNuwnnxkKlk= -github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAElWljhcU= github.com/agext/levenshtein v1.2.1/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki2W0IB5558= github.com/agext/levenshtein v1.2.2/go.mod h1:JEDfjyjHDjOF/1e4FlBE/PkbqA9OfWu2ki2W0IB5558= github.com/agext/levenshtein v1.2.3 h1:YB2fHEn0UJagG8T1rrWknE3ZQzWM06O8AMAatNn7lmo= @@ -246,7 +245,6 @@ github.com/cactus/go-statsd-client/statsd v0.0.0-20200423205355-cb0885a1018c/go. github.com/cactus/go-statsd-client/statsd v0.0.0-20200623234511-94959e3146b2 h1:GgJnJEJYymy/lx+1zXOO2TvGPRQJJ9vz4onxnA9gF3k= github.com/cactus/go-statsd-client/statsd v0.0.0-20200623234511-94959e3146b2/go.mod h1:l/bIBLeOl9eX+wxJAzxS4TveKRtAqlyDpHjhkfO0MEI= github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU= -github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc= github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cheekybits/is v0.0.0-20150225183255-68e9c0620927 h1:SKI1/fuSdodxmNNyVBR8d7X/HuLnRpvvFO0AgyQk764= github.com/cheekybits/is v0.0.0-20150225183255-68e9c0620927/go.mod h1:h/aW8ynjgkuj+NQRlZcDbAbM1ORAbXjXX77sX7T289U= @@ -311,8 +309,9 @@ github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7a github.com/gogo/protobuf v1.2.0/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= -github.com/gogo/status v1.1.0 h1:+eIkrewn5q6b30y+g/BJINVVdi2xH7je5MPJ3ZPK3JA= github.com/gogo/status v1.1.0/go.mod h1:BFv9nrluPLmrS0EmGVvLaPNmRosr9KapBYd5/hpY1WM= +github.com/gogo/status v1.1.1 h1:DuHXlSFHNKqTQ+/ACf5Vs6r4X/dH2EgIzR9Vr+H65kg= +github.com/gogo/status v1.1.1/go.mod h1:jpG3dM5QPcqu19Hg8lkUhBFBa3TcLs1DG7+2Jqci7oU= github.com/golang-jwt/jwt/v4 v4.4.1 h1:pC5DB52sCeK48Wlb9oPcdhnjkz1TKt1D/P7WKJ0kUcQ= github.com/golang-jwt/jwt/v4 v4.4.1/go.mod h1:m21LjoU+eqJr34lmDMbreY2eSTRJ1cv77w39/MY0Ch0= github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q= @@ -658,7 +657,6 @@ github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d h1:zE9ykE github.com/smartystreets/assertions v0.0.0-20180927180507-b2de0cb4f26d/go.mod h1:OnSkiWE9lh6wB0YB77sQom3nweQdgAjqCqsofrRNTgc= github.com/smartystreets/goconvey v1.6.4 h1:fv0U8FUIMPNf1L9lnHLvLhgicrIVChEkdzIKYqbNC9s= github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9QV7WQ/tjFTllLA= -github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/spf13/afero v1.6.0 h1:xoax2sJ2DT8S8xA2paPFjDCScCNeWsg75VG0DLRreiY= github.com/spf13/afero v1.6.0/go.mod h1:Ai8FlHk4v/PARR026UzYexafAt9roJ7LcLMAmO6Z93I= github.com/spf13/cast v1.3.1 h1:nFm6S0SMdyzrzcmThSipiEubIDy8WEXKNZ0UOgiRpng= @@ -734,11 +732,11 @@ go.opencensus.io v0.23.0 h1:gqCw0LfLxScz8irSi8exQc7fyQ0fKQU/qnC/X8+V/1M= go.opencensus.io v0.23.0/go.mod h1:XItmlyltB5F7CS4xOC1DcqMoFqwtC6OG2xF7mCv7P7E= go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqeYNgFYFoEGnI= go.temporal.io/api v1.5.0/go.mod h1:BqKxEJJYdxb5dqf0ODfzfMxh8UEQ5L3zKS51FiIYYkA= -go.temporal.io/api v1.8.0 h1:FzAMmBeLs6BEMFyHeJ9M9GAv6McFuH/GjnliBCdQ/Zw= -go.temporal.io/api v1.8.0/go.mod h1:7m1ZOVUFi/54a5IMzMeELnvDy5sJwRfz11zi3Jrww8w= +go.temporal.io/api v1.11.0 h1:RepwYaOZcEII7oKPC0QAUJHiep04NbwGRwMZ6Ofg0es= +go.temporal.io/api v1.11.0/go.mod h1:Z1uAa6iqyq/lu0+41qOG+BNO2JsxXxrgUe8UJv2iNZI= go.temporal.io/sdk v1.12.0/go.mod h1:lSp3lH1lI0TyOsus0arnO3FYvjVXBZGi/G7DjnAnm6o= -go.temporal.io/sdk v1.15.0 h1:1ZJEBNqLHAN0H64NpD4pydriYF9qhUIaimSVONm3ZKs= -go.temporal.io/sdk v1.15.0/go.mod h1:peqnjALtNpJMKRplWEubefPhDXdAtRTnebsLSFypSts= +go.temporal.io/sdk v1.16.0 h1:aWo7gJnrjGstqwRs/yfaIVIB3dX3oEak9A7f/Erb5nE= +go.temporal.io/sdk v1.16.0/go.mod h1:rnOV6mudCFB0n3/CJk/W9I89NVYuSjcCErrirzPGNP8= go.temporal.io/sdk/contrib/tally v0.1.0 h1:edAcGKNIDYU7fd10e4C/43dHw/h1F9cACupcmIKwzPI= go.temporal.io/sdk/contrib/tally v0.1.0/go.mod h1:PckZI8gA0AxIBvrgT2FQlm8TaqptYmqRdy2NxOibsZQ= go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE= @@ -861,11 +859,11 @@ golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su golang.org/x/net v0.0.0-20220325170049-de3da57026de/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/net v0.0.0-20220412020605-290c469a71a5/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk= -golang.org/x/net v0.0.0-20220531201128-c960675eff93/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.0.0-20220607020251-c690dde0001d/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.0.0-20220617184016-355a448f1bc9/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.0.0-20220624214902-1bab6f366d9e/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c= +golang.org/x/net v0.0.0-20220728181054-f92ba40d432d/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk= golang.org/x/net v0.0.0-20220909164309-bea034e7d591/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk= golang.org/x/net v0.0.0-20221014081412-f15817d10f9b/go.mod h1:YDH+HFinaLZZlnHAfSS6ZXJJ9M9t4Dl22yv3iI2vPwk= golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco= @@ -1229,13 +1227,13 @@ google.golang.org/genproto v0.0.0-20220502173005-c8bf987b8c21/go.mod h1:RAyBrSAP google.golang.org/genproto v0.0.0-20220505152158-f39f71e6c8f3/go.mod h1:RAyBrSAP7Fh3Nc84ghnVLDPuV51xc9agzmm4Ph6i0Q4= google.golang.org/genproto v0.0.0-20220518221133-4f43b3371335/go.mod h1:RAyBrSAP7Fh3Nc84ghnVLDPuV51xc9agzmm4Ph6i0Q4= google.golang.org/genproto v0.0.0-20220523171625-347a074981d8/go.mod h1:RAyBrSAP7Fh3Nc84ghnVLDPuV51xc9agzmm4Ph6i0Q4= -google.golang.org/genproto v0.0.0-20220602131408-e326c6e8e9c8/go.mod h1:yKyY4AMRwFiC8yMMNaMi+RkCnjZJt9LoWuvhXjMs+To= google.golang.org/genproto v0.0.0-20220608133413-ed9918b62aac/go.mod h1:KEWEmljWE5zPzLBa/oHl6DaEt9LmfH6WtH1OHIvleBA= google.golang.org/genproto v0.0.0-20220616135557-88e70c0c3a90/go.mod h1:KEWEmljWE5zPzLBa/oHl6DaEt9LmfH6WtH1OHIvleBA= google.golang.org/genproto v0.0.0-20220617124728-180714bec0ad/go.mod h1:KEWEmljWE5zPzLBa/oHl6DaEt9LmfH6WtH1OHIvleBA= google.golang.org/genproto v0.0.0-20220624142145-8cd45d7dbd1f/go.mod h1:KEWEmljWE5zPzLBa/oHl6DaEt9LmfH6WtH1OHIvleBA= google.golang.org/genproto v0.0.0-20220628213854-d9e0b6570c03/go.mod h1:KEWEmljWE5zPzLBa/oHl6DaEt9LmfH6WtH1OHIvleBA= google.golang.org/genproto v0.0.0-20220722212130-b98a9ff5e252/go.mod h1:GkXuJDJ6aQ7lnJcRF+SJVgFdQhypqgl3LB1C9vabdRE= +google.golang.org/genproto v0.0.0-20220725144611-272f38e5d71b/go.mod h1:iHe1svFLAZg9VWz891+QbRMwUv9O/1Ww+/mngYeThbc= google.golang.org/genproto v0.0.0-20220801145646-83ce21fca29f/go.mod h1:iHe1svFLAZg9VWz891+QbRMwUv9O/1Ww+/mngYeThbc= google.golang.org/genproto v0.0.0-20220815135757-37a418bb8959/go.mod h1:dbqgFATTzChvnt+ujMdZwITVAJHFtfyN1qUhDqEiIlk= google.golang.org/genproto v0.0.0-20220817144833-d7fd3f11b9b1/go.mod h1:dbqgFATTzChvnt+ujMdZwITVAJHFtfyN1qUhDqEiIlk= diff --git a/server/neptune/temporalworker/replay_test.go b/server/neptune/temporalworker/replay_test.go new file mode 100644 index 000000000..459f6a4a4 --- /dev/null +++ b/server/neptune/temporalworker/replay_test.go @@ -0,0 +1,185 @@ +package temporalworker + +import ( + "bytes" + "encoding/json" + "os" + "path/filepath" + "regexp" + "strings" + "testing" + + "github.com/gogo/protobuf/jsonpb" + lyftWorkflows "github.com/runatlantis/atlantis/server/neptune/lyft/workflows" + "github.com/runatlantis/atlantis/server/neptune/temporal" + "github.com/runatlantis/atlantis/server/neptune/workflows" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + "go.temporal.io/sdk/interceptor" + "go.temporal.io/sdk/worker" + "go.temporal.io/sdk/workflow" +) + +// replayHistoryDirEnv names a directory of workflow histories exported as JSON, +// e.g. with `temporal workflow show --output json`. Histories hold production +// workflow inputs, so they are read from a local directory rather than checked in. +const replayHistoryDirEnv = "ATLANTIS_REPLAY_HISTORY_DIR" + +// TestReplayHistories replays exported workflow histories against the current +// workflow code, registered the way the worker registers it. A failure means a +// code or SDK change would break workflows that are already running. +func TestReplayHistories(t *testing.T) { + dir := os.Getenv(replayHistoryDirEnv) + if dir == "" { + t.Skipf("set %s to a directory of exported workflow history JSON files", replayHistoryDirEnv) + } + + files, err := filepath.Glob(filepath.Join(dir, "*.json")) + require.NoError(t, err) + require.NotEmpty(t, files, "no *.json histories in %s", dir) + + replayer, err := worker.NewWorkflowReplayerWithOptions(worker.WorkflowReplayerOptions{ + Interceptors: []interceptor.WorkerInterceptor{temporal.NewWorkerInterceptor()}, + }) + require.NoError(t, err) + + replayer.RegisterWorkflowWithOptions(deployWorkflow(), workflow.RegisterOptions{Name: workflows.Deploy}) + replayer.RegisterWorkflow(workflows.PR) + replayer.RegisterWorkflow(workflows.Terraform) + replayer.RegisterWorkflow(lyftWorkflows.PRRevision) + + for _, f := range files { + f := f + t.Run(filepath.Base(f), func(t *testing.T) { + history, err := readHistory(f) + require.NoError(t, err) + require.NoError(t, replayer.ReplayWorkflowHistory(nil, history)) + }) + } +} + +// readHistory parses an exported history. Newer servers send fields this SDK's +// API version doesn't define; the worker ignores them when it receives history +// as protobuf, so they're ignored here too. +func readHistory(path string) (*historypb.History, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, err + } + converted, err := legacyEnumNames(raw) + if err != nil { + return nil, err + } + var history historypb.History + u := jsonpb.Unmarshaler{AllowUnknownFields: true} + if err := u.Unmarshal(bytes.NewReader(converted), &history); err != nil { + return nil, err + } + return &history, nil +} + +// historyEnums maps each enum's JSON prefix in current Temporal CLI exports +// (e.g. EVENT_TYPE_) to the value names the SDK's JSON parser accepts. +var historyEnums = map[string]map[string]int32{ + "ARCHIVAL_STATE_": enumspb.ArchivalState_value, + "CANCEL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_": enumspb.CancelExternalWorkflowExecutionFailedCause_value, + "COMMAND_TYPE_": enumspb.CommandType_value, + "CONTINUE_AS_NEW_INITIATOR_": enumspb.ContinueAsNewInitiator_value, + "ENCODING_TYPE_": enumspb.EncodingType_value, + "EVENT_TYPE_": enumspb.EventType_value, + "INDEXED_VALUE_TYPE_": enumspb.IndexedValueType_value, + "PARENT_CLOSE_POLICY_": enumspb.ParentClosePolicy_value, + "RETRY_STATE_": enumspb.RetryState_value, + "SIGNAL_EXTERNAL_WORKFLOW_EXECUTION_FAILED_CAUSE_": enumspb.SignalExternalWorkflowExecutionFailedCause_value, + "START_CHILD_WORKFLOW_EXECUTION_FAILED_CAUSE_": enumspb.StartChildWorkflowExecutionFailedCause_value, + "TASK_QUEUE_KIND_": enumspb.TaskQueueKind_value, + "TIMEOUT_TYPE_": enumspb.TimeoutType_value, + "WORKFLOW_EXECUTION_STATUS_": enumspb.WorkflowExecutionStatus_value, + "WORKFLOW_ID_REUSE_POLICY_": enumspb.WorkflowIdReusePolicy_value, + "WORKFLOW_TASK_FAILED_CAUSE_": enumspb.WorkflowTaskFailedCause_value, + "RESET_REAPPLY_TYPE_": enumspb.ResetReapplyType_value, + "SEVERITY_": enumspb.Severity_value, + "QUERY_REJECT_CONDITION_": enumspb.QueryRejectCondition_value, + "WORKFLOW_UPDATE_DURABILITY_PREFERENCE_": enumspb.WorkflowUpdateDurabilityPreference_value, + "WORKFLOW_UPDATE_RESULT_ACCESS_STYLE_": enumspb.WorkflowUpdateResultAccessStyle_value, + "SCHEDULE_OVERLAP_POLICY_": enumspb.ScheduleOverlapPolicy_value, + "PENDING_ACTIVITY_STATE_": enumspb.PendingActivityState_value, + "PENDING_WORKFLOW_TASK_STATE_": enumspb.PendingWorkflowTaskState_value, + "RESOURCE_EXHAUSTED_CAUSE_": enumspb.ResourceExhaustedCause_value, +} + +var screamingSnake = regexp.MustCompile(`^[A-Z][A-Z0-9_]*$`) + +// legacyEnumNames rewrites enum values in a history exported by a current +// Temporal CLI (EVENT_TYPE_WORKFLOW_EXECUTION_STARTED) to the names this SDK +// version's JSON parser accepts (WorkflowExecutionStarted). Other values are +// left alone. +func legacyEnumNames(raw []byte) ([]byte, error) { + var doc interface{} + if err := json.Unmarshal(raw, &doc); err != nil { + return nil, err + } + return json.Marshal(rewriteEnums(doc)) +} + +func rewriteEnums(v interface{}) interface{} { + switch t := v.(type) { + case map[string]interface{}: + for k, child := range t { + t[k] = rewriteEnums(child) + } + return t + case []interface{}: + for i, child := range t { + t[i] = rewriteEnums(child) + } + return t + case string: + return legacyEnumName(t) + default: + return v + } +} + +func legacyEnumName(s string) string { + if !screamingSnake.MatchString(s) { + return s + } + for prefix, values := range historyEnums { + if !strings.HasPrefix(s, prefix) { + continue + } + name := camelCase(strings.TrimPrefix(s, prefix)) + if _, ok := values[name]; ok { + return name + } + } + return s +} + +func camelCase(s string) string { + var b strings.Builder + for _, part := range strings.Split(strings.ToLower(s), "_") { + if part != "" { + b.WriteString(strings.ToUpper(part[:1]) + part[1:]) + } + } + return b.String() +} + +func TestLegacyEnumNames(t *testing.T) { + in := []byte(`{"events":[{"eventId":"1","eventType":"EVENT_TYPE_WORKFLOW_EXECUTION_STARTED",` + + `"workflowExecutionStartedEventAttributes":{"taskQueue":{"name":"deploy","kind":"TASK_QUEUE_KIND_NORMAL"},` + + `"parentClosePolicy":"PARENT_CLOSE_POLICY_REQUEST_CANCEL","identity":"KEEP_ME"}}]}`) + + out, err := legacyEnumNames(in) + require.NoError(t, err) + + s := string(out) + assert.Contains(t, s, `"eventType":"WorkflowExecutionStarted"`) + assert.Contains(t, s, `"kind":"Normal"`) + assert.Contains(t, s, `"parentClosePolicy":"RequestCancel"`) + assert.Contains(t, s, `"identity":"KEEP_ME"`) +} diff --git a/server/neptune/temporalworker/server.go b/server/neptune/temporalworker/server.go index a35c283e4..62d9a9f9a 100644 --- a/server/neptune/temporalworker/server.go +++ b/server/neptune/temporalworker/server.go @@ -409,7 +409,18 @@ func (s Server) buildDeployWorker() worker.Worker { deployWorker.RegisterActivity(s.GithubActivities) deployWorker.RegisterActivity(s.AuditActivity) deployWorker.RegisterActivity(s.TerraformActivities) - deployWorker.RegisterWorkflowWithOptions(workflows.GetDeployWithPlugins( + deployWorker.RegisterWorkflowWithOptions(deployWorkflow(), workflow.RegisterOptions{ + Name: workflows.Deploy, + }) + deployWorker.RegisterWorkflow(workflows.Terraform) + return deployWorker +} + +// deployWorkflow returns the Deploy workflow with the plugins the deploy worker +// runs it with. Replay tests register the same function so they replay what +// production executes. +func deployWorkflow() workflows.DeployFunc { + return workflows.GetDeployWithPlugins( func(ctx workflow.Context, dr workflows.DeployRequest) (plugins.Deploy, error) { var a *lyftActivities.Audit @@ -424,11 +435,7 @@ func (s Server) buildDeployWorker() worker.Worker { }, }, nil }, - ), workflow.RegisterOptions{ - Name: workflows.Deploy, - }) - deployWorker.RegisterWorkflow(workflows.Terraform) - return deployWorker + ) } func (s Server) buildTerraformWorker() worker.Worker { From 48718b8cd7ec11b3a2721a69d1ca46cc20a3f781 Mon Sep 17 00:00:00 2001 From: Emmanuel Maduwuba Date: Thu, 8 Oct 2026 11:33:13 -0400 Subject: [PATCH 2/2] Read replay histories through os.Root gosec G703 flagged os.ReadFile on a path built from ATLANTIS_REPLAY_HISTORY_DIR. Opening the directory with os.OpenRoot confines every read to it, so a history name can't reach outside the directory. --- server/neptune/temporalworker/replay_test.go | 21 ++++++++++++-------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/server/neptune/temporalworker/replay_test.go b/server/neptune/temporalworker/replay_test.go index 459f6a4a4..6f0bbc823 100644 --- a/server/neptune/temporalworker/replay_test.go +++ b/server/neptune/temporalworker/replay_test.go @@ -3,8 +3,8 @@ package temporalworker import ( "bytes" "encoding/json" + "io/fs" "os" - "path/filepath" "regexp" "strings" "testing" @@ -36,7 +36,12 @@ func TestReplayHistories(t *testing.T) { t.Skipf("set %s to a directory of exported workflow history JSON files", replayHistoryDirEnv) } - files, err := filepath.Glob(filepath.Join(dir, "*.json")) + // Reads go through os.Root, so a history name can't reach outside dir. + root, err := os.OpenRoot(dir) + require.NoError(t, err) + defer root.Close() + + files, err := fs.Glob(root.FS(), "*.json") require.NoError(t, err) require.NotEmpty(t, files, "no *.json histories in %s", dir) @@ -50,10 +55,10 @@ func TestReplayHistories(t *testing.T) { replayer.RegisterWorkflow(workflows.Terraform) replayer.RegisterWorkflow(lyftWorkflows.PRRevision) - for _, f := range files { - f := f - t.Run(filepath.Base(f), func(t *testing.T) { - history, err := readHistory(f) + for _, name := range files { + name := name + t.Run(name, func(t *testing.T) { + history, err := readHistory(root, name) require.NoError(t, err) require.NoError(t, replayer.ReplayWorkflowHistory(nil, history)) }) @@ -63,8 +68,8 @@ func TestReplayHistories(t *testing.T) { // readHistory parses an exported history. Newer servers send fields this SDK's // API version doesn't define; the worker ignores them when it receives history // as protobuf, so they're ignored here too. -func readHistory(path string) (*historypb.History, error) { - raw, err := os.ReadFile(path) +func readHistory(root *os.Root, name string) (*historypb.History, error) { + raw, err := root.ReadFile(name) if err != nil { return nil, err }