Dynamic Pubsub - Add synchronizer implementation #5109

liorsve · 2025-12-31T14:39:26Z

This commit's main goal -

Add the actual pubsub synchronizer implementation.
remove the existing pubsub logic (using setup_connection, passed through the cluster_params).
Unify tests - refactored all existing pubsub tests in python to also check dynamic subscription and added more test cases in python. Also added rust tests to verify behavior during slot migration.

In addition, the following is updated/fixed -

Changed subscription command request types - SUBSCRIBE_LAZY becomes SUBSCRIBE (non blocking variation), SUBSCRIBE becomes SUBSCRIBE_BLOCKING (blocking variation)
Moved synchronizer applier setting to the constructor (no 2 step initialization as before).
Synchronizer Trait changes - removed set_initial_subscriptions, was not needed.
Added a python --mock-pubsub flag to note tests that require real server connection (like those doing CLIENT KILL)
mock changes - added other pubsub command support.
Fixed Node tests cleanup - some node tests rely solely on unsubscription during cleanup for test isolation. Fixed the cleanup to ensure unsubscription between tests (formerly they were using custom-command UNSUBSCRIBE, which is now decided to be lazy).

Issue link

This Pull Request is linked to issue: #5098 #4919

Checklist

Before submitting the PR make sure the following are checked:

This Pull Request is related to one issue.
Commit message has a detailed description of what changed and why.
Tests are added or updated.
CHANGELOG.md and documentation files are updated.
Destination branch is correct - main or release
Create merge commit if merging release branch into main, squash otherwise.

Signed-off-by: Lior Sventitzky <[email protected]>

… fixed lint Signed-off-by: Lior Sventitzky <[email protected]>

Signed-off-by: Lior Sventitzky <[email protected]>

glide-core/src/pubsub/synchronizer.rs

Signed-off-by: Lior Sventitzky <[email protected]>

ikolomi · 2026-01-06T10:19:05Z

glide-core/redis-rs/redis/src/cluster_async/mod.rs

-                let mut params: ClusterParams = params.clone();
+                let params: ClusterParams = params.clone();
                let glide_connection_options = glide_connection_options.clone();
                // set subscriptions to none, they will be applied upon the topology discovery


remove comment

ikolomi · 2026-01-06T10:31:53Z

