From bcadca685c2ee5a67c98b20da351f55de72ceb80 Mon Sep 17 00:00:00 2001 From: Dmitriy Derepko Date: Sat, 22 Aug 2026 13:51:06 +0400 Subject: [PATCH] fix(plugin): read the workflow and activity definitions without the plugin mutex Plugin.Stop holds p.mu for the whole shutdown, and pool.Destroy inside it waits for the in-flight PHP workers to come back. An activity that calls Activity::heartbeat on its way out reaches RecordActivityHeartbeat, which takes p.mu.RLock twice to read rrActivityDef and blocks behind that writer, so the worker never returns to the pool, the drain never finishes and the lock is never released. The result is a circular wait that burns the entire endure grace period on every graceful stop that has a heartbeating activity in flight: the process is killed, the activity attempt is lost and is only recovered later by the server-side heartbeat timeout. Both definitions are written once, by initPool under Serve's lock, and are never replaced afterwards - Reset rebuilds the workers, not the definitions. A mutex was never the right tool for a write-once pointer, so store them in an atomic.Pointer, matching apiKey in the same struct, and drop the second RLock in RecordActivityHeartbeat which re-read what getActDef had already returned. The root package had no CI step, so add one - otherwise the regression test never runs. --- .github/workflows/linux.yml | 4 ++ internal.go | 12 ++--- plugin.go | 8 +-- plugin_stop_test.go | 103 ++++++++++++++++++++++++++++++++++++ rpc.go | 5 +- 5 files changed, 116 insertions(+), 16 deletions(-) create mode 100644 plugin_stop_test.go diff --git a/.github/workflows/linux.yml b/.github/workflows/linux.yml index 5284be15..a5b49a2e 100644 --- a/.github/workflows/linux.yml +++ b/.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/... diff --git a/internal.go b/internal.go index 0c0bf8c8..9fe5daaa 100644 --- a/internal.go +++ b/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 { diff --git a/plugin.go b/plugin.go index e0e9f8ae..b3fa9054 100644 --- a/plugin.go +++ b/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, diff --git a/plugin_stop_test.go b/plugin_stop_test.go new file mode 100644 index 00000000..1ea74497 --- /dev/null +++ b/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) +} diff --git a/rpc.go b/rpc.go index d52cf1a6..f1b27808 100644 --- a/rpc.go +++ b/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)