Snapshot task queue families at the worker-deployment level - #11952
Snapshot task queue families at the worker-deployment level#11952Shivs11 wants to merge 1 commit into
Conversation
|
Claude finished @Shivs11's task in 8m 45s —— View job Reviewed 55 files, 7 findings. |
Shivs11
changed the title
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| if tc.wantSignalCnt > 0 { | ||
| env.OnSignalExternalWorkflow( | ||
| mock.Anything, | ||
| GenerateDeploymentWorkflowID(tv.DeploymentSeries()), | ||
| "", | ||
| SyncVersionSummarySignal, | ||
| mock.Anything, | ||
| ).Return(func(namespace string, workflowID string, runID string, signalName string, arg any) error { | ||
| capturedSummary = arg.(*deploymentspb.WorkerDeploymentVersionSummary) | ||
| return nil | ||
| }).Once() | ||
| } |
There was a problem hiding this comment.
med — The "later v3 run" case is vacuous: no signal mock is registered, so capturedSummary is nil regardless of what the workflow does.
capturedSummary is only ever assigned from inside the OnSignalExternalWorkflow return func, and that mock is registered only when tc.wantSignalCnt > 0. In the summarySent: true case the mock is skipped entirely, so require.Nil(t, capturedSummary) at line 160 holds even if run() did send a bootstrap summary — the test env just handles the unmocked SignalExternalWorkflow itself. The guard against re-bootstrapping is the whole point of that case, and nothing currently exercises it.
Register the mock unconditionally with .Maybe() so the capture is live in both cases (env.AssertExpectations then still covers the positive case via the assertion on the captured summary).
Suggestion:
| if tc.wantSignalCnt > 0 { | |
| env.OnSignalExternalWorkflow( | |
| mock.Anything, | |
| GenerateDeploymentWorkflowID(tv.DeploymentSeries()), | |
| "", | |
| SyncVersionSummarySignal, | |
| mock.Anything, | |
| ).Return(func(namespace string, workflowID string, runID string, signalName string, arg any) error { | |
| capturedSummary = arg.(*deploymentspb.WorkerDeploymentVersionSummary) | |
| return nil | |
| }).Once() | |
| } | |
| env.OnSignalExternalWorkflow( | |
| mock.Anything, | |
| GenerateDeploymentWorkflowID(tv.DeploymentSeries()), | |
| "", | |
| SyncVersionSummarySignal, | |
| mock.Anything, | |
| ).Return(func(namespace string, workflowID string, runID string, signalName string, arg any) error { | |
| capturedSummary = arg.(*deploymentspb.WorkerDeploymentVersionSummary) | |
| return nil | |
| }).Maybe() |
| describeResponse, err := env.FrontendClient().DescribeWorkflowExecution( | ||
| s.Context(), | ||
| &workflowservice.DescribeWorkflowExecutionRequest{ | ||
| Namespace: env.Namespace().String(), | ||
| Execution: &commonpb.WorkflowExecution{WorkflowId: deploymentWorkflowID}, | ||
| }, | ||
| ) | ||
| s.Require().NoError(err) | ||
| deploymentWorkflowRunID := describeResponse.GetWorkflowExecutionInfo().GetExecution().GetRunId() | ||
| s.Require().NotEmpty(deploymentWorkflowRunID) |
There was a problem hiding this comment.
med — The "update was never accepted" assertion can pass vacuously: the run ID is captured before the poll, and the update ID is re-derived from a duplicated production format.
This history scan is the only assertion covering the PR's headline behaviour (registration rejected at the validator, no child-update traffic), and two things let it pass without checking anything:
deploymentWorkflowRunIDis resolved beforepollFromDeploymentExpectFail. The Deployment workflow continues-as-new whenever its state changes, so if a CAN happens between here and the poll, the scan runs over a run that could never have contained the event.updateID(line 411) reproduces the format fromclient.go:355by hand. If that format ever changes,accepted.GetProtocolInstanceId() == updateIDstops matching and the test goes green while the behaviour regresses.
For (1), move the DescribeWorkflowExecution call to after pollFromDeploymentExpectFail. For (2), consider adding a positive control — assert the same derivation does find the accepted event for the first, successful registration — so a format drift fails the test instead of silencing it.
Suggestion: Resolve deploymentWorkflowRunID after the failed poll rather than before it.
| if err := d.validateRegisterWorker(args); err != nil { | ||
| return err | ||
| } |
There was a problem hiding this comment.
small — Re-running the validator inside the handler is unexplained, and when it fires it produces exactly the accepted-then-failed update this PR removes.
The same function is already wired as the update validator at line 371, so on the normal path this call is dead. It is only reachable when a SyncVersionSummarySignal lands between validation and the handler acquiring d.lock — and in that window it converts a would-be rejection into an accepted update that fails, writing the update-accepted/completed events the PR is trying to avoid. It also runs after the defer d.setStateChanged(), so a rejection here still dirties the workflow and forces a CaN.
Either drop it and let the child update be the backstop, or keep it with a comment naming the window it covers.
Suggestion:
| if err := d.validateRegisterWorker(args); err != nil { | |
| return err | |
| } | |
| // The validator ran before the update was accepted; re-check here because a | |
| // SyncVersionSummarySignal may have advanced the version summary while this | |
| // handler waited for the lock. | |
| if err := d.validateRegisterWorker(args); err != nil { | |
| return err | |
| } |
| filter := bloom.NewWithEstimates(uint(len(taskQueueFamilies)), taskQueueFamilyBloomFalsePositiveRate) | ||
| for _, taskQueueName := range workflow.DeterministicKeys(taskQueueFamilies) { | ||
| filter.AddString(taskQueueName) | ||
| } | ||
|
|
||
| serializedFilter, err := filter.MarshalBinary() | ||
| if err != nil { | ||
| return nil | ||
| } | ||
| summary.BloomFilter = serializedFilter | ||
| return summary |
There was a problem hiding this comment.
small — Dead error branch that silently discards the exact count, plus an unnecessary DeterministicKeys sort.
bloom.BloomFilter.MarshalBinary serializes into an in-memory buffer and has no failure mode, so return nil is unreachable. If it somehow were reached, dropping the whole summary also drops Count, which is independently useful, and nothing is logged — the Deployment workflow just silently fails open forever.
Separately, Bloom insertion is order-independent: the resulting bit set is identical for any iteration order, so workflow.DeterministicKeys buys no determinism here and just adds an allocation and a sort.
Suggestion:
| filter := bloom.NewWithEstimates(uint(len(taskQueueFamilies)), taskQueueFamilyBloomFalsePositiveRate) | |
| for _, taskQueueName := range workflow.DeterministicKeys(taskQueueFamilies) { | |
| filter.AddString(taskQueueName) | |
| } | |
| serializedFilter, err := filter.MarshalBinary() | |
| if err != nil { | |
| return nil | |
| } | |
| summary.BloomFilter = serializedFilter | |
| return summary | |
| filter := bloom.NewWithEstimates(uint(len(taskQueueFamilies)), taskQueueFamilyBloomFalsePositiveRate) | |
| for taskQueueName := range taskQueueFamilies { | |
| filter.AddString(taskQueueName) | |
| } | |
| serializedFilter, err := filter.MarshalBinary() | |
| if err != nil { | |
| return summary | |
| } | |
| summary.BloomFilter = serializedFilter | |
| return summary |
| func (d *VersionWorkflowRunner) versionStateToSummary( | ||
| s *deploymentspb.VersionLocalState, | ||
| ) *deploymentspb.WorkerDeploymentVersionSummary { |
There was a problem hiding this comment.
small — Method versionStateToSummary has the same name as the package-level function it wraps.
d.versionStateToSummary(s) and versionStateToSummary(s) now sit a few lines apart in the same file and differ only by the receiver. At every call site the reader has to check for a leading d. to know whether the task queue family summary is included.
Suggestion:
| func (d *VersionWorkflowRunner) versionStateToSummary( | |
| s *deploymentspb.VersionLocalState, | |
| ) *deploymentspb.WorkerDeploymentVersionSummary { | |
| func (d *VersionWorkflowRunner) summaryForCurrentWorkflowVersion( | |
| s *deploymentspb.VersionLocalState, | |
| ) *deploymentspb.WorkerDeploymentVersionSummary { |
| // Compute status for this version. Synced from the version workflow when WCI signals a status change. | ||
| temporal.api.deployment.v1.ComputeStatus compute_status = 14; | ||
|
|
||
| TaskQueueFamilySummary task_queue_family_summary = 15; |
There was a problem hiding this comment.
nit — Undocumented proto field; every sibling in this message has a comment.
Worth noting here that the summary is a snapshot published by the version workflow and may lag its exact state, since that is what makes the Deployment-side check safe to fail open.
Suggestion:
| TaskQueueFamilySummary task_queue_family_summary = 15; | |
| // Snapshot of the version's registered task queue families, published by the version | |
| // workflow. May lag the version workflow's exact state. | |
| TaskQueueFamilySummary task_queue_family_summary = 15; |
| func TestBuildTaskQueueFamilySummary_Empty(t *testing.T) { | ||
| t.Parallel() | ||
|
|
||
| summary := buildTaskQueueFamilySummary(nil) | ||
| require.Equal(t, int32(0), summary.GetCount()) | ||
| require.Empty(t, summary.GetBloomFilter()) | ||
| } |
There was a problem hiding this comment.
nit — Underscore in the test name, and the empty case is the same behaviour as the case above it.
The two other tests in this file are already table-driven; folding the empty map in as a {name: "no families"} row keeps the style consistent and drops the Test_-style suffix, which the repo guidelines avoid.
Suggestion: Fold this into TestBuildTaskQueueFamilySummary as a table case rather than a separate _Empty function.
| testCases := []struct { | ||
| name string | ||
| summarySent bool | ||
| wantSignalCnt int |
There was a problem hiding this comment.
nit — wantSignalCnt int only ever holds 0 or 1.
Every use is a > 0 / == 0 test, so the count adds nothing over a bool.
Suggestion:
| wantSignalCnt int | |
| wantSignal bool |
1 participant
What changed?
Why?
#11347 allows a new task queue type to register under an existing family name at the family limit. This follow-up lets the Deployment workflow distinguish that allowed case from a definitely new family, avoiding update acceptance and child workflow traffic when the limit is already known to be exceeded.
How did you test it?
Potential risks and rollout
Note
Medium Risk
Changes worker deployment registration validation and workflow state/proto fields; incorrect Bloom handling or mixed-version rollout could affect limit enforcement until v3 is enabled cell-by-cell via dynamic config.
Overview
Introduces workflow version 3 (
TaskQueueFamilySummary) so the Deployment workflow can tell a definitely new task queue family from an existing one at the family limit, and reject registration updates before they reach the Version workflow.Version summaries now carry a
TaskQueueFamilySummary(exact family count plus a serialized Bloom filter of family names). The Version workflow publishes that snapshot when a new family is registered and performs a one-time bootstrap after an active v2 run continues-as-new into v3 (tracked withtask_queue_family_summary_signal_sent).RegisterWorkerInWorkerDeploymentgains an update validator that fails fast when count is at the limit and the Bloom filter says the name is absent; Bloom false positives and missing summaries still defer to the Version workflow’s exact check.Adds the
bits-and-blooms/bloomdependency, unit/replay/integration tests, and a v3 replay history corpus. Production workflow version default remains v2 for separate rollout.Reviewed by Cursor Bugbot for commit 9a608da. Bugbot is set up for automated code reviews on this repo. Configure here.