Skip to content

Adding OSS Notification Classes for SMIGRATING and SMIGRATED. Handling of SMIGRATING is completed and covered with tests. #3849

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 3 commits into from
Nov 20, 2025
+631 −65
Merged

Adding OSS Notification Classes for SMIGRATING and SMIGRATED. Handling of SMIGRATING is completed and covered with tests. #3849

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e0847c9
Adding maintenance notifications for OSS API enabled connections
petyaslavova Nov 17, 2025
e3c8b3f
Adding oss maint notifications handler configurations to parsers. Pla…
petyaslavova Nov 17, 2025
9ecae0a
Applying review comments. Finilized the notifications format.
petyaslavova Nov 19, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
111 changes: 103 additions & 8 deletions redis/_parsers/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@
NodeMigratedNotification,
NodeMigratingNotification,
NodeMovingNotification,
OSSNodeMigratedNotification,
OSSNodeMigratingNotification,
)

if sys.version_info.major >= 3 and sys.version_info.minor >= 11:
Expand Down Expand Up @@ -179,16 +181,59 @@ async def read_response(
class MaintenanceNotificationsParser:
"""Protocol defining maintenance push notification parsing functionality"""

@staticmethod
def parse_oss_maintenance_start_msg(response):
# Expected message format is:
# TODO Clarify which format will be the final one
# SMIGRATING <seq_number> <slot, range1-range2,...>
# or
# (using the one below for testing until the server changes are done)
# SMIGRATING <transaction_id> TO <endpoint> <list of slots and slot-ranges> ## received on source endpoint
# SMIGRATING <transaction_id> FROM <list of slots and slot-ranges>. ## received on target endpoint
id = response[1]

address_value = response[3]
if isinstance(address_value, bytes):
address_value = address_value.decode()
if response[2] in ("TO", b"TO"):
dest_node = address_value
src_node = None
else:
dest_node = None
src_node = address_value

slots = response[4]
return OSSNodeMigratingNotification(id, src_node, dest_node, slots)

@staticmethod
def parse_oss_maintenance_completed_msg(response):
# Expected message format is:
# TODO Clarify which format will be the final one
# SMIGRATED <seq_number> NodeA’<host:port> <slot, range1-range2,...>
# or
# SMIGRATED <transaction_id> <list of slots and slot-ranges>
# received on source and target endpoints when migration is over
id = response[1]
node_address = None
slots = response[2]
return OSSNodeMigratedNotification(id, node_address, slots)

@staticmethod
def parse_maintenance_start_msg(response, notification_type):
# Expected message format is: <notification_type> <seq_number> <time>
# Examples:
# MIGRATING 1 10
# FAILING_OVER 2 20
id = response[1]
ttl = response[2]
return notification_type(id, ttl)

@staticmethod
def parse_maintenance_completed_msg(response, notification_type):
# Expected message format is: <notification_type> <seq_number>
# Examples:
# MIGRATED 1
# FAILED_OVER 2
id = response[1]
return notification_type(id)

Expand All @@ -215,12 +260,15 @@ def parse_moving_msg(response):
_MIGRATED_MESSAGE = "MIGRATED"
_FAILING_OVER_MESSAGE = "FAILING_OVER"
_FAILED_OVER_MESSAGE = "FAILED_OVER"
_SMIGRATING_MESSAGE = "SMIGRATING"
_SMIGRATED_MESSAGE = "SMIGRATED"

_MAINTENANCE_MESSAGES = (
_MIGRATING_MESSAGE,
_MIGRATED_MESSAGE,
_FAILING_OVER_MESSAGE,
_FAILED_OVER_MESSAGE,
_SMIGRATING_MESSAGE,
)

MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING: dict[
Expand All @@ -246,6 +294,14 @@ def parse_moving_msg(response):
NodeMovingNotification,
MaintenanceNotificationsParser.parse_moving_msg,
),
_SMIGRATING_MESSAGE: (
OSSNodeMigratingNotification,
MaintenanceNotificationsParser.parse_oss_maintenance_start_msg,
),
_SMIGRATED_MESSAGE: (
OSSNodeMigratedNotification,
MaintenanceNotificationsParser.parse_oss_maintenance_completed_msg,
),
}


Expand All @@ -256,6 +312,7 @@ class PushNotificationsParser(Protocol):
invalidation_push_handler_func: Optional[Callable] = None
node_moving_push_handler_func: Optional[Callable] = None
maintenance_push_handler_func: Optional[Callable] = None
oss_cluster_maint_push_handler_func: Optional[Callable] = None

