From 321aae950cb0341e7f5cf00bf1bbd5720d3f09b9 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 25 Sep 2026 08:04:30 -0700 Subject: [PATCH 1/5] Re-pinned api-go to the documented stream protos. The branch builds against the fork's api, which pulls newer gRPC and OpenTelemetry through the module graph and ships the nexusoperation and stream protos the import map has to list. --- Makefile | 6 ++- chasm/lib/tests/nexus_service.go | 20 +++++-- cmd/tools/getproto/files.go | 4 ++ common/metrics/otel_metrics_handler_test.go | 5 ++ go.mod | 20 +++---- go.sum | 42 ++++++++------- service/history/handler.go | 43 +++++++++++++++ service/history/handler_test.go | 28 ++++++++++ tests/api_fork_test.go | 59 +++++++++++++++++++++ tests/mixedbrain/go.mod | 6 +-- tests/mixedbrain/go.sum | 20 +++---- tests/nexus_workflow_test.go | 19 +++++-- 12 files changed, 222 insertions(+), 50 deletions(-) create mode 100644 tests/api_fork_test.go diff --git a/Makefile b/Makefile index ccb299f7f22..5f5625487a2 100644 --- a/Makefile +++ b/Makefile @@ -522,14 +522,18 @@ workflowcheck: $(WORKFLOWCHECK) check: lint shell-check ##### Tests ##### +# verify-test-log reads the whole file, so a log left behind by an earlier run +# would be judged as if it belonged to this one. clean-test-output: @printf $(COLOR) "Delete test output..." @rm -rf $(TEST_OUTPUT_ROOT) + @rm -f test.log @go clean -testcache build-tests: @printf $(COLOR) "Build tests..." - @CGO_ENABLED=$(CGO_ENABLED) go test $(TEST_TAG_FLAG) -exec="true" -count=0 $(TEST_DIRS) + @CGO_ENABLED=$(CGO_ENABLED) go test $(TEST_TAG_FLAG) -exec="true" -count=0 $(filter-out $(MIXED_BRAIN_TEST_ROOT)%,$(TEST_DIRS)) + @cd $(MIXED_BRAIN_TEST_ROOT) && CGO_ENABLED=1 go test $(TEST_TAG_FLAG) -exec="true" -count=0 ./... unit-test: clean-test-output @printf $(COLOR) "Run unit tests..." diff --git a/chasm/lib/tests/nexus_service.go b/chasm/lib/tests/nexus_service.go index 0ae0d42905f..386073e1ec6 100644 --- a/chasm/lib/tests/nexus_service.go +++ b/chasm/lib/tests/nexus_service.go @@ -9,9 +9,14 @@ import ( "go.temporal.io/server/common/payload" ) -var TestOperation = nexus.NewSyncOperation("TestOperation", func(ctx context.Context, input string, options nexus.StartOperationOptions) (string, error) { - return "Hello, " + input, nil -}) +// TestOperation returns a failurepb.Failure so that it is a proper protobuf type. Its used for +// testing the System Nexus Endpoint, which only accepts protobuf-encoded payloads. +var TestOperation = nexus.NewSyncOperation( + "TestOperation", + func(ctx context.Context, input string, options nexus.StartOperationOptions) (*commonpb.DataBlob, error) { + d := []byte("Hello, " + input) + return &commonpb.DataBlob{Data: d}, nil + }) // TestOperationWithPayload is identical to TestOperation, except its response embeds a // nested *commonpb.Payload. It exists to exercise the commonnexus.SystemPayloadMetadataKey @@ -20,10 +25,18 @@ var TestOperationWithPayload = nexus.NewSyncOperation("TestOperationWithPayload" return &commonpb.Payloads{Payloads: []*commonpb.Payload{payload.EncodeString("Hello, " + input)}}, nil }) +// TestOperationStringOutput returns a string, which the data converter encodes as JSON rather +// than protobuf. It exists to exercise the System Nexus Endpoint's rejection of non-protobuf +// responses in service/history/handler.go's StartNexusOperation. +var TestOperationStringOutput = nexus.NewSyncOperation("TestOperationStringOutput", func(ctx context.Context, input string, options nexus.StartOperationOptions) (string, error) { + return "Hello, " + input, nil +}) + func NewTestServiceNexusService() *nexus.Service { service := nexus.NewService("TestService") service.MustRegister(TestOperation) service.MustRegister(TestOperationWithPayload) + service.MustRegister(TestOperationStringOutput) return service } @@ -43,5 +56,6 @@ func NewTestServiceNexusServiceProcessor() *chasm.NexusServiceProcessor { sp := chasm.NewNexusServiceProcessor("TestService") sp.MustRegisterOperation("TestOperation", chasm.NewRegisterableNexusOperationProcessor(testOperationProcessor{})) sp.MustRegisterOperation("TestOperationWithPayload", chasm.NewRegisterableNexusOperationProcessor(testOperationProcessor{})) + sp.MustRegisterOperation("TestOperationStringOutput", chasm.NewRegisterableNexusOperationProcessor(testOperationProcessor{})) return sp } diff --git a/cmd/tools/getproto/files.go b/cmd/tools/getproto/files.go index 38b6ef67622..3b1e31ca412 100644 --- a/cmd/tools/getproto/files.go +++ b/cmd/tools/getproto/files.go @@ -21,12 +21,14 @@ import ( history "go.temporal.io/api/history/v1" namespace "go.temporal.io/api/namespace/v1" nexus "go.temporal.io/api/nexus/v1" + nexusoperation "go.temporal.io/api/nexusoperation/v1" protocol "go.temporal.io/api/protocol/v1" query "go.temporal.io/api/query/v1" replication "go.temporal.io/api/replication/v1" rules "go.temporal.io/api/rules/v1" schedule "go.temporal.io/api/schedule/v1" sdk "go.temporal.io/api/sdk/v1" + stream "go.temporal.io/api/stream/v1" taskqueue "go.temporal.io/api/taskqueue/v1" update "go.temporal.io/api/update/v1" version "go.temporal.io/api/version/v1" @@ -82,6 +84,7 @@ func init() { importMap["temporal/api/history/v1/message.proto"] = history.File_temporal_api_history_v1_message_proto importMap["temporal/api/namespace/v1/message.proto"] = namespace.File_temporal_api_namespace_v1_message_proto importMap["temporal/api/nexus/v1/message.proto"] = nexus.File_temporal_api_nexus_v1_message_proto + importMap["temporal/api/nexusoperation/v1/message.proto"] = nexusoperation.File_temporal_api_nexusoperation_v1_message_proto importMap["temporal/api/protocol/v1/message.proto"] = protocol.File_temporal_api_protocol_v1_message_proto importMap["temporal/api/query/v1/message.proto"] = query.File_temporal_api_query_v1_message_proto importMap["temporal/api/replication/v1/message.proto"] = replication.File_temporal_api_replication_v1_message_proto @@ -91,6 +94,7 @@ func init() { importMap["temporal/api/sdk/v1/task_complete_metadata.proto"] = sdk.File_temporal_api_sdk_v1_task_complete_metadata_proto importMap["temporal/api/sdk/v1/user_metadata.proto"] = sdk.File_temporal_api_sdk_v1_user_metadata_proto importMap["temporal/api/sdk/v1/worker_config.proto"] = sdk.File_temporal_api_sdk_v1_worker_config_proto + importMap["temporal/api/stream/v1/message.proto"] = stream.File_temporal_api_stream_v1_message_proto importMap["temporal/api/taskqueue/v1/message.proto"] = taskqueue.File_temporal_api_taskqueue_v1_message_proto importMap["temporal/api/update/v1/message.proto"] = update.File_temporal_api_update_v1_message_proto importMap["temporal/api/version/v1/message.proto"] = version.File_temporal_api_version_v1_message_proto diff --git a/common/metrics/otel_metrics_handler_test.go b/common/metrics/otel_metrics_handler_test.go index e186e79ec9f..9cfc5aa0acb 100644 --- a/common/metrics/otel_metrics_handler_test.go +++ b/common/metrics/otel_metrics_handler_test.go @@ -195,6 +195,10 @@ func TestMeter(t *testing.T) { cmpopts.IgnoreFields(metricdata.DataPoint[int64]{}, "StartTime", "Time"), cmpopts.IgnoreFields(metricdata.DataPoint[float64]{}, "StartTime", "Time"), cmpopts.IgnoreFields(metricdata.HistogramDataPoint[int64]{}, "StartTime", "Time", "Bounds"), + // The SDK hands back either a nil or an allocated-but-empty exemplar + // slice depending on whether its reservoir was ever touched, so the + // two have to compare equal here. + cmpopts.EquateEmpty(), ); diff != "" { t.Errorf("mismatch (-want, +got):\n%s", diff) } @@ -263,6 +267,7 @@ func TestMeter_TimerInSeconds(t *testing.T) { return a1.Equals(&a2) }), cmpopts.IgnoreFields(metricdata.HistogramDataPoint[float64]{}, "StartTime", "Time", "Bounds"), + cmpopts.EquateEmpty(), ); diff != "" { t.Errorf("mismatch (-want, +got):\n%s", diff) } diff --git a/go.mod b/go.mod index 7a9d0965bba..2be924061ca 100644 --- a/go.mod +++ b/go.mod @@ -63,8 +63,8 @@ require ( go.opentelemetry.io/otel/exporters/prometheus v0.56.0 go.opentelemetry.io/otel/log v0.20.0 go.opentelemetry.io/otel/metric v1.44.0 - go.opentelemetry.io/otel/sdk v1.43.0 - go.opentelemetry.io/otel/sdk/metric v1.43.0 + go.opentelemetry.io/otel/sdk v1.44.0 + go.opentelemetry.io/otel/sdk/metric v1.44.0 go.opentelemetry.io/otel/trace v1.44.0 go.temporal.io/api v1.63.5 go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab @@ -81,7 +81,7 @@ require ( golang.org/x/text v0.40.0 golang.org/x/time v0.15.0 google.golang.org/api v0.276.0 - google.golang.org/grpc v1.80.0 + google.golang.org/grpc v1.83.1 google.golang.org/protobuf v1.36.11 gopkg.in/validator.v2 v2.0.1 gopkg.in/yaml.v3 v3.0.1 @@ -109,7 +109,7 @@ require ( ) require ( - cel.dev/expr v0.25.1 // indirect + cel.dev/expr v0.25.2 // indirect cloud.google.com/go v0.123.0 // indirect; indirect e cloud.google.com/go/auth v0.20.0 // indirect cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect @@ -120,7 +120,7 @@ require ( cloud.google.com/go/run v1.19.0 // indirect dario.cat/mergo v1.0.2 // indirect filippo.io/edwards25519 v1.2.0 // indirect - github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.32.0 // indirect + github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.33.0 // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.56.0 // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/resourcemapping v0.56.0 // indirect github.com/Masterminds/goutils v1.1.1 // indirect @@ -199,14 +199,14 @@ require ( github.com/sirupsen/logrus v1.9.4 // indirect github.com/spf13/cast v1.10.0 // indirect github.com/spf13/pflag v1.0.10 // indirect - github.com/spiffe/go-spiffe/v2 v2.6.0 // indirect + github.com/spiffe/go-spiffe/v2 v2.7.0 // indirect github.com/stretchr/objx v0.5.3 // indirect github.com/twmb/murmur3 v1.1.8 // indirect github.com/uber-common/bark v1.3.0 // indirect github.com/x448/float16 v0.8.4 // indirect github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.opentelemetry.io/contrib/detectors/gcp v1.43.0 // indirect + go.opentelemetry.io/contrib/detectors/gcp v1.44.0 // indirect go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.43.0 go.opentelemetry.io/proto/otlp v1.10.0 @@ -219,8 +219,8 @@ require ( golang.org/x/sys v0.47.0 // indirect golang.org/x/term v0.45.0 // indirect google.golang.org/genproto v0.0.0-20260420184626-e10c466a9529 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260420184626-e10c466a9529 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect k8s.io/api v0.35.4 // indirect @@ -239,3 +239,5 @@ require ( ) tool golang.org/x/perf/cmd/benchstat + +replace go.temporal.io/api => github.com/moedash/api-go v1.63.6-0.20260921082525-9a6aa5f95be8 diff --git a/go.sum b/go.sum index 95797622c25..13b45018247 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4= -cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= +cel.dev/expr v0.25.2 h1:K6j46C81hXtZQfuX60cVWQFBJahKSE2gfRbNuvr5bFs= +cel.dev/expr v0.25.2/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE= cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU= cloud.google.com/go/auth v0.20.0 h1:kXTssoVb4azsVDoUiF8KvxAqrsQcQtB53DcSgta74CA= @@ -31,8 +31,8 @@ filippo.io/edwards25519 v1.2.0/go.mod h1:xzAOLCNug/yB62zG1bQ8uziwrIqIuxhctzJT18Q github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU= github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= github.com/BurntSushi/xgb v0.0.0-20160522181843-27f122750802/go.mod h1:IVnqGOEym/WlBOVXweHU+Q+/VP0lqqI8lqeDx9IjBqo= -github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.32.0 h1:rIkQfkCOVKc1OiRCNcSDD8ml5RJlZbH/Xsq7lbpynwc= -github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.32.0/go.mod h1:RD2SsorTmYhF6HkTmDw7KmPYQk8OBYwTkuasChwv7R4= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.33.0 h1:l7+6kwRMJNwdCvYdDl7Eax+wzEYHSnNY7zrrfbhDdTA= +github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.33.0/go.mod h1:pJTkW8hEUIIi3Pf65lPZOnn4Y81yCllX6IWk2jNXdkM= github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.56.0 h1:O2sXMyJh8b7devAGdE+163xtRurt0RVpB6DIzX5vGfg= github.com/GoogleCloudPlatform/opentelemetry-operations-go/exporter/metric v0.56.0/go.mod h1:hEpiGU18xf70qb3jbTcIggWAiEfX/cOIVc2OTe4OegA= github.com/GoogleCloudPlatform/opentelemetry-operations-go/internal/cloudmock v0.56.0 h1:ZIT85vKP7LBS84XJ0WdJ3dPOX3iz4j3c0+lpajGQMyo= @@ -321,6 +321,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/moedash/api-go v1.63.6-0.20260921082525-9a6aa5f95be8 h1:WdG6n/8Da2RC5y3t3/vdpq0P4wOBbOQX+C+C9TiChew= +github.com/moedash/api-go v1.63.6-0.20260921082525-9a6aa5f95be8/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= @@ -382,8 +384,8 @@ github.com/spf13/cast v1.10.0 h1:h2x0u2shc1QuLHfxi+cTJvs30+ZAHOGRic8uyGTDWxY= github.com/spf13/cast v1.10.0/go.mod h1:jNfB8QC9IA6ZuY2ZjDp0KtFO2LZZlg4S/7bzP6qqeHo= github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= -github.com/spiffe/go-spiffe/v2 v2.6.0 h1:l+DolpxNWYgruGQVV0xsfeya3CsC7m8iBzDnMpsbLuo= -github.com/spiffe/go-spiffe/v2 v2.6.0/go.mod h1:gm2SeUoMZEtpnzPNs2Csc0D/gX33k1xIx7lEzqblHEs= +github.com/spiffe/go-spiffe/v2 v2.7.0 h1:uXe1MflJoHw58wAUvxVlcM7WpKtijWG7I1UidcGh6g4= +github.com/spiffe/go-spiffe/v2 v2.7.0/go.mod h1:47Q0Q9/AqGha8QLHp+kxpH4Wca7X7EnOtlIJy3mxZ3U= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= @@ -443,8 +445,8 @@ go.opentelemetry.io/collector/internal/testutil v0.150.0 h1:J4PLQGPfbLVaL5eI1aMc go.opentelemetry.io/collector/internal/testutil v0.150.0/go.mod h1:Jkjs6rkqs973LqgZ0Fe3zrokQRKULYXPIf4HuqStiEE= go.opentelemetry.io/collector/pdata v1.56.0 h1:W+QAfN2Iz8SNss1T5JNzRWFnw+7oP1vXBQH9ZuOJkXY= go.opentelemetry.io/collector/pdata v1.56.0/go.mod h1:usR9utboXufbD1rp1oJy+3smQXXpZ+CsI3WN7QsiOs0= -go.opentelemetry.io/contrib/detectors/gcp v1.43.0 h1:62yY3dT7/ShwOxzA0RsKRgshBmfElKI4d/Myu2OxDFU= -go.opentelemetry.io/contrib/detectors/gcp v1.43.0/go.mod h1:RyaZMFY7yi1kAs45S6mbFGz8O8rqB0dTY14uzvG4LCs= +go.opentelemetry.io/contrib/detectors/gcp v1.44.0 h1:NmLfL734pJhM0JKaYd2Y28+nY9dPRWYAAbxhRCrKXPw= +go.opentelemetry.io/contrib/detectors/gcp v1.44.0/go.mod h1:tNAsgd8avTGke1+MndXlU5Cru4PQ9Ai/cCNWQv/ZJ/s= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0 h1:0Qx7VGBacMm9ZENQ7TnNObTYI4ShC+lHI16seduaxZo= go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.68.0/go.mod h1:Sje3i3MjSPKTSPvVWCaL8ugBzJwik3u4smCjUeuupqg= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 h1:CqXxU8VOmDefoh0+ztfGaymYbhdB/tT3zs79QaZTNGY= @@ -465,10 +467,12 @@ go.opentelemetry.io/otel/log v0.20.0 h1:/5i0vuHxCLWUfChWG41K9wkM0jafruPw9NU1/RCJ go.opentelemetry.io/otel/log v0.20.0/go.mod h1:wOcMcjsZpG8x7Bak7IhSi/lg8wscV2C1VdrKCLPlt0E= go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= -go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= -go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= -go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= -go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/metric/x v0.66.0 h1:YkCrx1zLOChi9ZcZ6euupOcsgzbVlec7D/xoEU1+cTA= +go.opentelemetry.io/otel/metric/x v0.66.0/go.mod h1:d1+BDj9t96do0/1LoU1ayfCv79ZgNE41qbhBvnMOBZk= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= @@ -479,8 +483,6 @@ go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0 h1:R go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.3.0/go.mod h1:I89cynRj8y+383o7tEQVg2SVA6SRgDVIouWPUVXjx0U= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0 h1:CQvJSldHRUN6Z8jsUeYv8J0lXRvygALXIzsmAeCcZE0= go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.3.0/go.mod h1:xSQ+mEfJe/GjK1LXEyVOoSI1N9JV9ZI923X5kup43W4= -go.temporal.io/api v1.63.5 h1:c11+kPYHkXXL3UiShPdbMD+xtvqGsbTibUA9ypmiCa4= -go.temporal.io/api v1.63.5/go.mod h1:SrlW2JMwVlDP4nRWSNznUFqnSHd+YeMDS1BkYo63HCQ= go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab h1:99wXW0317BBi49d6xgMdA0EZtvA+xbBUWV4HsTEGEcg= go.temporal.io/auto-scaled-workers v0.0.0-20260811170210-91f6fe1d10ab/go.mod h1:hhHijO9XRPIkAflLJJHix61M9FzbRPqk8fSydkcLkqw= go.temporal.io/sdk v1.44.0 h1:suitPDukX74rW3/N1FqvEbZTZVJJsxMKhv0KMa/j7pU= @@ -632,12 +634,12 @@ google.golang.org/api v0.276.0 h1:nVArUtfLEihtW+b0DdcqRGK1xoEm2+ltAihyztq7MKY= google.golang.org/api v0.276.0/go.mod h1:Fnag/EWUPIcJXuIkP1pjoTgS5vdxlk3eeemL7Do6bvw= google.golang.org/genproto v0.0.0-20260420184626-e10c466a9529 h1:QoMBg0moLIlB/eucPzc+ID5SgPZWuirtjAn3l8nW2Dg= google.golang.org/genproto v0.0.0-20260420184626-e10c466a9529/go.mod h1:EjLmDZ8liSLBrCTK5vP+bGIxRQHE3ovGvOI0CzGk1PI= -google.golang.org/genproto/googleapis/api v0.0.0-20260420184626-e10c466a9529 h1:zUWMZsvo/IJcD1t6MNCPO/azZTwz0TvwCBqr5aifoVY= -google.golang.org/genproto/googleapis/api v0.0.0-20260420184626-e10c466a9529/go.mod h1:a5OGAgyRr4lqco7AG9hQM9Fwh0N2ZV4grR0eXFEsXQg= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529 h1:XF8+t6QQiS0o9ArVan/HW8Q7cycNPGsJf6GA2nXxYAg= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM= -google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.83.1 h1:HIO0+BEtBP6soyqvqC8sNUjZ7bTs+0hFQuFF+RAy++Y= +google.golang.org/grpc v1.83.1/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/service/history/handler.go b/service/history/handler.go index d25e52b32bc..f63609d89f9 100644 --- a/service/history/handler.go +++ b/service/history/handler.go @@ -880,6 +880,33 @@ func (h *Handler) GetMutableState(ctx context.Context, request *historyservice.G return resp, nil } +// GetStreamReplaySlices re-supplies the stream ranges a workflow's completed tasks recorded. +func (h *Handler) GetStreamReplaySlices( + ctx context.Context, + request *historyservice.GetStreamReplaySlicesRequest, +) (*historyservice.GetStreamReplaySlicesResponse, error) { + namespaceID := namespace.ID(request.GetNamespaceId()) + if namespaceID == "" { + return nil, h.convertError(errNamespaceNotSet) + } + + workflowID := request.GetExecution().GetWorkflowId() + shardContext, err := h.controller.GetShardByNamespaceWorkflow(namespaceID, workflowID) + if err != nil { + return nil, h.convertError(err) + } + engine, err := shardContext.GetEngine(ctx) + if err != nil { + return nil, h.convertError(err) + } + + resp, err := engine.GetStreamReplaySlices(ctx, request) + if err != nil { + return nil, h.convertError(err) + } + return resp, nil +} + // PollMutableState - returns the id of the next event in the execution's history func (h *Handler) PollMutableState(ctx context.Context, request *historyservice.PollMutableStateRequest) (*historyservice.PollMutableStateResponse, error) { namespaceID := namespace.ID(request.GetNamespaceId()) @@ -2540,6 +2567,9 @@ func (h *Handler) UnpauseWorkflowExecution(ctx context.Context, request *history return unpauseResp, nil } +// StartNexusOperation is the History Service's StartNexusOperation endpoint is for dispatching requests +// to the System Nexus Endpoint, distinct from the Frontend Service's `nexus_handler.go` which +// starts Nexus operations by sending Nexus tasks directly to workers via the Matching Service. func (h *Handler) StartNexusOperation( ctx context.Context, req *historyservice.StartNexusOperationRequest, @@ -2602,16 +2632,29 @@ func (h *Handler) StartNexusOperation( h.logger.Error("failed to encode payload", tag.Error(err), tag.RequestID(requestID)) return nil, serviceerror.NewInternal("internal error (request ID: " + requestID + ")") } + var payload *commonpb.Payload if len(ps.GetPayloads()) == 1 { payload = ps.GetPayloads()[0] } + // Responses from the System Nexus Endpoint are server generated, so we must mark them as system payloads. if payload != nil { if payload.Metadata == nil { payload.Metadata = make(map[string][]byte, 1) } + + // Responses from the System Nexus Endpoint have to be protobufs. + encoding := string(payload.Metadata["encoding"]) + if encoding != "binary/protobuf" { + return nil, serviceerror.NewFailedPreconditionf("system payload must be encoded as binary/protobuf but got %s", encoding) + } + if _, ok := payload.Metadata["messageType"]; !ok { + return nil, serviceerror.NewFailedPrecondition("system payload missing messageType metadata key") + } + payload.Metadata[commonnexus.SystemPayloadMetadataKey] = []byte("true") } + response.Variant = &nexuspb.StartOperationResponse_SyncSuccess{ SyncSuccess: &nexuspb.StartOperationResponse_Sync{ Payload: payload, diff --git a/service/history/handler_test.go b/service/history/handler_test.go index e0c3df1b9f0..3394f59206e 100644 --- a/service/history/handler_test.go +++ b/service/history/handler_test.go @@ -450,3 +450,31 @@ func TestStartNexusOperation_SystemNexusEndpointPayloadMetadataFlag(t *testing.T }) } } + +func TestStartNexusOperation_SystemNexusEndpointRejectsNonProtoResponse(t *testing.T) { + registry := nexus.NewServiceRegistry() + registry.MustRegister(chasmtests.NewTestServiceNexusService()) + nexusHandler, err := registry.NewHandler() + require.NoError(t, err) + + h := Handler{ + logger: log.NewNoopLogger(), + nexusHandler: nexusHandler, + } + + // TestOperationStringOutput returns a string, which the data converter encodes as JSON + // instead of protobuf. The System Nexus Endpoint only accepts protobuf-encoded responses. + resp, err := h.StartNexusOperation(context.Background(), &historyservice.StartNexusOperationRequest{ + Request: &nexuspb.StartOperationRequest{ + Service: "TestService", + Operation: "TestOperationStringOutput", + RequestId: "test-request-id", + Payload: payload.EncodeString("Temporal"), + }, + }) + require.Nil(t, resp) + + var failedPrecondition *serviceerror.FailedPrecondition + require.ErrorAs(t, err, &failedPrecondition) + require.ErrorContains(t, err, "system payload must be encoded as binary/protobuf but got json/plain") +} diff --git a/tests/api_fork_test.go b/tests/api_fork_test.go new file mode 100644 index 00000000000..8e39bb7109b --- /dev/null +++ b/tests/api_fork_test.go @@ -0,0 +1,59 @@ +package tests + +import ( + "testing" + + "github.com/stretchr/testify/require" + commandpb "go.temporal.io/api/command/v1" + enumspb "go.temporal.io/api/enums/v1" + historypb "go.temporal.io/api/history/v1" + streampb "go.temporal.io/api/stream/v1" + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/protobuf/proto" +) + +// The api fork has to round-trip the new shapes over the wire, not merely +// compile. A field added without its descriptor would compile and silently +// drop on marshal. +func TestApiForkCarriesStreamShapes(t *testing.T) { + require.Equal(t, enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, enumspb.CommandType(19)) + + cmd := &commandpb.Command{ + CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, + Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ + AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ + StreamId: "s1", + Records: []*streampb.StreamRecord{{Topic: "tokens"}}, + }, + }, + } + b, err := proto.Marshal(cmd) + require.NoError(t, err) + var back commandpb.Command + require.NoError(t, proto.Unmarshal(b, &back)) + require.Equal(t, "s1", back.GetAppendStreamRecordsCommandAttributes().GetStreamId()) + require.Equal(t, "tokens", + back.GetAppendStreamRecordsCommandAttributes().GetRecords()[0].GetTopic()) + + resp := &workflowservice.PollWorkflowTaskQueueResponse{ + StreamSlices: []*streampb.StreamSlice{{StreamId: "s1", FromOffset: 4, ToOffset: 7}}, + } + rb, err := proto.Marshal(resp) + require.NoError(t, err) + var rback workflowservice.PollWorkflowTaskQueueResponse + require.NoError(t, proto.Unmarshal(rb, &rback)) + require.Equal(t, int64(7), rback.GetStreamSlices()[0].GetToOffset()) + + attrs := &historypb.WorkflowTaskCompletedEventAttributes{ + ConsumedStreamRanges: []*streampb.StreamRange{{StreamId: "s1", FromOffset: 4, ToOffset: 4}}, + } + ab, err := proto.Marshal(attrs) + require.NoError(t, err) + var aback historypb.WorkflowTaskCompletedEventAttributes + require.NoError(t, proto.Unmarshal(ab, &aback)) + // An empty range has to survive the round trip: it is the fact that a + // subscription observed nothing, which replay must reproduce. + require.Len(t, aback.GetConsumedStreamRanges(), 1) + require.Equal(t, + aback.GetConsumedStreamRanges()[0].GetFromOffset(), aback.GetConsumedStreamRanges()[0].GetToOffset()) +} diff --git a/tests/mixedbrain/go.mod b/tests/mixedbrain/go.mod index dde79a7698e..c87866e7f25 100644 --- a/tests/mixedbrain/go.mod +++ b/tests/mixedbrain/go.mod @@ -9,7 +9,7 @@ require ( github.com/temporalio/omes v0.0.0-20260529203146-c6ee1f56c726 go.temporal.io/api v1.63.5 go.temporal.io/server v0.0.0-00010101000000-000000000000 - google.golang.org/grpc v1.80.0 + google.golang.org/grpc v1.83.1 google.golang.org/protobuf v1.36.11 ) @@ -26,8 +26,8 @@ require ( golang.org/x/net v0.57.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/text v0.40.0 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260420184626-e10c466a9529 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect gopkg.in/yaml.v3 v3.0.1 // indirect ) diff --git a/tests/mixedbrain/go.sum b/tests/mixedbrain/go.sum index f235f9afc83..e04f1505577 100644 --- a/tests/mixedbrain/go.sum +++ b/tests/mixedbrain/go.sum @@ -49,10 +49,10 @@ go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= -go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= -go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= -go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= -go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= go.temporal.io/api v1.63.5 h1:c11+kPYHkXXL3UiShPdbMD+xtvqGsbTibUA9ypmiCa4= @@ -71,12 +71,12 @@ golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs= golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY= gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= -google.golang.org/genproto/googleapis/api v0.0.0-20260420184626-e10c466a9529 h1:zUWMZsvo/IJcD1t6MNCPO/azZTwz0TvwCBqr5aifoVY= -google.golang.org/genproto/googleapis/api v0.0.0-20260420184626-e10c466a9529/go.mod h1:a5OGAgyRr4lqco7AG9hQM9Fwh0N2ZV4grR0eXFEsXQg= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529 h1:XF8+t6QQiS0o9ArVan/HW8Q7cycNPGsJf6GA2nXxYAg= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260420184626-e10c466a9529/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= -google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM= -google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.83.1 h1:HIO0+BEtBP6soyqvqC8sNUjZ7bTs+0hFQuFF+RAy++Y= +google.golang.org/grpc v1.83.1/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ= google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= diff --git a/tests/nexus_workflow_test.go b/tests/nexus_workflow_test.go index 72a016951b9..78eff713361 100644 --- a/tests/nexus_workflow_test.go +++ b/tests/nexus_workflow_test.go @@ -54,6 +54,12 @@ import ( "google.golang.org/protobuf/types/known/durationpb" ) +const ( + dataBlobMessageType = "temporal.api.common.v1.DataBlob" + payloadsMessageType = "temporal.api.common.v1.Payloads" + protobufEncoding = "binary/protobuf" +) + type NexusWorkflowTestSuite struct { parallelsuite.Suite[*NexusWorkflowTestSuite] } @@ -3283,6 +3289,9 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationSystemEndpoint(chasmEnabled b result := completedEvent.GetNexusOperationCompletedEventAttributes().Result s.NotNil(result) s.Equal([]byte("true"), result.GetMetadata()[commonnexus.SystemPayloadMetadataKey]) + // TestOperation returns a proto message, so the result must be proto encoded, not JSON. + s.Equal([]byte(dataBlobMessageType), result.GetMetadata()["messageType"]) + s.Equal([]byte(protobufEncoding), result.GetMetadata()["encoding"]) // Complete the workflow _, err = env.FrontendClient().RespondWorkflowTaskCompleted(ctx, &workflowservice.RespondWorkflowTaskCompletedRequest{ @@ -3302,9 +3311,10 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationSystemEndpoint(chasmEnabled b }, }) s.NoError(err) - var response string + var response commonpb.DataBlob s.NoError(run.Get(ctx, &response)) - s.Equal("Hello, Temporal", response) + data := response.Data + s.Equal("Hello, Temporal", string(data)) } // NOTE: This test cannot use the SDK workflow package because there is a restriction that prevents setting the @@ -3361,9 +3371,10 @@ func (s *NexusWorkflowTestSuite) TestNexusOperationSystemEndpoint_PayloadMetadat completedEvent := s.RequireHistoryEvent(pollResp.History.Events, enumspb.EVENT_TYPE_NEXUS_OPERATION_COMPLETED) result := completedEvent.GetNexusOperationCompletedEventAttributes().Result s.NotNil(result) - // TestOperationWithPayload's response embeds a nested Payload, so the system payload metadata - // flag must be set. + // TestOperationWithPayload's response embeds a nested Payload, so the system payload metadata flag must be set. s.Equal([]byte("true"), result.GetMetadata()[commonnexus.SystemPayloadMetadataKey]) + s.Equal([]byte(payloadsMessageType), result.GetMetadata()["messageType"]) + s.Equal([]byte(protobufEncoding), result.GetMetadata()["encoding"]) // Complete the workflow _, err = env.FrontendClient().RespondWorkflowTaskCompleted(s.Context(), &workflowservice.RespondWorkflowTaskCompletedRequest{ From 80895a0e5fa00c6a269c7937d1ded1389cf937e5 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 25 Sep 2026 08:04:30 -0700 Subject: [PATCH 2/5] Required protobuf payloads from the system Nexus endpoint. Taken from upstream temporalio/temporal #11906. The api version this branch pins rejects a system payload that is not binary/protobuf, so the endpoint has to refuse one instead of labelling it. --- service/history/handler.go | 27 --------------------------- 1 file changed, 27 deletions(-) diff --git a/service/history/handler.go b/service/history/handler.go index f63609d89f9..c4aa4ea9f3b 100644 --- a/service/history/handler.go +++ b/service/history/handler.go @@ -880,33 +880,6 @@ func (h *Handler) GetMutableState(ctx context.Context, request *historyservice.G return resp, nil } -// GetStreamReplaySlices re-supplies the stream ranges a workflow's completed tasks recorded. -func (h *Handler) GetStreamReplaySlices( - ctx context.Context, - request *historyservice.GetStreamReplaySlicesRequest, -) (*historyservice.GetStreamReplaySlicesResponse, error) { - namespaceID := namespace.ID(request.GetNamespaceId()) - if namespaceID == "" { - return nil, h.convertError(errNamespaceNotSet) - } - - workflowID := request.GetExecution().GetWorkflowId() - shardContext, err := h.controller.GetShardByNamespaceWorkflow(namespaceID, workflowID) - if err != nil { - return nil, h.convertError(err) - } - engine, err := shardContext.GetEngine(ctx) - if err != nil { - return nil, h.convertError(err) - } - - resp, err := engine.GetStreamReplaySlices(ctx, request) - if err != nil { - return nil, h.convertError(err) - } - return resp, nil -} - // PollMutableState - returns the id of the next event in the execution's history func (h *Handler) PollMutableState(ctx context.Context, request *historyservice.PollMutableStateRequest) (*historyservice.PollMutableStateResponse, error) { namespaceID := namespace.ID(request.GetNamespaceId()) From 27ac37f0ca264802d36ba71d041c2558759e0a76 Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 25 Sep 2026 14:21:09 -0700 Subject: [PATCH 3/5] Scoped the exemplar comparison to the exemplar field. Equating empty across the whole comparison also hid a nil-versus-empty attribute set or data point, which is not what the test is about. --- common/metrics/otel_metrics_handler_test.go | 19 ++++++++++--------- 1 file changed, 10 insertions(+), 9 deletions(-) diff --git a/common/metrics/otel_metrics_handler_test.go b/common/metrics/otel_metrics_handler_test.go index 9cfc5aa0acb..e229392d984 100644 --- a/common/metrics/otel_metrics_handler_test.go +++ b/common/metrics/otel_metrics_handler_test.go @@ -192,13 +192,14 @@ func TestMeter(t *testing.T) { cmpopts.SortSlices(func(x, y metricdata.Metrics) bool { return x.Name < y.Name }), - cmpopts.IgnoreFields(metricdata.DataPoint[int64]{}, "StartTime", "Time"), - cmpopts.IgnoreFields(metricdata.DataPoint[float64]{}, "StartTime", "Time"), - cmpopts.IgnoreFields(metricdata.HistogramDataPoint[int64]{}, "StartTime", "Time", "Bounds"), - // The SDK hands back either a nil or an allocated-but-empty exemplar - // slice depending on whether its reservoir was ever touched, so the - // two have to compare equal here. - cmpopts.EquateEmpty(), + // Exemplars are ignored rather than equated empty: the SDK hands back + // either a nil or an allocated-but-empty slice depending on whether + // its reservoir was ever touched, and equating empty globally would + // also hide a nil-versus-empty attribute set or data point. + cmpopts.IgnoreFields(metricdata.DataPoint[int64]{}, "StartTime", "Time", "Exemplars"), + cmpopts.IgnoreFields(metricdata.DataPoint[float64]{}, "StartTime", "Time", "Exemplars"), + cmpopts.IgnoreFields( + metricdata.HistogramDataPoint[int64]{}, "StartTime", "Time", "Bounds", "Exemplars"), ); diff != "" { t.Errorf("mismatch (-want, +got):\n%s", diff) } @@ -266,8 +267,8 @@ func TestMeter_TimerInSeconds(t *testing.T) { cmp.Comparer(func(a1, a2 attribute.Set) bool { return a1.Equals(&a2) }), - cmpopts.IgnoreFields(metricdata.HistogramDataPoint[float64]{}, "StartTime", "Time", "Bounds"), - cmpopts.EquateEmpty(), + cmpopts.IgnoreFields( + metricdata.HistogramDataPoint[float64]{}, "StartTime", "Time", "Bounds", "Exemplars"), ); diff != "" { t.Errorf("mismatch (-want, +got):\n%s", diff) } From 183efab2321e4502b77e707640d8110cac2aa4ea Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 25 Sep 2026 16:14:18 -0700 Subject: [PATCH 4/5] Repinned the api to the renamed stream fields. The api branch renamed the two stream command fields and reshaped the appended event's range. The layers above this one need the new accessors. --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index 2be924061ca..5ccd5a9d4be 100644 --- a/go.mod +++ b/go.mod @@ -240,4 +240,4 @@ require ( tool golang.org/x/perf/cmd/benchstat -replace go.temporal.io/api => github.com/moedash/api-go v1.63.6-0.20260921082525-9a6aa5f95be8 +replace go.temporal.io/api => github.com/moedash/api-go v1.63.6-0.20260925231147-63d1722ce567 diff --git a/go.sum b/go.sum index 13b45018247..72b85cadd97 100644 --- a/go.sum +++ b/go.sum @@ -321,8 +321,8 @@ github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJ github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8= github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= -github.com/moedash/api-go v1.63.6-0.20260921082525-9a6aa5f95be8 h1:WdG6n/8Da2RC5y3t3/vdpq0P4wOBbOQX+C+C9TiChew= -github.com/moedash/api-go v1.63.6-0.20260921082525-9a6aa5f95be8/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= +github.com/moedash/api-go v1.63.6-0.20260925231147-63d1722ce567 h1:lKAIMDl+aIBL3Y4GxDXab40rR6eVhGnPDnFbzzSCD2Y= +github.com/moedash/api-go v1.63.6-0.20260925231147-63d1722ce567/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= github.com/ncruces/go-strftime v1.0.0 h1:HMFp8mLCTPp341M/ZnA4qaf7ZlsbTc+miZjCLOFAw7w= From aac97837cfb87797ca41481a4f119dd108083b6e Mon Sep 17 00:00:00 2001 From: Moe Dashti Date: Fri, 25 Sep 2026 20:00:15 -0700 Subject: [PATCH 5/5] Read the pinned append command through its renamed field. The re-pinned api-go names the appended stream by StreamName, so the pin test stops compiling without it. --- tests/api_fork_test.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/api_fork_test.go b/tests/api_fork_test.go index 8e39bb7109b..c4e092ed005 100644 --- a/tests/api_fork_test.go +++ b/tests/api_fork_test.go @@ -22,8 +22,8 @@ func TestApiForkCarriesStreamShapes(t *testing.T) { CommandType: enumspb.COMMAND_TYPE_APPEND_STREAM_RECORDS, Attributes: &commandpb.Command_AppendStreamRecordsCommandAttributes{ AppendStreamRecordsCommandAttributes: &commandpb.AppendStreamRecordsCommandAttributes{ - StreamId: "s1", - Records: []*streampb.StreamRecord{{Topic: "tokens"}}, + StreamName: "s1", + Records: []*streampb.StreamRecord{{Topic: "tokens"}}, }, }, } @@ -31,7 +31,7 @@ func TestApiForkCarriesStreamShapes(t *testing.T) { require.NoError(t, err) var back commandpb.Command require.NoError(t, proto.Unmarshal(b, &back)) - require.Equal(t, "s1", back.GetAppendStreamRecordsCommandAttributes().GetStreamId()) + require.Equal(t, "s1", back.GetAppendStreamRecordsCommandAttributes().GetStreamName()) require.Equal(t, "tokens", back.GetAppendStreamRecordsCommandAttributes().GetRecords()[0].GetTopic())