Enable go sdk to use the new restate-sdk-test-tool (#19)

* Enable go sdk to use the new restate-sdk-test-tool

.github/workflows/test.yaml

@@ -1,6 +1,10 @@
 name: Go
 on: [push]
 
+permissions:
+  checks: write
+  pull-requests: write
+
 jobs:
   build:
     runs-on: ubuntu-latest
@@ -19,3 +23,49 @@ jobs:
         run: go build -v ./...
       - name: Test with the Go CLI
         run: go test -v ./...
+
+  sdk-test-suite:
+    runs-on: ubuntu-latest
+    name: "Integration Test (Test tool ${{ matrix.sdk-test-suite }})"
+    strategy:
+      matrix:
+        sdk-test-suite: [ "1.4" ]
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Setup Go
+        uses: actions/setup-go@v4
+        with:
+          go-version: "1.21.x"
+      - name: Setup Java
+        uses: actions/setup-java@v4
+        with:
+          distribution: 'temurin'
+          java-version: '17'
+      - name: Setup ko
+        uses: ko-build/[email protected]
+      - name: Setup sdk-test-suite
+        run: wget --no-verbose https://github.com/restatedev/sdk-test-suite/releases/download/v${{ matrix.sdk-test-suite }}/restate-sdk-test-suite.jar
+
+      # Build docker image
+      - name: Install dependencies
+        run: go get .
+      - name: Build Docker image
+        run: KO_DOCKER_REPO=restatedev ko build -B -L github.com/restatedev/sdk-go/test-services
+
+      # Run test suite
+      - name: Run test suite
+        run: java -jar restate-sdk-test-suite.jar run --report-dir=test-report --exclusions-file test-services/exclusions.yaml restatedev/test-services
+
+      # Upload logs and publish test result
+      - uses: actions/upload-artifact@v4
+        if: always() # Make sure this is run even when test fails
+        with:
+          name: test-report
+          path: test-report
+      - name: Publish Test Results
+        uses: EnricoMi/publish-unit-test-result-action@v2
+        if: always()
+        with:
+          files: |
+            test-report/*/*.xml

test-services/README.md

@@ -0,0 +1,25 @@
+# Test services to run the sdk-test-suite
+
+## To run locally
+
+* Grab the release of sdk-test-suite: https://github.com/restatedev/sdk-test-suite/releases
+
+* Prepare the docker image: 
+```shell
+KO_DOCKER_REPO=restatedev ko build -B -L github.com/restatedev/sdk-go/test-services
+```
+
+* Run the tests (requires JVM >= 17):
+```shell
+java -jar restate-sdk-test-suite.jar run --exclusions-file exclusions.yaml restatedev/test-services
+```
+
+## To debug a single test:
+
+* Run the golang service using your IDE
+* Run the test runner in debug mode specifying test suite and test:
+```shell
+java -jar restate-sdk-test-suite.jar debug --image-pull-policy=CACHED --test-config=lazyState --test-name=dev.restate.sdktesting.tests.State default-service=9080
+```
+
+For more info: https://github.com/restatedev/sdk-test-suite

test-services/counter.go

@@ -0,0 +1,68 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	restate "github.com/restatedev/sdk-go"
+)
+
+const COUNTER_KEY = "counter"
+
+type CounterUpdateResponse struct {
+	OldValue int64 `json:"oldValue"`
+	NewValue int64 `json:"newValue"`
+}
+
+func RegisterCounter() {
+	REGISTRY.AddRouter(
+		restate.NewObjectRouter("Counter").
+			Handler("reset", restate.NewObjectHandler(
+				func(ctx restate.ObjectContext, _ restate.Void) (restate.Void, error) {
+					ctx.Clear(COUNTER_KEY)
+					return restate.Void{}, nil
+				})).
+			Handler("get", restate.NewObjectSharedHandler(
+				func(ctx restate.ObjectSharedContext, _ restate.Void) (int64, error) {
+					c, err := restate.GetAs[int64](ctx, COUNTER_KEY)
+					if errors.Is(err, restate.ErrKeyNotFound) {
+						c = 0
+					} else if err != nil {
+						return 0, err
+					}
+					return c, nil
+				})).
+			Handler("add", restate.NewObjectHandler(
+				func(ctx restate.ObjectContext, addend int64) (CounterUpdateResponse, error) {
+					oldValue, err := restate.GetAs[int64](ctx, COUNTER_KEY)
+					if errors.Is(err, restate.ErrKeyNotFound) {
+						oldValue = 0
+					} else if err != nil {
+						return CounterUpdateResponse{}, err
+					}
+
+					newValue := oldValue + addend
+					err = ctx.Set(COUNTER_KEY, newValue)
+
+					return CounterUpdateResponse{
+						OldValue: oldValue,
+						NewValue: newValue,
+					}, err
+				})).
+			Handler("addThenFail", restate.NewObjectHandler(
+				func(ctx restate.ObjectContext, addend int64) (restate.Void, error) {
+					oldValue, err := restate.GetAs[int64](ctx, COUNTER_KEY)
+					if errors.Is(err, restate.ErrKeyNotFound) {
+						oldValue = 0
+					} else if err != nil {
+						return restate.Void{}, err
+					}
+
+					newValue := oldValue + addend
+					err = ctx.Set(COUNTER_KEY, newValue)
+					if err != nil {
+						return restate.Void{}, err
+					}
+
+					return restate.Void{}, restate.TerminalError(fmt.Errorf("%s", ctx.Key()))
+				})))
+}

test-services/exclusions.yaml

@@ -0,0 +1,39 @@
+exclusions:
+  "default":
+    - "dev.restate.sdktesting.tests.KillInvocation"
+    - "dev.restate.sdktesting.tests.CallOrdering"
+    - "dev.restate.sdktesting.tests.UserErrors"
+    - "dev.restate.sdktesting.tests.SleepWithFailures"
+    - "dev.restate.sdktesting.tests.Ingress"
+    - "dev.restate.sdktesting.tests.WorkflowAPI"
+    - "dev.restate.sdktesting.tests.State"
+    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication"
+    - "dev.restate.sdktesting.tests.CancelInvocation"
+    - "dev.restate.sdktesting.tests.Sleep"
+    - "dev.restate.sdktesting.tests.AwaitTimeout"
+  "alwaysSuspending":
+    - "dev.restate.sdktesting.tests.WorkflowAPI"
+    - "dev.restate.sdktesting.tests.SideEffect"
+    - "dev.restate.sdktesting.tests.State"
+    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication"
+    - "dev.restate.sdktesting.tests.UserErrors"
+    - "dev.restate.sdktesting.tests.Sleep"
+    - "dev.restate.sdktesting.tests.AwaitTimeout"
+    - "dev.restate.sdktesting.tests.SleepWithFailures"
+    - "dev.restate.sdktesting.tests.NonDeterminismErrors"
+  "singleThreadSinglePartition":
+    - "dev.restate.sdktesting.tests.KillInvocation"
+    - "dev.restate.sdktesting.tests.CallOrdering"
+    - "dev.restate.sdktesting.tests.UserErrors"
+    - "dev.restate.sdktesting.tests.SleepWithFailures"
+    - "dev.restate.sdktesting.tests.Ingress"
+    - "dev.restate.sdktesting.tests.WorkflowAPI"
+    - "dev.restate.sdktesting.tests.State"
+    - "dev.restate.sdktesting.tests.ServiceToServiceCommunication"
+    - "dev.restate.sdktesting.tests.CancelInvocation"
+    - "dev.restate.sdktesting.tests.Sleep"
+    - "dev.restate.sdktesting.tests.AwaitTimeout"
+  "lazyState":
+    - "dev.restate.sdktesting.tests.State"
+  "persistedTimers":
+    - "dev.restate.sdktesting.tests.Sleep"

test-services/main.go

@@ -0,0 +1,45 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+	"os"
+	"strings"
+
+	"github.com/restatedev/sdk-go/server"
+)
+
+func init() {
+	RegisterProxy()
+	RegisterCounter()
+}
+
+func main() {
+	services := "*"
+	if os.Getenv("SERVICES") != "" {
+		services = os.Getenv("SERVICES")
+	}
+
+	server := server.NewRestate()
+
+	if services == "*" {
+		REGISTRY.RegisterAll(server)
+	} else {
+		fqdns := strings.Split(services, ",")
+		set := make(map[string]struct{}, len(fqdns))
+		for _, fqdn := range fqdns {
+			set[fqdn] = struct{}{}
+		}
+		REGISTRY.Register(set, server)
+	}
+
+	port := os.Getenv("PORT")
+	if port == "" {
+		port = "9080"
+	}
+
+	if err := server.Start(context.Background(), ":"+port); err != nil {
+		slog.Error("application exited unexpectedly", "err", err)
+		os.Exit(1)
+	}
+}

test-services/proxy.go

@@ -0,0 +1,154 @@
+package main
+
+import (
+	restate "github.com/restatedev/sdk-go"
+)
+
+type ProxyRequest struct {
+	ServiceName      string  `json:"serviceName"`
+	VirtualObjectKey *string `json:"virtualObjectKey,omitempty"`
+	HandlerName      string  `json:"handlerName"`
+	// We need to use []int because Golang takes the opinionated choice of treating []byte as Base64
+	Message []int `json:"message"`
+}
+
+type ManyCallRequest struct {
+	ProxyRequest  ProxyRequest `json:"proxyRequest"`
+	OneWayCall    bool         `json:"oneWayCall"`
+	AwaitAtTheEnd bool         `json:"awaitAtTheEnd"`
+}
+
+func RegisterProxy() {
+	REGISTRY.AddRouter(
+		restate.NewServiceRouter("Proxy").
+			Handler("call", restate.NewServiceHandler(
+				// We need to use []int because Golang takes the opinionated choice of treating []byte as Base64
+				func(ctx restate.Context, req ProxyRequest) ([]int, error) {
+					input := intArrayToByteArray(req.Message)
+					if req.VirtualObjectKey != nil {
+						var output []byte
+						err := ctx.Object(
+							req.ServiceName,
+							*req.VirtualObjectKey,
+							req.HandlerName,
+							restate.WithBinary).
+							Request(input, &output)
+						if err != nil {
+							return nil, err
+						}
+						return byteArrayToIntArray(output), nil
+					} else {
+						var output []byte
+						err := ctx.Service(
+							req.ServiceName,
+							req.HandlerName,
+							restate.WithBinary).
+							Request(input, &output)
+						if err != nil {
+							return nil, err
+						}
+						return byteArrayToIntArray(output), nil
+					}
+				})).
+			Handler("oneWayCall", restate.NewServiceHandler(
+				// We need to use []int because Golang takes the opinionated choice of treating []byte as Base64
+				func(ctx restate.Context, req ProxyRequest) (restate.Void, error) {
+					input := intArrayToByteArray(req.Message)
+					if req.VirtualObjectKey != nil {
+						err := ctx.Object(
+							req.ServiceName,
+							*req.VirtualObjectKey,
+							req.HandlerName,
+							restate.WithBinary).
+							Send(input, 0)
+						return restate.Void{}, err
+					} else {
+						err := ctx.Service(
+							req.ServiceName,
+							req.HandlerName,
+							restate.WithBinary).
+							Send(input, 0)
+						return restate.Void{}, err
+					}
+				})).
+			Handler("manyCalls", restate.NewServiceHandler(
+				// We need to use []int because Golang takes the opinionated choice of treating []byte as Base64
+				func(ctx restate.Context, requests []ManyCallRequest) (restate.Void, error) {
+					var toAwait []restate.ResponseFuture
+
+					for _, req := range requests {
+						input := intArrayToByteArray(req.ProxyRequest.Message)
+						if req.OneWayCall {
+							if req.ProxyRequest.VirtualObjectKey != nil {
+								err := ctx.Object(
+									req.ProxyRequest.ServiceName,
+									*req.ProxyRequest.VirtualObjectKey,
+									req.ProxyRequest.HandlerName,
+									restate.WithBinary).
+									Send(input, 0)
+								return restate.Void{}, err
+							} else {
+								err := ctx.Service(
+									req.ProxyRequest.ServiceName,
+									req.ProxyRequest.HandlerName,
+									restate.WithBinary).
+									Send(input, 0)
+								return restate.Void{}, err
+							}
+						} else {
+							if req.ProxyRequest.VirtualObjectKey != nil {
+								fut, err := ctx.Object(
+									req.ProxyRequest.ServiceName,
+									*req.ProxyRequest.VirtualObjectKey,
+									req.ProxyRequest.HandlerName,
+									restate.WithBinary).
+									RequestFuture(input)
+								if err != nil {
+									return restate.Void{}, err
+								}
+								if req.AwaitAtTheEnd {
+									toAwait = append(toAwait, fut)
+								}
+							} else {
+								fut, err := ctx.Service(
+									req.ProxyRequest.ServiceName,
+									req.ProxyRequest.HandlerName,
+									restate.WithBinary).
+									RequestFuture(input)
+								if err != nil {
+									return restate.Void{}, err
+								}
+								if req.AwaitAtTheEnd {
+									toAwait = append(toAwait, fut)
+								}
+							}
+						}
+					}
+
+					// TODO replace this with select
+					for _, fut := range toAwait {
+						var output []byte
+						err := fut.Response(&output)
+						if err != nil {
+							return restate.Void{}, err
+						}
+					}
+					return restate.Void{}, nil
+				})))
+}
+
+func intArrayToByteArray(in []int) []byte {
+	out := make([]byte, len(in))
+	for idx, val := range in {
+		out[idx] = byte(val)
+	}
+	return out
+}
+
+func byteArrayToIntArray(in []byte) []int {
+	out := make([]int, len(in))
+	for idx, val := range in {
+		out[idx] = int(val)
+	}
+	return out
+}