Snapshot task queue families at the worker-deployment level

common/metrics/metric_defs.go

@@ -1676,6 +1676,7 @@ var (
 	WorkerDeploymentVersioningOverrideCounter         = NewCounterDef("worker_deployment_versioning_override_count")
 	WorkerDeploymentVersioningOneTimeOverrideCounter  = NewCounterDef("worker_deployment_versioning_one_time_override_count")
 	WorkerDeploymentVersionDeletePropagationFailure   = NewCounterDef("worker_deployment_version_delete_propagation_failure")
+	WorkerDeploymentTaskQueueFamilyBloomFilterOutcome = NewCounterDef("worker_deployment_task_queue_family_bloom_filter_outcome")
 	StartDeploymentTransitionCounter                  = NewCounterDef("start_deployment_transition_count")
 	VersioningDataPropagationLatency                  = NewTimerDef("versioning_data_propagation_latency")
 	SlowVersioningDataPropagationCounter              = NewCounterDef("slow_versioning_data_propagation")

go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/credentials v1.19.15
 	github.com/aws/aws-sdk-go-v2/service/s3 v1.99.1
 	github.com/aws/smithy-go v1.25.0
+	github.com/bits-and-blooms/bloom/v3 v3.7.1
 	github.com/blang/semver/v4 v4.0.0
 	github.com/cactus/go-statsd-client/v5 v5.1.0
 	github.com/caio/go-tdigest/v5 v5.0.0
@@ -90,6 +91,7 @@ require (
 
 require (
 	github.com/aclements/go-moremath v0.0.0-20210112150236-f10218a38794 // indirect
+	github.com/bits-and-blooms/bitset v1.24.2 // indirect
 	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
 	github.com/clipperhouse/uax29/v2 v2.7.0 // indirect
 	github.com/go-openapi/swag/cmdutils v0.26.0 // indirect

go.sum

@@ -99,6 +99,10 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932 h1:mXoPYz/Ul5HYEDvkta6I8/rnYM5gSdSV2tJ6XbZuEtY=
 github.com/bitly/go-hostpool v0.0.0-20171023180738-a3a6125de932/go.mod h1:NOuUCSz6Q9T7+igc/hlvDOUdtWKryOrtFyIVABv/p7k=
+github.com/bits-and-blooms/bitset v1.24.2 h1:M7/NzVbsytmtfHbumG+K2bremQPMJuqv1JD3vOaFxp0=
+github.com/bits-and-blooms/bitset v1.24.2/go.mod h1:7hO7Gc7Pp1vODcmWvKMRA9BNmbv6a/7QIWpPxHddWR8=
+github.com/bits-and-blooms/bloom/v3 v3.7.1 h1:WXovk4TRKZttAMJfoQx6K2DM0zNIt8w+c67UqO+etV0=
+github.com/bits-and-blooms/bloom/v3 v3.7.1/go.mod h1:rZzYLLje2dfzXfAkJNxQQHsKurAyK55KUnL43Euk0hU=
 github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM=
 github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
 github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=

proto/internal/temporal/server/api/deployment/v1/message.proto

@@ -153,6 +153,13 @@ message VersionLocalState {
 
   // Cached compute status, updated when WCI signals this version workflow.
   temporal.api.deployment.v1.ComputeStatus compute_status = 19;
+
+  // Active Current or Ramping v2 Version workflows do not publish task queue family Bloom filter snapshots.
+  // After a successful task queue registration changes their state, they continue as new into v3. The new run
+  // must publish a snapshot of this Bloom filter so that the Deployment workflow can learn about them and use it
+  // for validation. Note: this field, once set to true, is carried across every continues-as-new run of the
+  // version workflow. It only becomes false when the version workflow is deleted and is then later recreated.
+  bool task_queue_family_summary_signal_sent = 20;
 }
 
 // Data specific to a task queue, from the perspective of a worker deployment version.
@@ -239,6 +246,10 @@ message 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;
+
+  // Snapshot of registered task queue families published by the Version workflow.
+  // It is refreshed periodically or after relevant state changes and may lag the Version workflow's exact state.
+  TaskQueueFamilySummary task_queue_family_summary = 15;
 }
 
 // used as Worker Deployment Version workflow update input:
@@ -625,3 +636,14 @@ message ForceCANVersionSignalArgs {
 message DemoteVersionSignalArgs {
   temporal.api.deployment.v1.RoutingConfig routing_config = 1;
 }
+
+message TaskQueueFamilySummary {
+  // Exact number of task queue families registered in the Version. This is n when sizing the Bloom filter.
+  int32 count = 1;
+  // Number of bits in the Bloom filter.
+  int64 bloom_filter_size = 2;
+  // Number of hash functions used by the Bloom filter.
+  int32 bloom_filter_hash_count = 3;
+  // Bloom filter words containing the registered task queue family names.
+  repeated int64 bloom_filter_words = 4;
+}

service/worker/workerdeployment/fx.go

@@ -33,6 +33,8 @@ const (
 	AsyncSetCurrentAndRamping
 	// Version Data has its own revision number with TaskQueue registration being async as well
 	VersionDataRevisionNumber
+	// Version summaries include task queue family membership information.
+	TaskQueueFamilySummary
 )
 
 type (

service/worker/workerdeployment/replaytester/replay_test.go

@@ -53,7 +53,7 @@ func TestReplays(t *testing.T) {
 
 func testReplays(t *testing.T, versionDemotionSignalEnabled bool) {
 	// For each workflow implementation version we run all the replay tests for snapshots created by that version or older versions
-	for wv := workerdeployment.InitialVersion; wv <= workerdeployment.VersionDataRevisionNumber; wv++ {
+	for wv := workerdeployment.InitialVersion; wv <= workerdeployment.TaskQueueFamilySummary; wv++ {
 		replayer := worker.NewWorkflowReplayer()
 
 		// Create version workflow wrapper to match production registration

service/worker/workerdeployment/replaytester/testdata/v3/run_1788826427/expected_counts.txt

@@ -0,0 +1,6 @@
+# Expected workflow counts for replay testing
+# Generated by generate_history.sh on Mon Sep  7 20:13:49 EDT 2026
+EXPECTED_DEPLOYMENT_WORKFLOWS=25
+EXPECTED_VERSION_WORKFLOWS=14
+ACTUAL_DEPLOYMENT_WORKFLOWS=25
+ACTUAL_VERSION_WORKFLOWS=14

service/worker/workerdeployment/task_queue_family_summary.go

@@ -0,0 +1,52 @@
+package workerdeployment
+
+import (
+	"github.com/bits-and-blooms/bloom/v3"
+	"go.temporal.io/sdk/workflow"
+	deploymentspb "go.temporal.io/server/api/deployment/v1"
+)
+
+const taskQueueFamilyBloomFalsePositiveRate = 0.01
+
+func buildTaskQueueFamilySummary(
+	taskQueueFamilies map[string]*deploymentspb.VersionLocalState_TaskQueueFamilyData,
+) *deploymentspb.TaskQueueFamilySummary {
+	summary := &deploymentspb.TaskQueueFamilySummary{Count: int32(len(taskQueueFamilies))}
+	if len(taskQueueFamilies) == 0 {
+		return summary
+	}
+
+	filter := bloom.NewWithEstimates(uint(len(taskQueueFamilies)), taskQueueFamilyBloomFalsePositiveRate)
+	for _, taskQueueName := range workflow.DeterministicKeys(taskQueueFamilies) {
+		filter.AddString(taskQueueName)
+	}
+
+	summary.BloomFilterSize = int64(filter.Cap())
+	summary.BloomFilterHashCount = int32(filter.K())
+	summary.BloomFilterWords = make([]int64, len(filter.BitSet().Words()))
+	for index, word := range filter.BitSet().Words() {
+		summary.BloomFilterWords[index] = int64(word)
+	}
+	return summary
+}
+
+func taskQueueFamilyMayExist(summary *deploymentspb.TaskQueueFamilySummary, taskQueueName string) bool {
+	count := summary.GetCount()
+	if count <= 0 {
+		return true
+	}
+
+	filterWords := summary.GetBloomFilterWords()
+	// TODO: Validate the Bloom filter metadata before reconstructing it.
+	// The proto uses int64 words to satisfy proto lint, while the Bloom library requires uint64.
+	// This conversion preserves the existing bitset; it does not rebuild the filter from task queue names.
+	words := make([]uint64, len(filterWords))
+	for index, word := range filterWords {
+		words[index] = uint64(word)
+	}
+	return bloom.FromWithM(
+		words,
+		uint(summary.GetBloomFilterSize()),
+		uint(summary.GetBloomFilterHashCount()),
+	).TestString(taskQueueName)
+}