Test Go and Java SDK client retry behavior

Author: Quinn-With-Two-Ns

cmd/run.go

@@ -10,6 +10,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"os/exec"
 	"os/signal"
 	"path/filepath"
 	"runtime"
@@ -30,8 +31,9 @@ import (
 )
 
 const (
-	summaryListenAddr = "127.0.0.1:0"
-	FeaturePassed     = "PASSED"
+	proxyExecutableAuto = "auto"
+	freePortListenAddr  = "127.0.0.1:0"
+	FeaturePassed       = "PASSED"
 )
 
 func runCmd() *cli.Command {
@@ -66,6 +68,40 @@ type RunConfig struct {
 	RetainTempDir       bool
 	SummaryURI          string
 	HTTPProxyURL        string
+	ProxyControlURI     string
+	ProxyListenHostPort string
+}
+
+func (config RunConfig) appendFlags(out []string) ([]string, error) {
+	out = append(out, "--server="+config.Server)
+	out = append(out, "--namespace="+config.Namespace)
+	if config.ClientCertPath != "" {
+		clientCertPath, err := filepath.Abs(config.ClientCertPath)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, "--client-cert-path="+clientCertPath)
+	}
+	if config.ClientKeyPath != "" {
+		clientKeyPath, err := filepath.Abs(config.ClientKeyPath)
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, "--client-key-path="+clientKeyPath)
+	}
+	if config.SummaryURI != "" {
+		out = append(out, "--summary-uri="+config.SummaryURI)
+	}
+	if config.HTTPProxyURL != "" {
+		out = append(out, "--http-proxy-url", config.HTTPProxyURL)
+	}
+	if config.ProxyControlURI != "" {
+		out = append(out, "--proxy-control-uri="+config.ProxyControlURI)
+	}
+	if config.ProxyListenHostPort != "" {
+		out = append(out, "--proxy-listen-host-port="+config.ProxyListenHostPort)
+	}
+	return out, nil
 }
 
 // dockerRunFlags are a subset of flags that apply when running in a docker container
@@ -134,6 +170,7 @@ type Runner struct {
 	rootDir    string
 	createTime time.Time
 	program    sdkbuild.Program
+	proxy      *exec.Cmd
 }
 
 // NewRunner creates a new runner for the given config.
@@ -188,13 +225,17 @@ func (r *Runner) Run(ctx context.Context, patterns []string) error {
 	}
 	// Aa task queue to every feature
 	run := &cmd.Run{Features: make([]cmd.RunFeature, len(features))}
-	var expectsProxy bool
+	var expectsHTTPProxy bool
+	var expectsGrpcProxy bool
 	for i, feature := range features {
 		run.Features[i].Dir = feature.Dir
 		run.Features[i].TaskQueue = fmt.Sprintf("features-%v-%v", feature.Dir, uuid.NewString())
 		run.Features[i].Config = feature.Config
 		if feature.Config.ExpectUnauthedProxyCount > 0 || feature.Config.ExpectAuthedProxyCount > 0 {
-			expectsProxy = true
+			expectsHTTPProxy = true
+		}
+		if feature.Config.GrpcProxy {
+			expectsGrpcProxy = true
 		}
 	}
 
@@ -228,15 +269,32 @@ func (r *Runner) Run(ctx context.Context, patterns []string) error {
 	}
 
 	// If any feature requires an HTTP proxy, we must run it
