diff --git a/.github/workflows/claude-review-teams.yml b/.github/workflows/claude-review-teams.yml index 751fc7068ed..a1a9ee3f8ad 100644 --- a/.github/workflows/claude-review-teams.yml +++ b/.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 }} diff --git a/common/namespace/nsregistry/registry.go b/common/namespace/nsregistry/registry.go index 19651cc90c4..76f1af18688 100644 --- a/common/namespace/nsregistry/registry.go +++ b/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,26 +662,16 @@ 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 } @@ -654,14 +679,13 @@ func (r *registry) refreshNamespaces(ctx context.Context) (err error) { // 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,10 +694,15 @@ 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))) } @@ -681,16 +710,8 @@ func (r *registry) processWatchEvent(event *persistence.NamespaceWatchEvent) err 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) { diff --git a/common/namespace/nsregistry/registry_test.go b/common/namespace/nsregistry/registry_test.go index ad42f508ad4..487c8b2c6e3 100644 --- a/common/namespace/nsregistry/registry_test.go +++ b/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) + }) } }) diff --git a/common/namespace/nsregistry/registry_watch_test.go b/common/namespace/nsregistry/registry_watch_test.go index 470caab14b0..0dc4052c397 100644 --- a/common/namespace/nsregistry/registry_watch_test.go +++ b/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