def handle_pubsub_push_response(self, response):
"""Handle pubsub push responses"""
Expand All @@ -270,6 +327,7 @@ def handle_push_response(self, response, **kwargs):
_INVALIDATION_MESSAGE,
*_MAINTENANCE_MESSAGES,
_MOVING_MESSAGE,
_SMIGRATED_MESSAGE,
):
return self.pubsub_push_handler_func(response)

Expand All @@ -292,13 +350,27 @@ def handle_push_response(self, response, **kwargs):
parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[
msg_type
][1]
notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[
msg_type
][0]
notification = parser_function(response, notification_type)
if msg_type == _SMIGRATING_MESSAGE:
notification = parser_function(response)
else:
notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[
msg_type
][0]
notification = parser_function(response, notification_type)

if notification is not None:
return self.maintenance_push_handler_func(notification)
if (
msg_type == _SMIGRATED_MESSAGE
and self.oss_cluster_maint_push_handler_func
):
parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[
msg_type
][1]
notification = parser_function(response)

if notification is not None:
return self.oss_cluster_maint_push_handler_func(notification)
except Exception as e:
logger.error(
"Error handling {} message ({}): {}".format(msg_type, response, e)
Expand All @@ -318,6 +390,9 @@ def set_node_moving_push_handler(self, node_moving_push_handler_func):
def set_maintenance_push_handler(self, maintenance_push_handler_func):
self.maintenance_push_handler_func = maintenance_push_handler_func

def set_oss_cluster_maint_push_handler(self, oss_cluster_maint_push_handler_func):
self.oss_cluster_maint_push_handler_func = oss_cluster_maint_push_handler_func


class AsyncPushNotificationsParser(Protocol):
"""Protocol defining async RESP3-specific parsing functionality"""
Expand All @@ -326,6 +401,7 @@ class AsyncPushNotificationsParser(Protocol):
invalidation_push_handler_func: Optional[Callable] = None
node_moving_push_handler_func: Optional[Callable[..., Awaitable[None]]] = None
maintenance_push_handler_func: Optional[Callable[..., Awaitable[None]]] = None
oss_cluster_maint_push_handler_func: Optional[Callable[..., Awaitable[None]]] = None

async def handle_pubsub_push_response(self, response):
"""Handle pubsub push responses asynchronously"""
Expand All @@ -342,6 +418,7 @@ async def handle_push_response(self, response, **kwargs):
_INVALIDATION_MESSAGE,
*_MAINTENANCE_MESSAGES,
_MOVING_MESSAGE,
_SMIGRATED_MESSAGE,
):
return await self.pubsub_push_handler_func(response)

Expand All @@ -366,13 +443,28 @@ async def handle_push_response(self, response, **kwargs):
parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[
msg_type
][1]
notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[
msg_type
][0]
notification = parser_function(response, notification_type)
if msg_type == _SMIGRATING_MESSAGE:
notification = parser_function(response)
else:
notification_type = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[
msg_type
][0]
notification = parser_function(response, notification_type)

if notification is not None:
return await self.maintenance_push_handler_func(notification)
if (
msg_type == _SMIGRATED_MESSAGE
and self.oss_cluster_maint_push_handler_func
):
parser_function = MSG_TYPE_TO_MAINT_NOTIFICATION_PARSER_MAPPING[
msg_type
][1]
notification = parser_function(response)
if notification is not None:
return await self.oss_cluster_maint_push_handler_func(
notification
)
except Exception as e:
logger.error(
"Error handling {} message ({}): {}".format(msg_type, response, e)
Expand All @@ -394,6 +486,9 @@ def set_node_moving_push_handler(self, node_moving_push_handler_func):
def set_maintenance_push_handler(self, maintenance_push_handler_func):
self.maintenance_push_handler_func = maintenance_push_handler_func

def set_oss_cluster_maint_push_handler(self, oss_cluster_maint_push_handler_func):
self.oss_cluster_maint_push_handler_func = oss_cluster_maint_push_handler_func


class _AsyncRESPBase(AsyncBaseParser):
"""Base class for async resp parsing"""
Expand Down
1 change: 1 addition & 0 deletions redis/_parsers/hiredis.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@ def __init__(self, socket_read_size):
self.pubsub_push_handler_func = self.handle_pubsub_push_response
self.node_moving_push_handler_func = None
self.maintenance_push_handler_func = None
self.oss_cluster_maint_push_handler_func = None
self.invalidation_push_handler_func = None
self._hiredis_PushNotificationType = None

Expand Down
1 change: 1 addition & 0 deletions redis/_parsers/resp3.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ def __init__(self, socket_read_size):
self.pubsub_push_handler_func = self.handle_pubsub_push_response
self.node_moving_push_handler_func = None
self.maintenance_push_handler_func = None
self.oss_cluster_maint_push_handler_func = None
self.invalidation_push_handler_func = None

def handle_pubsub_push_response(self, response):
Expand Down
Loading
Loading