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)