Repository navigation
feat(actuator): add pause scheduling stream listener and local cache - #912
Merged
Merged
Conversation
There was a problem hiding this comment.
Code Review
This pull request introduces a new pause scheduling mechanism for the Actuator, including a thread-safe in-memory cache to track the pause state and a background listener that synchronizes this state using Spanner and Redis broadcast streams. The review feedback highlights critical safety issues, specifically potential nil pointer dereference panics when querying the pause state repository, and a logical flaw in the cache's timestamp fallback mechanism that could lead to legitimate updates being incorrectly ignored.
Implement background stream listener worker subscribing to Redis broadcast stream and thread-safe local atomic cache for the Pause Scheduling feature. Includes cold-start state synchronization, reconnection resync logic, and unit test coverage. TAG=agy CONV=59b60431-f41c-4f8a-a0e7-8653629dfe3e
andrzejfudro
force-pushed
the
feat/pause-scheduling-actuator-listener
branch
from
September 17, 2026 13:31
58e2798 to
f0c6a86
Compare
andrzejfudro
marked this pull request as ready for review
September 17, 2026 13:33
andrzejfudro
enabled auto-merge
September 17, 2026 13:34
janzak-dev
previously approved these changes
Sep 17, 2026
…nabled flag Address review comments by: 1. Updating Cache.Update to accept a newState State struct. 2. Removing redundant Enabled boolean from Params and Listener, as the feature flag is already enforced at the caller site in prod.go. TAG=agy CONV=59b60431-f41c-4f8a-a0e7-8653629dfe3e
andrzejfudro
force-pushed
the
feat/pause-scheduling-actuator-listener
branch
from
September 17, 2026 15:19
2beba01 to
15adb17
Compare
janzak-dev
approved these changes
Sep 18, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Author note:
This PR operates on the fake implementation of the broadcast instead of real redis-streams-based one. This still remains mostly untested but it is hidden behind the feature flag and will be fully verified once I connect all the parts together.
Summary
Implements the background stream listener worker and thread-safe local cache in Actuator for the Pause Scheduling feature.
Key Changes
cmd/actuator/pausescheduling/cache.go): Thread-safe in-memory cache exposing an atomicState()point-in-time view (IsPaused,UpdatedAt) and anti-thrashingUpdate().cmd/actuator/pausescheduling/listener.go): Background worker subscribing toxmc-pause-stream. Features synchronous cold-start seeding (Init) before task queue worker starts, and backoff retry resync (resyncOnReconnect) on stream disconnections.cmd/actuator/prod.go): Initializespausescheduling.Cacheand startspausescheduling.ListenerduringNewProdwhenFeatureFlags.PauseSchedulingEnabledis active.internal/broadcast/fake.go): AddedLatestOffsetimplementation toFakeClient.Verification
bazel test //cmd/actuator:actuator_test //cmd/actuator/pausescheduling:pausescheduling_test //internal/broadcast:broadcast_test(Passed)bazel test --@io_bazel_rules_go//go/config:race //cmd/actuator/pausescheduling:pausescheduling_test(Passed with 0 races)./.githooks/pre-commitTAG=agy
CONV=59b60431-f41c-4f8a-a0e7-8653629dfe3e