Github/temporal
1
0
Fork 0
mirror of https://github.com/temporalio/temporal.git synced 2026-08-30 18:41:49 -07:00
Code Issues Packages Projects Releases Wiki Activity

Use refreshArgs snapshot consistently in startWorker

Browse Source 
startWorker was reading w.ns and w.opts directly while refresh used the
snapshot (args) for the componentSet cache key. This could cause the
cache key to disagree with the running worker config. Pass the snapshot
through to startWorker so both the cache key and the worker are built
from the same values.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Kannan Rajah
2026-08-27 19:27:12 -07:00
parent b3dede2eb3
commit 8d5fa06545
1 changed files with 14 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
service/worker/pernamespaceworker.go
View File

@@ -236,6 +236,8 @@ func (wm *PerNamespaceWorkerManager) getWorkerByNamespace(ns *namespace.Namespac
logger: log.With(wm.logger, tag.WorkflowNamespace(ns.Name().String())),
retrier: backoff.NewRetrier(backoff.NewExponentialRetryPolicy(wm.initialRetry), clock.NewRealTimeSource()),
}
// Hold lock before registering dynamic config subscriptions: their callbacks can fire
// concurrently and must not observe uninitialized fields.
worker.lock.Lock()
count, c1 := wm.config.PerNamespaceWorkerCount(ns.Name().String(), worker.setWorkerCount)
opts, c2 := wm.config.PerNamespaceWorkerOptions(ns.Name().String(), worker.setWorkerOptions)
@@ -448,7 +450,7 @@ func (w *perNamespaceWorker) refresh(args refreshArgs) (retErr error) {
// create new one. note that even before startWorker returns, the worker may have started
// and already called the fatal error handler. we need to set w.client+worker+componentSet
// before releasing the lock to keep our state consistent.
client, worker, err := w.startWorker(enabledComponents, workerAllocation)
client, worker, err := w.startWorker(args, enabledComponents, workerAllocation)
if err != nil {
// TODO: add metric also
w.stopWorkerLocked() // for calling cleanup
@@ -462,10 +464,11 @@ func (w *perNamespaceWorker) refresh(args refreshArgs) (retErr error) {
}
func (w *perNamespaceWorker) startWorker(
args refreshArgs,
components []workercommon.PerNSWorkerComponent,
allocation workerAllocation,
) (sdkclient.Client, sdkworker.Worker, error) {
nsName := w.ns.Name().String()
nsName := args.ns.Name().String()
// this should not block because it uses an existing grpc connection
client := w.wm.sdkClientFactory.NewClient(sdkclient.Options{
Namespace: nsName,
@@ -476,14 +479,14 @@ func (w *perNamespaceWorker) startWorker(
// copy from dynamic config. apply explicit defaults for some instead of using the sdk
// defaults so that we can multiply below.
sdkoptions.MaxConcurrentActivityExecutionSize = cmp.Or(w.opts.MaxConcurrentActivityExecutionSize, 1000)
sdkoptions.WorkerActivitiesPerSecond = w.opts.WorkerActivitiesPerSecond
sdkoptions.MaxConcurrentLocalActivityExecutionSize = cmp.Or(w.opts.MaxConcurrentLocalActivityExecutionSize, 1000)
sdkoptions.WorkerLocalActivitiesPerSecond = w.opts.WorkerLocalActivitiesPerSecond
sdkoptions.MaxConcurrentActivityTaskPollers = max(cmp.Or(w.opts.MaxConcurrentActivityTaskPollers, 2), 2)
sdkoptions.MaxConcurrentWorkflowTaskExecutionSize = cmp.Or(w.opts.MaxConcurrentWorkflowTaskExecutionSize, 1000)
sdkoptions.MaxConcurrentWorkflowTaskPollers = max(cmp.Or(w.opts.MaxConcurrentWorkflowTaskPollers, 2), 2)
sdkoptions.StickyScheduleToStartTimeout = w.opts.StickyScheduleToStartTimeout
sdkoptions.MaxConcurrentActivityExecutionSize = cmp.Or(args.opts.MaxConcurrentActivityExecutionSize, 1000)
sdkoptions.WorkerActivitiesPerSecond = args.opts.WorkerActivitiesPerSecond
sdkoptions.MaxConcurrentLocalActivityExecutionSize = cmp.Or(args.opts.MaxConcurrentLocalActivityExecutionSize, 1000)
sdkoptions.WorkerLocalActivitiesPerSecond = args.opts.WorkerLocalActivitiesPerSecond
sdkoptions.MaxConcurrentActivityTaskPollers = max(cmp.Or(args.opts.MaxConcurrentActivityTaskPollers, 2), 2)
sdkoptions.MaxConcurrentWorkflowTaskExecutionSize = cmp.Or(args.opts.MaxConcurrentWorkflowTaskExecutionSize, 1000)
sdkoptions.MaxConcurrentWorkflowTaskPollers = max(cmp.Or(args.opts.MaxConcurrentWorkflowTaskPollers, 2), 2)
sdkoptions.StickyScheduleToStartTimeout = args.opts.StickyScheduleToStartTimeout
sdkoptions.BackgroundActivityContext = headers.SetCallerInfo(context.Background(), headers.NewBackgroundHighCallerInfo(nsName))
sdkoptions.Identity = fmt.Sprintf("temporal-system@%s@%s", w.wm.hostName, nsName)
@@ -502,7 +505,7 @@ func (w *perNamespaceWorker) startWorker(
Multiplicity: allocation.local,
}
for _, cmp := range components {
cleanup := cmp.Register(worker, w.ns, details)
cleanup := cmp.Register(worker, args.ns, details)
if cleanup != nil {
w.cleanup = append(w.cleanup, cleanup)
}