Implement unified async callback queue for namespace registry

.github/workflows/claude-review-teams.yml

@@ -40,12 +40,24 @@ jobs:
       head_sha: ${{ steps.resolve.outputs.head_sha }}
       pr_number: ${{ steps.resolve.outputs.pr_number }}
     steps:
+      - name: Check secrets
+        id: check_secrets
+        env:
+          APP_ID: ${{ secrets.TEMPORAL_CICD_APP_ID }}
+        run: |
+          if [ -z "$APP_ID" ]; then
+            echo "TEMPORAL_CICD_APP_ID is not set. Skipping review."
+            echo "has_secret=false" >> "$GITHUB_OUTPUT"
+          else
+            echo "has_secret=true" >> "$GITHUB_OUTPUT"
+          fi
       - name: Generate app token
         id: generate_token
+        if: steps.check_secrets.outputs.has_secret == 'true'
         uses: actions/create-github-app-token@fee1f7d63c2ff003460e3d139729b119787bc349 # v2.2.2
         with:
-          app-id: ${{ secrets.TEMPORAL_AI_REVIEW_APP_ID }}
-          private-key: ${{ secrets.TEMPORAL_AI_REVIEW_APP_PRIVATE_KEY }}
+          app-id: ${{ secrets.TEMPORAL_CICD_APP_ID }}
+          private-key: ${{ secrets.TEMPORAL_CICD_PRIVATE_KEY }}
           owner: ${{ github.repository_owner }}
           # Least-privilege: members:read for team-membership lookups,
           # pull-requests:read for PR metadata, and
@@ -54,6 +66,7 @@ jobs:
           permission-pull-requests: read
           permission-issues: read
       - id: resolve
+        if: steps.check_secrets.outputs.has_secret == 'true'
         env:
           EVENT_PR_NUMBER: ${{ github.event.pull_request.number }}
           PR_AUTHOR: ${{ github.event.pull_request.user.login }}
@@ -72,6 +85,7 @@ jobs:
             echo "head_sha=${head_sha}"
           } >> "$GITHUB_OUTPUT"
       - id: decision
+        if: steps.check_secrets.outputs.has_secret == 'true'
         env:
           GH_TOKEN: ${{ steps.generate_token.outputs.token }}
           AUTHOR: ${{ steps.resolve.outputs.author }}

common/namespace/nsregistry/registry.go

