[#426]: chore: local repo tests with Go workspaces

.github/workflows/linux.yml

@@ -1,37 +1,56 @@
-name: Linux
+name: rrtemporal
 
 on:
   push:
     branches:
       - master
       - stable
-    tags-ignore:
-      - '**'
-    paths-ignore:
-      - '**.md'
   pull_request:
-    paths-ignore:
-      - '**.md'
+    branches:
+      - master
+      - stable
 
 jobs:
-  golang:
-    name: Build (Go ${{ matrix.go }}, OS ${{matrix.os}})
+  rrtemporal_test:
+    name: RR Temporal plugin (Go ${{ matrix.go }}, PHP ${{ matrix.php }}, OS ${{matrix.os}})
     runs-on: ${{ matrix.os }}
     timeout-minutes: 60
     strategy:
-      fail-fast: true
       matrix:
+        php: [ "8.2" ]
         go: [ stable ]
-        os: [ ubuntu-latest ]
+        os: [ "ubuntu-latest" ]
     steps:
       - name: Set up Go ${{ matrix.go }}
         uses: actions/setup-go@v4 # action page: <https://github.com/actions/setup-go>
         with:
           go-version: ${{ matrix.go }}
 
+      - name: Set up PHP ${{ matrix.php }}
+        uses: shivammathur/setup-php@v2 # action page: <https://github.com/shivammathur/setup-php>
+        with:
+          php-version: ${{ matrix.php }}
+          extensions: sockets
+
       - name: Check out code
         uses: actions/checkout@v4
 
+      - name: Get Composer Cache Directory
+        id: composer-cache
+        run: |
+          cd tests/php_test_files
+          echo "dir=$(composer config cache-files-dir)" >> $GITHUB_OUTPUT
+
+      - name: Init Composer Cache # Docs: <https://git.io/JfAKn#php---composer>
+        uses: actions/cache@v3
+        with:
+          path: ${{ steps.composer-cache.outputs.dir }}
+          key: ${{ runner.os }}-composer-${{ matrix.php }}-${{ hashFiles('**/composer.json') }}
+          restore-keys: ${{ runner.os }}-composer-
+
+      - name: Install Composer dependencies
+        run: cd tests/php_test_files && composer update --prefer-dist --no-progress --ansi
+
       - name: Init Go modules Cache # Docs: <https://git.io/JfAKn#go---modules>
         uses: actions/cache@v3
         with:
@@ -40,13 +59,62 @@ jobs:
           restore-keys: ${{ runner.os }}-go-
 
       - name: Install Go dependencies
-        run: go mod tidy && go mod download
+        run: go mod download
+
+      - name: Create folders
+        run: |
+          mkdir ./tests/coverage-ci
 
-      - name: Golang tests
-        run: make test_coverage
+      - name: Run Temporal root module tests
+        run: |
+          go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=$(cat ./tests/pkgs.txt) -coverprofile=./tests/coverage-ci/rrt_root.out -covermode=atomic ./...
+
+      - name: Run Temporal TLS tests with coverage
+        run: |
+          cd tests
+          
+          ./env/temporal_tls/generate-test-certs.sh
+          export TEMPORAL_TLS_CERTS_DIR=/etc/temporal/config/certs
+          export TEMPORAL_LOCAL_CERT_DIR=$(pwd)/env/temporal_tls/certs
+          docker-compose -f env/temporal_tls/docker-compose.yml up -d --remove-orphans
+          sleep 60
+          
+          go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=$(cat pkgs.txt) -coverprofile=./coverage-ci/rrt_tls.out -covermode=atomic ./tls/
+          docker-compose -f env/temporal_tls/docker-compose.yml down
+
+      - name: Run Temporal tests with coverage
+        run: |
+          cd tests
+          docker-compose -f env/docker-compose-temporal.yaml up -d --remove-orphans
+
+          go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=$(cat pkgs.txt) -coverprofile=./coverage-ci/rrt.out -covermode=atomic ./
+          docker-compose -f env/docker-compose-temporal.yaml up -d --remove-orphans
+
+      - name: Archive code coverage results
+        uses: actions/upload-artifact@v3
+        with:
+          name: coverage
+          path: ./tests/coverage-ci
+
+
+  codecov:
+    name: Upload codecov
+    runs-on: ubuntu-latest
+    needs:
+      - rrtemporal_test
+
+    timeout-minutes: 60
+    steps:
+      - name: Download code coverage results
+        uses: actions/download-artifact@v3
+      - run: |
+          cd coverage
+          echo 'mode: atomic' > summary.txt
+          tail -q -n +2 *.out >> summary.txt
+          sed -i '2,${/roadrunner/!d}' summary.txt
 
-      - uses: codecov/codecov-action@v3.1.4 # Docs: <https://github.com/codecov/codecov-action>
+      - name: upload to codecov
+        uses: codecov/codecov-action@v3 # Docs: <https://github.com/codecov/codecov-action>
         with:
-          token: ${{ secrets.CODECOV_TOKEN }}
-          file: ./coverage-ci/summary.txt
+          file: ./coverage/summary.txt
           fail_ci_if_error: false

.gitignore

@@ -14,9 +14,7 @@
 *.out
 
 # Dependency directories (remove the comment below to include it)
-# vendor/
+vendor/
 .idea
 
-**/composer.json
-**/composer.lock
 **/vendor

README.md

@@ -1,4 +1,4 @@
-[![Linux](https://github.com/temporalio/roadrunner-temporal/workflows/Linux/badge.svg)](https://github.com/temporalio/roadrunner-temporal/actions)
+[![Linux](https://github.com/temporalio/roadrunner-temporal/workflows/rrtemporal/badge.svg)](https://github.com/temporalio/roadrunner-temporal/actions)
 [![Discourse](https://img.shields.io/static/v1?label=Discourse&message=Get%20Help&color=informational)](https://community.temporal.io)
 <a href="https://discord.gg/TFeEmCs"><img src="https://img.shields.io/badge/discord-chat-magenta.svg"></a>
 
@@ -7,7 +7,7 @@ The repository contains a number of plugins which enable workflow and activity p
 supervisor, load-balancer is based on [RoadRunner PHP Application Server](https://roadrunner.dev).
 
 ## Installation
-Temporal is official plugin of RoadRunner and available out of the box in [RoadRunner 2.0](https://github.com/roadrunner-server/roadrunner).
+Temporal is an official plugin of RoadRunner and available out-of-the-box in >= [RoadRunner 2023.0](https://github.com/roadrunner-server/roadrunner).
 
 Read more about application server installation [here](https://roadrunner.dev/docs/intro-install).
 

aggregatedpool/activity.go

@@ -91,15 +91,16 @@ func (a *Activity) execute(ctx context.Context, args *commonpb.Payloads) (*commo
 		msg.Payloads.Payloads = append(msg.Payloads.Payloads, heartbeatDetails.Payloads...)
 	}
 
-	pld := a.getPld()
-	defer a.putPld(pld)
+	pl := a.getPld()
+	defer a.putPld(pl)
 
-	err := a.codec.Encode(&internal.Context{TaskQueue: info.TaskQueue}, pld, msg)
+	err := a.codec.Encode(&internal.Context{TaskQueue: info.TaskQueue}, pl, msg)
 	if err != nil {
 		return nil, err
 	}
 
-	result, err := a.pool.Exec(ctx, pld, nil)
+	ch := make(chan struct{}, 1)
+	result, err := a.pool.Exec(ctx, pl, ch)
 	if err != nil {
 		a.running.Delete(bytesToStr(info.TaskToken))
 		return nil, errors.E(op, err)
@@ -115,6 +116,7 @@ func (a *Activity) execute(ctx context.Context, args *commonpb.Payloads) (*commo
 		}
 		// streaming is not supported
 		if pld.Payload().Flags&frame.STREAM != 0 {
+			ch <- struct{}{}
 			return nil, errors.E(op, errors.Str("streaming is not supported"))
 		}
 

aggregatedpool/handler.go

@@ -355,14 +355,15 @@ func (wp *Workflow) flushQueue() error {
 		defer wp.mh.Gauge(RrWorkflowsMetricName).Update(float64(wp.pool.QueueSize()))
 	}
 
-	pld := wp.getPld()
-	defer wp.putPld(pld)
-	err := wp.codec.Encode(wp.getContext(), pld, wp.mq.Messages()...)
+	pl := wp.getPld()
+	defer wp.putPld(pl)
+	err := wp.codec.Encode(wp.getContext(), pl, wp.mq.Messages()...)
 	if err != nil {
 		return err
 	}
 
-	result, err := wp.pool.Exec(context.Background(), pld, nil)
+	ch := make(chan struct{}, 1)
+	result, err := wp.pool.Exec(context.Background(), pl, ch)
 	if err != nil {
 		return err
 	}
@@ -375,6 +376,7 @@ func (wp *Workflow) flushQueue() error {
 		}
 		// streaming is not supported
 		if pld.Payload().Flags&frame.STREAM != 0 {
+			ch <- struct{}{}
 			return errors.E(op, errors.Str("streaming is not supported"))
 		}
 
@@ -411,48 +413,52 @@ func (wp *Workflow) runCommand(cmd any, payloads *commonpb.Payloads, header *com
 		defer wp.mh.Gauge(RrMetricName).Update(float64(wp.pool.QueueSize()))
 	}
 
-	pld := wp.getPld()
-	err := wp.codec.Encode(wp.getContext(), pld, msg)
+	pl := wp.getPld()
+	err := wp.codec.Encode(wp.getContext(), pl, msg)
 	if err != nil {
-		wp.putPld(pld)
+		wp.putPld(pl)
 		return nil, err
 	}
 
 	// todo(rustatian): do we need a timeout here??
-	result, err := wp.pool.Exec(context.Background(), pld, nil)
+	ch := make(chan struct{}, 1)
+	result, err := wp.pool.Exec(context.Background(), pl, ch)
 	if err != nil {
-		wp.putPld(pld)
+		wp.putPld(pl)
 		return nil, err
 	}
 
 	var r *payload.Payload
-	for p := range result {
-		if p.Error() != nil {
-			wp.putPld(pld)
-			return nil, errors.E(op, p.Error())
+	select {
+	case pld := <-result:
+		if pld.Error() != nil {
+			return nil, errors.E(op, pld.Error())
 		}
 		// streaming is not supported
-		if p.Payload().Flags&frame.STREAM != 0 {
-			wp.putPld(pld)
+		if pld.Payload().Flags&frame.STREAM != 0 {
+			ch <- struct{}{}
 			return nil, errors.E(op, errors.Str("streaming is not supported"))
 		}
-		// save the payload
-		r = p.Payload()
+
+		// assign the payload
+		r = pld.Payload()
+	default:
+		return nil, errors.E(op, errors.Str("worker empty response"))
 	}
 
 	msgs := make([]*internal.Message, 0, 2)
 	err = wp.codec.Decode(r, &msgs)
 	if err != nil {
-		wp.putPld(pld)
+		wp.putPld(pl)
 		return nil, err
 	}
 
 	if len(msgs) != 1 {
-		wp.putPld(pld)
+		wp.putPld(pl)
 		return nil, errors.E(op, errors.Str("unexpected pool response"))
 	}
 
-	wp.putPld(pld)
+	wp.putPld(pl)
 	return msgs[0], nil
 }
 

aggregatedpool/local_activity.go

@@ -58,15 +58,16 @@ func (la *LocalActivityFn) execute(ctx context.Context, args *commonpb.Payloads)
 
 	la.log.Debug("executing local activity fn", zap.Uint64("ID", msg.ID), zap.String("task-queue", info.TaskQueue), zap.String("la ID", info.ActivityID))
 
-	pld := getPld()
-	defer putPld(pld)
+	pl := getPld()
+	defer putPld(pl)
 
-	err := la.codec.Encode(&internal.Context{TaskQueue: info.TaskQueue}, pld, msg)
+	err := la.codec.Encode(&internal.Context{TaskQueue: info.TaskQueue}, pl, msg)
 	if err != nil {
 		return nil, err
 	}
 
-	result, err := la.pool.Exec(ctx, pld, nil)
+	ch := make(chan struct{}, 1)
+	result, err := la.pool.Exec(ctx, pl, ch)
 	if err != nil {
 		return nil, errors.E(op, err)
 	}
@@ -79,6 +80,7 @@ func (la *LocalActivityFn) execute(ctx context.Context, args *commonpb.Payloads)
 		}
 		// streaming is not supported
 		if pld.Payload().Flags&frame.STREAM != 0 {
+			ch <- struct{}{}
 			return nil, errors.E(op, errors.Str("streaming is not supported"))
 		}
 

go.mod

@@ -2,17 +2,17 @@ module github.com/temporalio/roadrunner-temporal/v4
 
 go 1.21
 
-toolchain go1.21.0
+toolchain go1.21.1
 
 require (
 	github.com/goccy/go-json v0.10.2
 	github.com/google/uuid v1.3.1
 	github.com/roadrunner-server/api/v4 v4.8.0
 	github.com/roadrunner-server/endure/v2 v2.4.2
 	github.com/roadrunner-server/errors v1.3.0
-	github.com/roadrunner-server/sdk/v4 v4.5.0-beta.1
+	github.com/roadrunner-server/sdk/v4 v4.5.2
 	github.com/stretchr/testify v1.8.4
-	github.com/uber-go/tally/v4 v4.1.7
+	github.com/uber-go/tally/v4 v4.1.9
 	go.temporal.io/api v1.24.0
 	go.temporal.io/sdk v1.24.0
 	go.temporal.io/sdk/contrib/tally v0.2.0
@@ -39,8 +39,8 @@ require (
 	github.com/pborman/uuid v1.2.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/client_golang v1.16.0
-	github.com/prometheus/client_model v0.4.0 // indirect
+	github.com/prometheus/client_golang v1.17.0
+	github.com/prometheus/client_model v0.5.0 // indirect
 	github.com/prometheus/common v0.44.0 // indirect
 	github.com/prometheus/procfs v0.12.0 // indirect
 	github.com/roadrunner-server/goridge/v3 v3.8.1
@@ -54,13 +54,13 @@ require (
 	go.uber.org/atomic v1.11.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/net v0.15.0 // indirect
-	golang.org/x/sync v0.3.0 // indirect
-	golang.org/x/sys v0.12.0 // indirect
+	golang.org/x/sync v0.4.0 // indirect
+	golang.org/x/sys v0.13.0 // indirect
 	golang.org/x/text v0.13.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
-	google.golang.org/genproto v0.0.0-20230920204549-e6e6cdab5c13 // indirect
-	google.golang.org/genproto/googleapis/api v0.0.0-20230920204549-e6e6cdab5c13 // indirect
-	google.golang.org/genproto/googleapis/rpc v0.0.0-20230920204549-e6e6cdab5c13 // indirect
+	google.golang.org/genproto v0.0.0-20231002182017-d307bd883b97 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20231002182017-d307bd883b97 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20231002182017-d307bd883b97 // indirect
 	google.golang.org/grpc v1.58.2
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )