fix(plugin): read the workflow and activity definitions without the plugin mutex

.github/workflows/linux.yml

@@ -222,6 +222,10 @@ jobs:
 
           sudo update-ca-certificates
 
+      - name: Run Temporal plugin tests
+        run: |
+          go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=github.com/temporalio/roadrunner-temporal/v5/... -coverprofile=./tests/coverage-ci/rrt_p.out -covermode=atomic .
+
       - name: Run Temporal canceller module tests
         run: |
           go test -timeout 20m -v -race -cover -tags=debug -failfast -coverpkg=github.com/temporalio/roadrunner-temporal/v6/... -coverprofile=./tests/coverage-ci/rrt_c.out -covermode=atomic ./canceller/...

internal.go

@@ -126,8 +126,8 @@ func (p *Plugin) initPool() error {
 		}
 	}
 
-	p.temporal.rrWorkflowDef = wfDef
-	p.temporal.rrActivityDef = actDef
+	p.temporal.rrWorkflowDef.Store(wfDef)
+	p.temporal.rrActivityDef.Store(actDef)
 	p.temporal.workers = workers
 	p.codec = codec
 
@@ -140,15 +140,11 @@ func (p *Plugin) initPool() error {
 }
 
 func (p *Plugin) getWfDef() *aggregatedpool.Workflow {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	return p.temporal.rrWorkflowDef
+	return p.temporal.rrWorkflowDef.Load()
 }
 
 func (p *Plugin) getActDef() *aggregatedpool.Activity {
-	p.mu.RLock()
-	defer p.mu.RUnlock()
-	return p.temporal.rrActivityDef
+	return p.temporal.rrActivityDef.Load()
 }
 
 func (p *Plugin) initTemporalClient(phpSdkVersion string, flags map[string]string, dc converter.DataConverter) error {

plugin.go

@@ -53,8 +53,8 @@ type Logger interface {
 
 // temporal structure contains temporal specific structures
 type temporal struct {
-	rrActivityDef *aggregatedpool.Activity
-	rrWorkflowDef *aggregatedpool.Workflow
+	rrActivityDef atomic.Pointer[aggregatedpool.Activity]
+	rrWorkflowDef atomic.Pointer[aggregatedpool.Workflow]
 	workflows     map[string]*internal.WorkflowInfo
 	activities    map[string]*internal.ActivityInfo
 	mh            tclient.MetricsHandler
@@ -360,8 +360,8 @@ func (p *Plugin) Reset() error {
 
 	// based on the worker info -> initialize workers
 	workers, err := aggregatedpool.TemporalWorkers(
-		p.temporal.rrWorkflowDef,
-		p.temporal.rrActivityDef,
+		p.temporal.rrWorkflowDef.Load(),
+		p.temporal.rrActivityDef.Load(),
 		wi,
 		p.log,
 		p.temporal.client,

plugin_stop_test.go

@@ -0,0 +1,103 @@
+package rrtemporal
+
+import (
+	"context"
+	"log/slog"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/roadrunner-server/events"
+	"github.com/stretchr/testify/require"
+	"github.com/temporalio/roadrunner-temporal/v6/internal"
+	"go.temporal.io/sdk/worker"
+)
+
+const stopTestTimeout = 5 * time.Second
+
+type blockingWorker struct {
+	worker.Worker
+
+	stopping chan struct{}
+	release  chan struct{}
+}
+
+func (b *blockingWorker) Stop() {
+	close(b.stopping)
+	<-b.release
+}
+
+type drainingPlugin struct {
+	plugin  *Plugin
+	release func()
+	stopped chan error
+}
+
+// startDraining leaves Stop blocked inside the shutdown, holding p.mu, which is
+// where a heartbeat from a still-running activity arrives.
+func startDraining(t *testing.T) *drainingPlugin {
+	t.Helper()
+
+	w := &blockingWorker{
+		stopping: make(chan struct{}),
+		release:  make(chan struct{}),
+	}
+
+	p := &Plugin{
+		log:    slog.New(slog.DiscardHandler),
+		stopCh: make(chan struct{}, 1),
+		temporal: &temporal{
+			activities: map[string]*internal.ActivityInfo{},
+			workflows:  map[string]*internal.WorkflowInfo{},
+			workers:    []worker.Worker{w},
+		},
+	}
+	p.eventBus, p.id = events.NewEventBus()
+
+	var once sync.Once
+	release := func() { once.Do(func() { close(w.release) }) }
+	t.Cleanup(release)
+
+	stopped := make(chan error, 1)
+	go func() { stopped <- p.Stop(context.Background()) }()
+
+	select {
+	case <-w.stopping:
+	case <-time.After(stopTestTimeout):
+		require.FailNow(t, "Stop did not reach the worker drain")
+	}
+
+	return &drainingPlugin{plugin: p, release: release, stopped: stopped}
+}
+
+func (d *drainingPlugin) finish(t *testing.T) {
+	t.Helper()
+
+	d.release()
+
+	select {
+	case err := <-d.stopped:
+		require.NoError(t, err)
+	case <-time.After(stopTestTimeout):
+		require.FailNow(t, "Stop did not return after the drain finished")
+	}
+}
+
+func TestDefinitionsAreReadableWhileStopping(t *testing.T) {
+	d := startDraining(t)
+
+	done := make(chan struct{})
+	go func() {
+		d.plugin.getActDef()
+		d.plugin.getWfDef()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(stopTestTimeout):
+		require.FailNow(t, "reading the activity definition blocked while the plugin was stopping")
+	}
+
+	d.finish(t)
+}

rpc.go

@@ -67,13 +67,10 @@ func (r *rpc) RecordActivityHeartbeat(in RecordHeartbeatRequest, out *RecordHear
 	}
 
 	// find running activity
-	r.plugin.mu.RLock()
-	ctx, err := r.plugin.temporal.rrActivityDef.GetActivityContext(in.TaskToken)
+	ctx, err := r.plugin.getActDef().GetActivityContext(in.TaskToken)
 	if err != nil {
-		r.plugin.mu.RUnlock()
 		return err
 	}
-	r.plugin.mu.RUnlock()
 
 	activity.RecordHeartbeat(ctx, details)