@@ -60,6 +60,8 @@ const (
 	startWatchMaxAttempts = 10
 	// Metrics and logs are emitted for callbacks that take longer than slowCallbackDuration
 	slowCallbackDuration = 250 * time.Millisecond
+
+	callbackQueueSize = 10000
 )
 
 var (
@@ -113,15 +115,16 @@ type (
 		refreshInterval         dynamicconfig.DurationPropertyFn
 		namespaceStateChangedFn namespace.NamespaceStateChangedFn
 
-		// nsMapsLock protects nameToID, idToNamespace, and stateChangedDuringReadthrough
+		// nsMapsLock protects nameToID and idToNamespace
 		nsMapsLock    sync.RWMutex
 		nameToID      map[namespace.Name]namespace.ID
 		idToNamespace map[namespace.ID]*namespace.Namespace
 
 		// stateChangeCallbacks is a sync.Map so that it can be used without contending for the lock protecting the cache maps; we don't
 		// need to block namespace operations while running or updating callbacks.
-		stateChangeCallbacks          sync.Map // map[any]StateChangeCallbackFn
-		stateChangedDuringReadthrough []*namespace.Namespace
+		stateChangeCallbacks sync.Map // map[any]StateChangeCallbackFn
+		callbackQueue        chan callbackEvent
+		callbackWorker       *goro.Handle
 
 		// readthroughLock protects readthroughNotFoundCache and requests to persistence
 		// it should be acquired before checking readthroughNotFoundCache, making a request
@@ -143,6 +146,11 @@ type (
 		watchCtx    context.Context
 		watchCancel context.CancelFunc
 	}
+
+	callbackEvent struct {
+		ns       *namespace.Namespace
+		isDelete bool
+	}
 )
 
 var _ namespace.Registry = (*registry)(nil)
@@ -171,6 +179,7 @@ func NewRegistry(
 		idToNamespace:            make(map[namespace.ID]*namespace.Namespace),
 		refreshInterval:          refreshInterval,
 		readthroughNotFoundCache: cache.New(readthroughCacheSize, &readthroughNotFoundCacheOpts),
+		callbackQueue:            make(chan callbackEvent, callbackQueueSize),
 
 		forceSearchAttributesCacheRefreshOnRead: forceSearchAttributesCacheRefreshOnRead,
 		replicationResolverFactory:              replicationResolverFactory,
@@ -204,7 +213,9 @@ func (r *registry) RefreshNamespaceById(id namespace.ID) (*namespace.Namespace,
 	if err != nil {
 		return nil, err
 	}
-	r.updateSingleNamespace(ns, false)
+	if r.updateSingleNamespace(ns) {
+		r.callbackQueue <- callbackEvent{ns: ns, isDelete: false}
+	}
 	return ns, nil
 }
 
@@ -220,6 +231,8 @@ func (r *registry) Start() {
 		headers.SystemBackgroundHighCallerInfo,
 	)
 
+	r.callbackWorker = goro.NewHandle(ctx).Go(r.runCallbackWorker)
+
 	watchStarted := false
 	r.refresher, watchStarted = r.runWatchLoop(ctx)
 	if watchStarted {
@@ -250,6 +263,28 @@ func (r *registry) Stop() {
 		r.refresher.Cancel()
 		<-r.refresher.Done()
 	}
+
+	if r.callbackWorker != nil {
+		r.callbackWorker.Cancel()
+		<-r.callbackWorker.Done()
+	}
+}
+
+func (r *registry) runCallbackWorker(ctx context.Context) error {
+	for {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case event := <-r.callbackQueue:
+			r.stateChangeCallbacks.Range(
+				func(key, value any) bool {
+					//revive:disable-next-line:unchecked-type-assertion
+					cb := value.(namespace.StateChangeCallbackFn)
+					cb(event.ns, event.isDelete)
+					return true
+				})
+		}
+	}
 }
 
 func (r *registry) GetPingChecks() []pingable.Check {
@@ -627,41 +662,30 @@ func (r *registry) refreshNamespaces(ctx context.Context) (err error) {
 	totalNamespaceCount := len(newIDToNamespace) // record metric value within lock boundary
 	r.idToNamespace = newIDToNamespace
 	r.nameToID = newNameToID
-	stateChanged = append(stateChanged, r.stateChangedDuringReadthrough...)
-	r.stateChangedDuringReadthrough = nil
 	r.nsMapsLock.Unlock()
 
 	metrics.TotalNamespaces.With(r.metricsHandler).Record(float64(totalNamespaceCount))
 
-	r.stateChangeCallbacks.Range(
-		func(_, value any) bool {
-			//revive:disable-next-line:unchecked-type-assertion
-			cb := value.(namespace.StateChangeCallbackFn)
-
-			for _, ns := range deletedEntries {
-				cb(ns, true)
-			}
-			for _, ns := range stateChanged {
-				cb(ns, false)
-			}
-
-			return true
-		})
+	for _, ns := range deletedEntries {
+		r.callbackQueue <- callbackEvent{ns: ns, isDelete: true}
+	}
+	for _, ns := range stateChanged {
+		r.callbackQueue <- callbackEvent{ns: ns, isDelete: false}
+	}
 
 	return nil
 }
 
 // processWatchEvent handles a single namespace watch event by updating the cache
 // and invoking state change callbacks.
 func (r *registry) processWatchEvent(event *persistence.NamespaceWatchEvent) error {
-	var executeCallbacks bool
-	var ns *namespace.Namespace
+	var changedNS *namespace.Namespace
+	var isDelete bool
 
 	switch event.Type {
 	case persistence.NamespaceWatchEventTypeCreate, persistence.NamespaceWatchEventTypeUpdate:
 		// event.Response is assumed non-nil here; persistence implementations must ensure this.
-		var err error
-		ns, err = namespace.FromPersistentState(
+		ns, err := namespace.FromPersistentState(
 			event.Response.Namespace,
 			r.replicationResolverFactory(event.Response.Namespace),
 			namespace.WithGlobalFlag(event.Response.IsGlobalNamespace),
@@ -670,27 +694,24 @@ func (r *registry) processWatchEvent(event *persistence.NamespaceWatchEvent) err
 		if err != nil {
 			return err
 		}
-		executeCallbacks = r.updateSingleNamespace(ns, true)
+		if r.updateSingleNamespace(ns) {
+			changedNS = ns
+			isDelete = false
+		}
 	case persistence.NamespaceWatchEventTypeDelete:
-		ns = r.deleteNamespace(event.NamespaceID)
-		executeCallbacks = ns != nil
+		if ns := r.deleteNamespace(event.NamespaceID); ns != nil {
+			changedNS = ns
+			isDelete = true
+		}
 	default:
 		r.logger.Warn("Unknown namespace watch event type", tag.Int("eventType", int(event.Type)))
 	}
 
 	idCount, _ := r.GetRegistrySize()
 	metrics.TotalNamespaces.With(r.metricsHandler).Record(float64(idCount))
 
-	if executeCallbacks {
-		isDelete := event.Type == persistence.NamespaceWatchEventTypeDelete
-
-		r.stateChangeCallbacks.Range(
-			func(key, value any) bool {
-				//revive:disable-next-line:unchecked-type-assertion
-				cb := value.(namespace.StateChangeCallbackFn)
-				cb(ns, isDelete)
-				return true
-			})
+	if changedNS != nil {
+		r.callbackQueue <- callbackEvent{ns: changedNS, isDelete: isDelete}
 	}
 
 	return nil
@@ -773,7 +794,9 @@ func (r *registry) getOrReadthroughNamespace(name namespace.Name) (*namespace.Na
 	}
 
 	// update main entry if found
-	r.updateSingleNamespace(ns, false)
+	if r.updateSingleNamespace(ns) {
+		r.callbackQueue <- callbackEvent{ns: ns, isDelete: false}
+	}
 
 	return ns, nil
 }
@@ -808,16 +831,16 @@ func (r *registry) getOrReadthroughNamespaceByID(id namespace.ID) (*namespace.Na
 	}
 
 	// update main entry if found
-	r.updateSingleNamespace(ns, false)
+	if r.updateSingleNamespace(ns) {
+		r.callbackQueue <- callbackEvent{ns: ns, isDelete: false}
+	}
 
 	return ns, nil
 }
 
 // updateSingleNamespace updates the cache with a namespace if it's newer than what we have.
 // Returns true if the namespace state changed.
-// When updatedViaWatch is true, we skip adding to stateChangedDuringReadthrough since watch events
-// trigger callbacks immediately and don't need to be queued for later delivery.
-func (r *registry) updateSingleNamespace(ns *namespace.Namespace, updatedViaWatch bool) bool {
+func (r *registry) updateSingleNamespace(ns *namespace.Namespace) bool {
 	r.nsMapsLock.Lock()
 	defer r.nsMapsLock.Unlock()
 
@@ -835,12 +858,7 @@ func (r *registry) updateSingleNamespace(ns *namespace.Namespace, updatedViaWatc
 	}
 	r.nameToID[ns.Name()] = ns.ID()
 
-	changed := r.namespaceStateChanged(oldNS, ns)
-	if changed && !updatedViaWatch {
-		r.stateChangedDuringReadthrough = append(r.stateChangedDuringReadthrough, ns)
-	}
-
-	return changed
+	return r.namespaceStateChanged(oldNS, ns)
 }
 
 func (r *registry) getNamespaceByNamePersistence(name namespace.Name) (*namespace.Namespace, error) {

common/namespace/nsregistry/registry_test.go

@@ -283,14 +283,32 @@ func (s *registrySuite) TestRegisterStateChangeCallback_CatchUp() {
 	defer s.registry.Stop()
 
 	var entriesNotification []*namespace.Namespace
+	var entriesLock sync.Mutex
+
+	wg := &sync.WaitGroup{}
+	wg.Add(2)
+
 	s.registry.RegisterStateChangeCallback(
 		"0",
 		func(ns *namespace.Namespace, deletedFromDb bool) {
+			entriesLock.Lock()
+			defer entriesLock.Unlock()
 			s.False(deletedFromDb)
+			// Deduplicate
+			for _, e := range entriesNotification {
+				if e.ID() == ns.ID() && e.NotificationVersion() == ns.NotificationVersion() {
+					return
+				}
+			}
 			entriesNotification = append(entriesNotification, ns)
+			wg.Done()
 		},
 	)
 
+	wg.Wait()
+
+	entriesLock.Lock()
+	defer entriesLock.Unlock()
 	s.Len(entriesNotification, 2)
 	if entriesNotification[0].NotificationVersion() > entriesNotification[1].NotificationVersion() {
 		entriesNotification[0], entriesNotification[1] = entriesNotification[1], entriesNotification[0]
@@ -445,23 +463,34 @@ func (s *registrySuite) TestUpdateCache_TriggerCallBack() {
 	defer s.registry.Stop()
 
 	var entries []*namespace.Namespace
+	var entriesLock sync.Mutex
 
 	wg := &sync.WaitGroup{}
 	wg.Add(2)
 	s.registry.RegisterStateChangeCallback(
 		"0",
 		func(ns *namespace.Namespace, deletedFromDb bool) {
-			defer wg.Done()
+			entriesLock.Lock()
+			defer entriesLock.Unlock()
 			s.False(deletedFromDb)
+			// Deduplicate
+			for _, e := range entries {
+				if e.ID() == ns.ID() && e.NotificationVersion() == ns.NotificationVersion() {
+					return
+				}
+			}
 			entries = append(entries, ns)
+			wg.Done()
 		},
 	)
 	wg.Wait()
 
+	entriesLock.Lock()
 	s.Len(entries, 2)
 	if entries[0].NotificationVersion() > entries[1].NotificationVersion() {
 		entries[0], entries[1] = entries[1], entries[0]
 	}
+	entriesLock.Unlock()
 	// Compare by ID and key properties instead of pointer equality
 	s.Equal(entry1Old.ID(), entries[0].ID())
 	s.Equal(entry1Old.Name(), entries[0].Name())
@@ -636,11 +665,14 @@ func (s *registrySuite) TestRemoveDeletedNamespace() {
 	// use WaitGroup and callback to wait until refresh loop picks up delete
 	wg := &sync.WaitGroup{}
 	wg.Add(1)
+	var doneOnce sync.Once
 	s.registry.RegisterStateChangeCallback(
 		"1",
 		func(ns *namespace.Namespace, deletedFromDb bool) {
 			if deletedFromDb {
-				wg.Done()
+				doneOnce.Do(func() {
+					wg.Done()
+				})
 			}
 		},
 	)
@@ -842,9 +874,12 @@ func (s *registrySuite) TestNamespaceRename() {
 	defer s.registry.Stop()
 	// Register callback to detect when the rename is applied
 	refreshCompletedCh := make(chan struct{})
+	var closeOnce sync.Once
 	s.registry.RegisterStateChangeCallback("test", func(ns *namespace.Namespace, deletedFromDb bool) {
 		if ns.Name() == "renamed-name" {
-			close(refreshCompletedCh)
+			closeOnce.Do(func() {
+				close(refreshCompletedCh)
+			})
 		}
 	})
 

common/namespace/nsregistry/registry_watch_test.go

@@ -1061,7 +1061,14 @@ func (s *registryWatchSuite) TestWatchEmptyInitialRefresh() {
 	ns, err := s.registry.GetNamespace("new-namespace")
 	s.NoError(err)
 	s.Equal(nsID, ns.ID())
-	s.InEpsilon(float64(1), s.capture.Snapshot()[metrics.TotalNamespaces.Name()][1].Value, 0.01)
+	s.Eventually(func() bool {
+		snapshots := s.capture.Snapshot()[metrics.TotalNamespaces.Name()]
+		if len(snapshots) >= 2 {
+			s.InEpsilon(float64(1), snapshots[len(snapshots)-1].Value, 0.01)
+			return true
+		}
+		return false
+	}, 2*time.Second, 10*time.Millisecond)
 }
 
 // TestWatchUpdateForUnknownNamespace verifies that an update event for a namespace