-	var proxyServer *harness.HTTPConnectProxyServer
-	if expectsProxy {
-		proxyServer, err = harness.StartHTTPConnectProxyServer(harness.HTTPConnectProxyServerOptions{Log: r.log})
+	var httpProxyServer *harness.HTTPConnectProxyServer
+	if expectsHTTPProxy {
+		httpProxyServer, err = harness.StartHTTPConnectProxyServer(harness.HTTPConnectProxyServerOptions{Log: r.log})
+		if err != nil {
+			return fmt.Errorf("could not start http proxy server: %w", err)
+		}
+		r.config.HTTPProxyURL = "https://" + httpProxyServer.Address
+		r.log.Info("Started HTTP CONNECT proxy server", "address", httpProxyServer.Address)
+		defer httpProxyServer.Close()
+	}
+
+	// if any feature requires a gRPC proxy, we must run it
+	if expectsGrpcProxy {
+		grpcProxyServer, err := harness.StartGRPCProxyServer(harness.GRPCProxyServerOptions{
+			DialAddress:    r.config.Server,
+			ClientCertPath: r.config.ClientCertPath,
+			ClientKeyPath:  r.config.ClientKeyPath,
+			Log:            r.log,
+		})
 		if err != nil {
-			return fmt.Errorf("could not start proxy server: %w", err)
+			return fmt.Errorf("could not start grpc proxy server: %w", err)
 		}
-		r.config.HTTPProxyURL = "https://" + proxyServer.Address
-		r.log.Info("Started HTTP CONNECT proxy server", "address", proxyServer.Address)
-		defer proxyServer.Close()
+		r.config.ProxyListenHostPort = grpcProxyServer.ProxyAddress()
+		r.config.ProxyControlURI = "http://" + grpcProxyServer.ControlAddress()
+		r.log.Info("Started gRPC proxy server", "address", grpcProxyServer.ProxyAddress(), "control", grpcProxyServer.ControlAddress())
+		defer grpcProxyServer.Close()
 	}
 
 	// Ensure any created temp dir is cleaned on ctrl-c or normal exit
@@ -251,14 +309,14 @@ func (r *Runner) Run(ctx context.Context, patterns []string) error {
 		defer r.destroyTempDir()
 	}
 
-	l, err := net.Listen("tcp", summaryListenAddr)
+	summaryListener, err := net.Listen("tcp", freePortListenAddr)
 	if err != nil {
 		return err
 	}
-	defer l.Close()
+	defer summaryListener.Close()
 	summaryChan := make(chan Summary)
-	go r.summaryServer(l, summaryChan)
-	r.config.SummaryURI = "tcp://" + l.Addr().String()
+	go r.summaryServer(summaryListener, summaryChan)
+	r.config.SummaryURI = "tcp://" + summaryListener.Addr().String()
 
 	err = nil
 	switch r.config.Lang {
@@ -273,12 +331,14 @@ func (r *Runner) Run(ctx context.Context, patterns []string) error {
 			}
 		} else {
 			err = cmd.NewRunner(cmd.RunConfig{
-				Server:         r.config.Server,
-				Namespace:      r.config.Namespace,
-				ClientCertPath: r.config.ClientCertPath,
-				ClientKeyPath:  r.config.ClientKeyPath,
-				SummaryURI:     r.config.SummaryURI,
-				HTTPProxyURL:   r.config.HTTPProxyURL,
+				Server:              r.config.Server,
+				Namespace:           r.config.Namespace,
+				ClientCertPath:      r.config.ClientCertPath,
+				ClientKeyPath:       r.config.ClientKeyPath,
+				SummaryURI:          r.config.SummaryURI,
+				HTTPProxyURL:        r.config.HTTPProxyURL,
+				ProxyControlURI:     r.config.ProxyControlURI,
+				ProxyListenHostPort: r.config.ProxyListenHostPort,
 			}).Run(ctx, run)
 		}
 	case "java":
@@ -315,7 +375,7 @@ func (r *Runner) Run(ctx context.Context, patterns []string) error {
 	if err != nil {
 		return err
 	}
-	l.Close()
+	summaryListener.Close()
 	summary, ok := <-summaryChan
 	if !ok {
 		r.log.Debug("did not receive a test run summary - adopting legacy behavior of assuming no tests were skipped")
@@ -327,7 +387,7 @@ func (r *Runner) Run(ctx context.Context, patterns []string) error {
 	// For features that expected proxy connections, count how many expected
 	// ignoring skips and compare count with actual. If any failed we don't need
 	// even do the comparison.
-	if proxyServer != nil {
+	if httpProxyServer != nil {
 		var anyFailed bool
 		var expectUnauthedProxyCount, expectAuthedProxyCount int
 		for _, summ := range summary {
@@ -345,21 +405,26 @@ func (r *Runner) Run(ctx context.Context, patterns []string) error {
 			}
 		}
 		if !anyFailed {
-			if proxyServer.UnauthedConnectionsTunneled.Load() != uint32(expectUnauthedProxyCount) {
+			if httpProxyServer.UnauthedConnectionsTunneled.Load() != uint32(expectUnauthedProxyCount) {
 				return fmt.Errorf("expected %v unauthed HTTP proxy connections, got %v",
-					expectUnauthedProxyCount, proxyServer.UnauthedConnectionsTunneled.Load())
-			} else if proxyServer.AuthedConnectionsTunneled.Load() != uint32(expectAuthedProxyCount) {
+					expectUnauthedProxyCount, httpProxyServer.UnauthedConnectionsTunneled.Load())
+			} else if httpProxyServer.AuthedConnectionsTunneled.Load() != uint32(expectAuthedProxyCount) {
 				return fmt.Errorf("expected %v authed HTTP proxy connections, got %v",
-					expectAuthedProxyCount, proxyServer.AuthedConnectionsTunneled.Load())
+					expectAuthedProxyCount, httpProxyServer.AuthedConnectionsTunneled.Load())
 			} else {
 				r.log.Debug("Matched expected HTTP proxy connections",
-					"expectUnauthed", expectUnauthedProxyCount, "actualUnauthed", proxyServer.UnauthedConnectionsTunneled.Load(),
-					"expectAuthed", expectAuthedProxyCount, "actualAuthed", proxyServer.AuthedConnectionsTunneled.Load())
+					"expectUnauthed", expectUnauthedProxyCount, "actualUnauthed", httpProxyServer.UnauthedConnectionsTunneled.Load(),
+					"expectAuthed", expectAuthedProxyCount, "actualAuthed", httpProxyServer.AuthedConnectionsTunneled.Load())
 			}
 		}
 	}
 
-	return r.handleHistory(ctx, run, summary)
+	err = r.handleHistory(ctx, run, summary)
+	if err != nil {
+		return err
+	}
+
+	return nil
 }
 
 func (r *Runner) handleHistory(ctx context.Context, run *cmd.Run, summary Summary) error {

cmd/run_dotnet.go

@@ -64,11 +64,15 @@ func (r *Runner) RunDotNetExternal(ctx context.Context, run *cmd.Run) error {
 		}
 	}
 
-	args := []string{"--server", r.config.Server, "--namespace", r.config.Namespace}
-	if r.config.ClientCertPath != "" {
-		args = append(args, "--client-cert-path", r.config.ClientCertPath, "--client-key-path", r.config.ClientKeyPath)
+	// Build args
+	args := make([]string, 0, 64)
+	args, err := r.config.appendFlags(args)
+	if err != nil {
+		return err
 	}
 	args = append(args, run.ToArgs()...)
+
+	// Run
 	cmd, err := r.program.NewCommand(ctx, args...)
 	if err == nil {
 		r.log.Debug("Running Go separately", "Args", cmd.Args)

cmd/run_go.go

@@ -68,6 +68,12 @@ func (r *Runner) RunGoExternal(ctx context.Context, run *cmd.Run) error {
 	if r.config.HTTPProxyURL != "" {
 		args = append(args, "--http-proxy-url", r.config.HTTPProxyURL)
 	}
+	if r.config.ProxyControlURI != "" {
+		args = append(args, "--proxy-control-uri", r.config.ProxyControlURI)
+	}
+	if r.config.ProxyListenHostPort != "" {
+		args = append(args, "--proxy-listen-host-port", r.config.ProxyListenHostPort)
+	}
 	args = append(args, run.ToArgs()...)
 	cmd, err := r.program.NewCommand(ctx, args...)
 	if err == nil {

cmd/run_java.go

@@ -3,7 +3,6 @@ package cmd
 import (
 	"context"
 	"fmt"
-	"path/filepath"
 
 	"github.com/temporalio/features/harness/go/cmd"
 	"github.com/temporalio/features/sdkbuild"
@@ -40,26 +39,10 @@ func (r *Runner) RunJavaExternal(ctx context.Context, run *cmd.Run) error {
 	}
 
 	// Build args
-	args := []string{"--server", r.config.Server, "--namespace", r.config.Namespace}
-	if r.config.ClientCertPath != "" {
-		clientCertPath, err := filepath.Abs(r.config.ClientCertPath)
-		if err != nil {
-			return err
-		}
-		args = append(args, "--client-cert-path", clientCertPath)
-	}
-	if r.config.ClientKeyPath != "" {
-		clientKeyPath, err := filepath.Abs(r.config.ClientKeyPath)
-		if err != nil {
-			return err
-		}
-		args = append(args, "--client-key-path", clientKeyPath)
-	}
-	if r.config.SummaryURI != "" {
-		args = append(args, "--summary-uri", r.config.SummaryURI)
-	}
-	if r.config.HTTPProxyURL != "" {
-		args = append(args, "--http-proxy-url", r.config.HTTPProxyURL)
+	args := make([]string, 0, 64)
+	args, err := r.config.appendFlags(args)
+	if err != nil {
+		return err
 	}
 	args = append(args, run.ToArgs()...)
 

cmd/run_python.go

@@ -60,23 +60,11 @@ func (r *Runner) RunPythonExternal(ctx context.Context, run *cmd.Run) error {
 	}
 
 	// Build args
-	args := []string{"harness.python.main", "--server", r.config.Server, "--namespace", r.config.Namespace}
-	if r.config.ClientCertPath != "" {
-		clientCertPath, err := filepath.Abs(r.config.ClientCertPath)
-		if err != nil {
-			return err
-		}
-		args = append(args, "--client-cert-path", clientCertPath)
-	}
-	if r.config.ClientKeyPath != "" {
-		clientKeyPath, err := filepath.Abs(r.config.ClientKeyPath)
-		if err != nil {
-			return err
-		}
-		args = append(args, "--client-key-path", clientKeyPath)
-	}
-	if r.config.HTTPProxyURL != "" {
-		args = append(args, "--http-proxy-url", r.config.HTTPProxyURL)
+	args := make([]string, 0, 64)
+	args = append(args, "harness.python.main")
+	args, err := r.config.appendFlags(args)
+	if err != nil {
+		return err
 	}
 	args = append(args, run.ToArgs()...)
 

cmd/run_typescript.go

@@ -58,26 +58,11 @@ func (r *Runner) RunTypeScriptExternal(ctx context.Context, run *cmd.Run) error
 	}
 
 	// Build args
-	args := []string{
-		"./tslib/harness/ts/main.js",
-		"--server",
-		r.config.Server,
-		"--namespace",
-		r.config.Namespace,
-	}
-	if r.config.ClientCertPath != "" {
-		clientCertPath, err := filepath.Abs(r.config.ClientCertPath)
-		if err != nil {
-			return err
-		}
-		args = append(args, "--client-cert-path", clientCertPath)
-	}
-	if r.config.ClientKeyPath != "" {
-		clientKeyPath, err := filepath.Abs(r.config.ClientKeyPath)
-		if err != nil {
-			return err
-		}
-		args = append(args, "--client-key-path", clientKeyPath)
+	args := make([]string, 0, 64)
+	args = append(args, "./tslib/harness/ts/main.js")
+	args, err := r.config.appendFlags(args)
+	if err != nil {
+		return err
 	}
 	args = append(args, run.ToArgs()...)
 

features/features.go

@@ -26,6 +26,9 @@ import (
 	data_converter_json_protobuf "github.com/temporalio/features/features/data_converter/json_protobuf"
 	eager_activity_non_remote_activities_worker "github.com/temporalio/features/features/eager_activity/non_remote_activities_worker"
 	eager_workflow_successful_start "github.com/temporalio/features/features/eager_workflow/successful_start"
+	grpc_retry_server_frozen_for_initiator "github.com/temporalio/features/features/grpc_retry/server_frozen_for_initiator"
+	grpc_retry_server_restarted_for_initiator "github.com/temporalio/features/features/grpc_retry/server_restarted_for_initiator"
+	grpc_retry_server_unavailable_for_initiator "github.com/temporalio/features/features/grpc_retry/server_unavailable_for_initiator"
 	query_successful_query "github.com/temporalio/features/features/query/successful_query"
 	query_timeout_due_to_no_active_workers "github.com/temporalio/features/features/query/timeout_due_to_no_active_workers"
 	query_unexpected_arguments "github.com/temporalio/features/features/query/unexpected_arguments"
@@ -71,15 +74,18 @@ func init() {
 		client_http_proxy.Feature,
 		client_http_proxy_auth.Feature,
 		continue_as_new_continue_as_same.Feature,
-		data_converter_binary_protobuf.Feature,
 		data_converter_binary.Feature,
+		data_converter_binary_protobuf.Feature,
 		data_converter_codec.Feature,
 		data_converter_empty.Feature,
 		data_converter_failure.Feature,
-		data_converter_json_protobuf.Feature,
 		data_converter_json.Feature,
+		data_converter_json_protobuf.Feature,
 		eager_activity_non_remote_activities_worker.Feature,
 		eager_workflow_successful_start.Feature,
+		grpc_retry_server_frozen_for_initiator.Feature,
+		grpc_retry_server_restarted_for_initiator.Feature,
+		grpc_retry_server_unavailable_for_initiator.Feature,
 		query_successful_query.Feature,
 		query_timeout_due_to_no_active_workers.Feature,
 		query_unexpected_arguments.Feature,
@@ -96,8 +102,8 @@ func init() {
 		update_activities.Feature,
 		update_async_accepted.Feature,
 		update_basic.Feature,
-		update_deduplication.Feature,
 		update_client_interceptor.Feature,
+		update_deduplication.Feature,
 		update_non_durable_reject.Feature,
 		update_self.Feature,
 		update_task_failure.Feature,

features/grpc_retry/server_frozen_for_initiator/.config.json

@@ -0,0 +1,3 @@
+{
+  "grpcProxy": true
+}