glide-core/redis-rs/redis/src/cluster_async/mod.rs

        let connections = stream::iter(initial_nodes.iter().cloned())
            .map(|(node_addr, socket_addr)| {
-                let mut params: ClusterParams = params.clone();
+                let params: ClusterParams = params.clone();


lets ensure we are reverting all the unnesesary mods that were introduced in the original pubsub impl
amazon-contributing/redis-rs@b36c959...cb81fb7#diff-5ddfd8b2718c21504d43c50b0490bc22de4189208d97f5604a2297124e529577

ikolomi · 2026-01-06T11:09:39Z

glide-core/redis-rs/redis/src/cluster_async/mod.rs

            .await;
        }
        in_progress.store(false, Ordering::Relaxed);
-


we need to be able to update the active subscriptions both on the pasive and active disconnect. The primary candidate for sourcing event for subsctiption removal is the desctructor of MultiplexedConnection (only user type)

We need also to pay attention to the order of the events. e.i. when we remove the active subscription and when we add it

Before that, we need to ensure that the current implementation with the passive notifier suffice for this situation:

We remove node from the connection map BUT reconcile is not triggered

I moved it to PipelineSink::drop (not MultiplexedConnection::drop - it's clonable, it's possible that one would drop but the TCP would remain alive on other clones). I fixed the management connection to never be set with the synchronizer, so drop there won't trigger subscription cleanup. And I removed the cleanup for both cluster and standalone, since this covers both. Since PiplelineSink struct is defined in a pin_project! macro, I had to put it inside with impl<T> PinnedDrop for PipelineSink<T>.

I tried to trace whether there are places where connections are actively dropped without passing through trigger_refresh_connection_tasks (the previous location of remove_subscriptions_for_addresses, always triggered on passive disconnect bc of the disconnect notifier):

refresh_slots_inner - Replaces the entire ConnectionsContainer, which drops old connections (ones that aren't responsible for slots anymore). However, it already calls handle_topology_refresh for cleanup.

Client shutdown - But we don't particularly care about cleanup in this case because the synchronizer is dropped anyway.

So in theory we'd be covered even if we kept it in trigger_refresh_connection_tasks.
Why move it to PipelineSink::drop anyway?

No inconsistency window - With the previous approach, there's a gap between when the connection dies and when validate_all_user_connections detects is_closed(). During this window, our state thinks we're subscribed but we're not.

Cleanup is tied to actual connection lifetime, not to explicit refresh calls.

Future-proof - Any new code path that drops a connection will automatically get cleanup, without needing to remember to add the call.

ikolomi · 2026-01-06T12:09:13Z

glide-core/redis-rs/redis/src/pubsub_synchronizer.rs

+    /// Wait for initial subscriptions (set via config) to be synchronized.
+    /// Returns Ok(()) when all desired subscriptions are established, or error on timeout.
+    /// Default implementation returns immediately (no-op for clients without initial subscriptions).
+    async fn wait_for_initial_sync(&self, _timeout_ms: u64) -> RedisResult<()> {


maybe not to have it specificaly for "initial" subs?

ikolomi · 2026-01-06T12:13:15Z

glide-core/src/client/mod.rs

+        // to channels provided via config. Keeping the guard would cause a deadlock. We wait
+        // for the subscription here to ensure the lazy client is subscribed immediately upon creation.
+        drop(guard);
+        if let Err(e) = self.pubsub_synchronizer.wait_for_initial_sync(1000).await {


why 1000? I think it should use the budget of connection timeout - verify me

ikolomi · 2026-01-06T12:24:11Z

glide-core/src/client/mod.rs


+            if !is_lazy {
+                pubsub_synchronizer.trigger_reconciliation();
+                if let Err(e) = pubsub_synchronizer.wait_for_initial_sync(2000).await {


why not blocking?

ikolomi · 2026-01-06T12:58:50Z

glide-core/src/client/standalone_client.rs


        let tls_mode = connection_request.tls_mode;
        let node_count = connection_request.addresses.len();
-        // randomize pubsub nodes, maybe a batter option is to always use the primary


two issues, one potential and one probable:

probable - We use default route for sub/unsub commands, which is Primary. In case of a role change, the primary will change, sending unsubscribe command on a node on which we did not subscribe. A test requried to show this.

potential - double subscribe. In case the guard desired_subs == actual_subs fails, we might to subscribe again on a new primary

ikolomi · 2026-01-06T13:04:15Z

glide-core/src/pubsub/synchronizer.rs

-/// Real implementation of PubSub synchronizer (stub for now)
-#[allow(dead_code)]
+const LOCK_ERR: &str = "Lock poisoned";
+const RECONCILIATION_INTERVAL: Duration = Duration::from_secs(5);


We dont know how it will behave, lets expose it in the client params.

The cost should be low, why not to have it each 3 second?

ikolomi · 2026-01-06T13:05:19Z

glide-core/src/pubsub/synchronizer.rs

 }

 impl GlidePubSubSynchronizer {
    pub async fn create(


why create and not new

ikolomi · 2026-01-06T13:22:14Z

glide-core/src/pubsub/synchronizer.rs

+
+    /// Single pass: aggregate current subscriptions, compare with desired, return both + sync status
+    fn compute_sync_state(&self) -> SyncCheckResult {
+        let desired = self.desired_subscriptions.read().expect(LOCK_ERR).clone();


lets try to compute the diffs here with minimal passes

ikolomi · 2026-01-06T13:26:54Z

glide-core/src/pubsub/synchronizer.rs

+        }
+
+        let _ = GlideOpenTelemetry::record_subscription_out_of_sync();
+        log_info(


not info but warn, also desired (and by extenstion actual) can get very large

remove altogether

ikolomi · 2026-01-06T13:29:50Z

glide-core/src/pubsub/synchronizer.rs

+        );
+    }
+
+    async fn send_command(


apply_pubsub

ikolomi · 2026-01-06T13:35:21Z

glide-core/src/pubsub/synchronizer.rs

+    }
+
+    /// Parse an address string into SingleNodeRoutingInfo::ByAddress
+    fn parse_address_to_routing(address: &str) -> RedisResult<SingleNodeRoutingInfo> {


check if we can key by ByAddress or long long

dont we have an util func that does parse_address_to_routing ?

ikolomi · 2026-01-06T13:36:44Z

glide-core/src/pubsub/synchronizer.rs

+    fn start_reconciliation_task(self: &Arc<Self>) {
+        // We use a weak ref to self here as an indication that the client has dropped and thus there
+        // is no longer a strong ref of the synchornizer and we should break from the loop
+        let sync_weak = Arc::downgrade(self);


ikolomi · 2026-01-06T13:45:33Z

glide-core/src/pubsub/synchronizer.rs

+        };
+        match self.send_command(&mut cmd, routing).await {
+            Ok(_) => {
+                log_debug(


please go over all new logs

consider removing

consider making them cheap

ikolomi · 2026-01-06T13:47:42Z

glide-core/src/pubsub/synchronizer.rs

+            });
+
+        for (slot, slot_channels) in channels_by_slot {
+            log_debug(


ikolomi · 2026-01-06T13:48:31Z

glide-core/src/pubsub/synchronizer.rs

+        }
+    }
+
+    fn format_subscription_info(info: &PubSubSubscriptionInfo) -> String {


ikolomi · 2026-01-06T14:08:19Z

glide-core/src/pubsub/synchronizer.rs


-    fn remove_current_subscriptions_for_addresses(&self, _addresses: &HashSet<String>) {
-        // TODO: Implement
+    async fn wait_for_initial_sync(&self, timeout_ms: u64) -> RedisResult<()> {


ikolomi

approving pending comments

jduo · 2026-01-06T18:31:08Z

glide-core/redis-rs/redis/src/push_manager.rs

        data: &[Value],
        address: Option<String>,
    ) {
+        logger_core::log_debug(


This will execute all the formatting logic, string conversion, etc. even if logging is off. This happens throughout the core, not specifically in this PR. We should really add logging APIs that take in functions instead of raw strings.

Signed-off-by: Lior Sventitzky <[email protected]>

liorsve added 5 commits December 31, 2025 06:47

fixed the client weak ref to be of ClientWrapper

79daf36

Signed-off-by: Lior Sventitzky <[email protected]>

fixed some tests, added many channels tests

749aed2

Signed-off-by: Lior Sventitzky <[email protected]>

added synchronizer, removed existing pubsub logic, added python tests

45548d3

Signed-off-by: Lior Sventitzky <[email protected]>

added topology tests, removed set_synchronizer_internal_client

e3fa7e3

Signed-off-by: Lior Sventitzky <[email protected]>

refactored tests, added wait_for_initial_sync, made notify non-arc

1d1b717

Signed-off-by: Lior Sventitzky <[email protected]>

liorsve force-pushed the add-synchonizer-impl branch 2 times, most recently from 5a80756 to ef23149 Compare December 31, 2025 15:30

removed leftover pubsub_subscription from cluster_params

3c7b7c3

Signed-off-by: Lior Sventitzky <[email protected]>

liorsve force-pushed the add-synchonizer-impl branch 9 times, most recently from 1bab7bf to bfa1e4f Compare January 1, 2026 16:24

fixed node cleanup, added reconcilliation in wait_for_state function,…

5d6887e

… fixed lint Signed-off-by: Lior Sventitzky <[email protected]>

liorsve force-pushed the add-synchonizer-impl branch from bfa1e4f to 5d6887e Compare January 1, 2026 19:03

yipin-chen requested review from alexr-bq and jduo January 2, 2026 23:50

liorsve marked this pull request as ready for review January 4, 2026 08:06

liorsve requested a review from a team as a code owner January 4, 2026 08:06

liorsve force-pushed the add-synchonizer-impl branch from f48d8b6 to c5076d0 Compare January 4, 2026 12:11

final tweaks

93b0d91

Signed-off-by: Lior Sventitzky <[email protected]>

liorsve force-pushed the add-synchonizer-impl branch from c5076d0 to 93b0d91 Compare January 4, 2026 12:33

liorsve commented Jan 5, 2026

View reviewed changes

glide-core/src/pubsub/synchronizer.rs Show resolved Hide resolved

added custom command test

f50f72a

Signed-off-by: Lior Sventitzky <[email protected]>

liorsve force-pushed the add-synchonizer-impl branch from 33576e5 to f50f72a Compare January 6, 2026 09:39

ikolomi reviewed Jan 6, 2026

View reviewed changes

glide-core/src/pubsub/synchronizer.rs

}

impl GlidePubSubSynchronizer {

pub async fn create(

Copy link

Collaborator

ikolomi Jan 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why create and not new

ikolomi reviewed Jan 6, 2026

View reviewed changes

glide-core/src/pubsub/synchronizer.rs

);

}

async fn send_command(

Copy link

Collaborator

ikolomi Jan 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

apply_pubsub

ikolomi reviewed Jan 6, 2026

View reviewed changes

glide-core/src/pubsub/synchronizer.rs

});

for (slot, slot_channels) in channels_by_slot {

log_debug(

Copy link

Collaborator

ikolomi Jan 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

same

ikolomi reviewed Jan 6, 2026

View reviewed changes

glide-core/src/pubsub/synchronizer.rs

}

}

fn format_subscription_info(info: &PubSubSubscriptionInfo) -> String {

Copy link

Collaborator

ikolomi Jan 6, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

?

ikolomi reviewed Jan 6, 2026

View reviewed changes

ikolomi approved these changes Jan 6, 2026

View reviewed changes

jduo reviewed Jan 6, 2026

View reviewed changes

removed remove subscription on disconnect to drop pipelinesink

ed4a267

Signed-off-by: Lior Sventitzky <[email protected]>

Dynamic Pubsub - Add synchronizer implementation #5109

Are you sure you want to change the base?

Dynamic Pubsub - Add synchronizer implementation #5109

Uh oh!

Conversation

liorsve commented Dec 31, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Issue link

Checklist

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

liorsve Jan 7, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ikolomi left a comment

Choose a reason for hiding this comment

Uh oh!

jduo Jan 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

liorsve commented Dec 31, 2025 •

edited

Loading

liorsve Jan 7, 2026 •

edited

Loading

jduo Jan 6, 2026 •

edited

Loading