diff --git a/redis/_parsers/base.py b/redis/_parsers/base.py
index 63bf5d7795..2da7a94b3d 100644
--- a/redis/_parsers/base.py
+++ b/redis/_parsers/base.py
@@ -11,6 +11,8 @@
     NodeMigratedNotification,
     NodeMigratingNotification,
     NodeMovingNotification,
+    OSSNodeMigratedNotification,
+    OSSNodeMigratingNotification,
 )
 
 if sys.version_info.major >= 3 and sys.version_info.minor >= 11:
@@ -179,9 +181,29 @@ 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:
+        # SMIGRATING <seq_number> <slot, range1-range2,...>
+        id = response[1]
+        slots = response[2]
+        return OSSNodeMigratingNotification(id, slots)
+
+    @staticmethod
+    def parse_oss_maintenance_completed_msg(response):
+        # Expected message format is:
+        # SMIGRATED <seq_number> <host:port> <slot, range1-range2,...>
+        id = response[1]
+        node_address = response[2]
+        slots = response[3]
+        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)
@@ -189,6 +211,9 @@ def parse_maintenance_start_msg(response, notification_type):
     @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)
 
@@ -215,12 +240,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[
@@ -246,6 +274,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,
+    ),
 }
 
 
@@ -256,6 +292,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"""
@@ -270,6 +307,7 @@ def handle_push_response(self, response, **kwargs):
             _INVALIDATION_MESSAGE,
             *_MAINTENANCE_MESSAGES,
             _MOVING_MESSAGE,
+            _SMIGRATED_MESSAGE,
         ):
             return self.pubsub_push_handler_func(response)
 
@@ -292,13 +330,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)
@@ -318,6 +370,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"""
@@ -326,6 +381,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"""
@@ -342,6 +398,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)
 
@@ -366,13 +423,26 @@ 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)
@@ -394,6 +464,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"""
diff --git a/redis/_parsers/hiredis.py b/redis/_parsers/hiredis.py
index d82fe99cd9..11b80c2cb6 100644
--- a/redis/_parsers/hiredis.py
+++ b/redis/_parsers/hiredis.py
@@ -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
 
diff --git a/redis/_parsers/resp3.py b/redis/_parsers/resp3.py
index 72957b464c..7d1e315b9d 100644
--- a/redis/_parsers/resp3.py
+++ b/redis/_parsers/resp3.py
@@ -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):
diff --git a/redis/maint_notifications.py b/redis/maint_notifications.py
index 5b8b08c1be..dbb97d37b9 100644
--- a/redis/maint_notifications.py
+++ b/redis/maint_notifications.py
@@ -5,10 +5,13 @@
 import threading
 import time
 from abc import ABC, abstractmethod
-from typing import TYPE_CHECKING, Literal, Optional, Union
+from typing import TYPE_CHECKING, List, Literal, Optional, Union
 
 from redis.typing import Number
 
+if TYPE_CHECKING:
+    from redis.cluster import NodesManager
+
 
 class MaintenanceState(enum.Enum):
     NONE = "none"
@@ -394,6 +397,125 @@ def __hash__(self) -> int:
         return hash((self.__class__.__name__, int(self.id)))
 
 
+class OSSNodeMigratingNotification(MaintenanceNotification):
+    """
+    Notification for when a Redis OSS API client is used and a node is in the process of migrating slots.
+
+    This notification is received when a node starts migrating its slots to another node
+    during cluster rebalancing or maintenance operations.
+
+    Args:
+        id (int): Unique identifier for this notification
+        slots (Optional[List[int]]): List of slots being migrated
+    """
+
+    DEFAULT_TTL = 30
+
+    def __init__(
+        self,
+        id: int,
+        slots: Optional[List[int]] = None,
+    ):
+        super().__init__(id, OSSNodeMigratingNotification.DEFAULT_TTL)
+        self.slots = slots
+
+    def __repr__(self) -> str:
+        expiry_time = self.creation_time + self.ttl
+        remaining = max(0, expiry_time - time.monotonic())
+        return (
+            f"{self.__class__.__name__}("
+            f"id={self.id}, "
+            f"slots={self.slots}, "
+            f"ttl={self.ttl}, "
+            f"creation_time={self.creation_time}, "
+            f"expires_at={expiry_time}, "
+            f"remaining={remaining:.1f}s, "
+            f"expired={self.is_expired()}"
+            f")"
+        )
+
+    def __eq__(self, other) -> bool:
+        """
+        Two OSSNodeMigratingNotification notifications are considered equal if they have the same
+        id and are of the same type.
+        """
+        if not isinstance(other, OSSNodeMigratingNotification):
+            return False
+        return self.id == other.id and type(self) is type(other)
+
+    def __hash__(self) -> int:
+        """
+        Return a hash value for the notification to allow
+        instances to be used in sets and as dictionary keys.
+
+        Returns:
+            int: Hash value based on notification type and id
+        """
+        return hash((self.__class__.__name__, int(self.id)))
+
+
+class OSSNodeMigratedNotification(MaintenanceNotification):
+    """
+    Notification for when a Redis OSS API client is used and a node has completed migrating slots.
+
+    This notification is received when a node has finished migrating all its slots
+    to other nodes during cluster rebalancing or maintenance operations.
+
+    Args:
+        id (int): Unique identifier for this notification
+        node_address (Optional[str]): Address of the node that has
+                                      completed migration - this is the destination node.
+        slots (Optional[List[int]]): List of slots that have been migrated
+    """
+
+    DEFAULT_TTL = 30
+
+    def __init__(
+        self,
+        id: int,
+        node_address: Optional[str] = None,
+        slots: Optional[List[int]] = None,
+    ):
+        super().__init__(id, OSSNodeMigratedNotification.DEFAULT_TTL)
+        self.node_address = node_address
+        self.slots = slots
+
+    def __repr__(self) -> str:
+        expiry_time = self.creation_time + self.ttl
+        remaining = max(0, expiry_time - time.monotonic())
+        return (
+            f"{self.__class__.__name__}("
+            f"id={self.id}, "
+            f"node_address={self.node_address}, "
+            f"slots={self.slots}, "
+            f"ttl={self.ttl}, "
+            f"creation_time={self.creation_time}, "
+            f"expires_at={expiry_time}, "
+            f"remaining={remaining:.1f}s, "
+            f"expired={self.is_expired()}"
+            f")"
+        )
+
+    def __eq__(self, other) -> bool:
+        """
+        Two OSSNodeMigratedNotification notifications are considered equal if they have the same
+        id and are of the same type.
+        """
+        if not isinstance(other, OSSNodeMigratedNotification):
+            return False
+        return self.id == other.id and type(self) is type(other)
+
+    def __hash__(self) -> int:
+        """
+        Return a hash value for the notification to allow
+        instances to be used in sets and as dictionary keys.
+
+        Returns:
+            int: Hash value based on notification type and id
+        """
+        return hash((self.__class__.__name__, int(self.id)))
+
+
 def _is_private_fqdn(host: str) -> bool:
     """
     Determine if an FQDN is likely to be internal/private.
@@ -755,6 +877,7 @@ class MaintNotificationsConnectionHandler:
     _NOTIFICATION_TYPES: dict[type["MaintenanceNotification"], int] = {
         NodeMigratingNotification: 1,
         NodeFailingOverNotification: 1,
+        OSSNodeMigratingNotification: 1,
         NodeMigratedNotification: 0,
         NodeFailedOverNotification: 0,
     }
@@ -808,3 +931,31 @@ def handle_maintenance_completed_notification(self):
         # timeouts by providing -1 as the relaxed timeout
         self.connection.update_current_socket_timeout(-1)
         self.connection.maintenance_state = MaintenanceState.NONE
+
+
+class OSSMaintNotificationsHandler:
+    def __init__(
+        self, nodes_manager: "NodesManager", config: MaintNotificationsConfig
+    ) -> None:
+        self.nodes_manager = nodes_manager
+        self.config = config
+        self._processed_notifications = set()
+        self._lock = threading.RLock()
+
+    def remove_expired_notifications(self):
+        with self._lock:
+            for notification in tuple(self._processed_notifications):
+                if notification.is_expired():
+                    self._processed_notifications.remove(notification)
+
+    def handle_notification(self, notification: MaintenanceNotification):
+        if isinstance(notification, OSSNodeMigratedNotification):
+            self.handle_oss_maintenance_completed_notification(notification)
+        else:
+            logging.error(f"Unhandled notification type: {notification}")
+
+    def handle_oss_maintenance_completed_notification(
+        self, notification: OSSNodeMigratedNotification
+    ):
+        self.remove_expired_notifications()
+        logging.info(f"Received OSS maintenance completed notification: {notification}")
diff --git a/tests/maint_notifications/proxy_server_helpers.py b/tests/maint_notifications/proxy_server_helpers.py
index f0276312f9..b3397beefa 100644
--- a/tests/maint_notifications/proxy_server_helpers.py
+++ b/tests/maint_notifications/proxy_server_helpers.py
@@ -1,5 +1,6 @@
 import base64
 import logging
+import re
 from typing import Union
 
 from redis.http.http_client import HttpClient, HttpError
@@ -10,6 +11,19 @@
 class RespTranslator:
     """Helper class to translate between RESP and other encodings."""
 
+    @staticmethod
+    def str_or_list_to_resp(txt: str) -> str:
+        """
+        Convert specific string or list to RESP format.
+        """
+        if re.match(r"^<.*>$", txt):
+            items = txt[1:-1].split(",")
+            return f"*{len(items)}\r\n" + "\r\n".join(
+                f"${len(x)}\r\n{x}" for x in items
+            )
+        else:
+            return f"${len(txt)}\r\n{txt}"
+
     @staticmethod
     def cluster_slots_to_resp(resp: str) -> str:
         """Convert query to RESP format."""
@@ -24,7 +38,9 @@ def smigrating_to_resp(resp: str) -> str:
         """Convert query to RESP format."""
         return (
             f">{len(resp.split())}\r\n"
-            + "\r\n".join(f"${len(x)}\r\n{x}" for x in resp.split())
+            + "\r\n".join(
+                f"{RespTranslator.str_or_list_to_resp(x)}" for x in resp.split()
+            )
             + "\r\n"
         )
 
@@ -118,6 +134,8 @@ def get_stats(self) -> dict:
 
         try:
             response = self.http_client.get(url)
+            if isinstance(response, dict):
+                return response
             return response.json()
 
         except HttpError as e:
@@ -134,6 +152,8 @@ def get_connections(self) -> dict:
 
         try:
             response = self.http_client.get(url)
+            if isinstance(response, dict):
+                return response
             return response.json()
         except HttpError as e:
             raise RuntimeError(f"Failed to get connections: {e}")
@@ -192,7 +212,9 @@ def send_notification(
         data = base64.b64encode(notification.encode("utf-8"))
 
         try:
-            response = self.http_client.post(url, json_body=data)
+            response = self.http_client.post(url, data=data)
+            if isinstance(response, dict):
+                return response
             results = response.json()
         except HttpError as e:
             results = {"error": str(e)}
diff --git a/tests/maint_notifications/test_cluster_maint_notifications_handling.py b/tests/maint_notifications/test_cluster_maint_notifications_handling.py
index 6d81224603..8e8d53b62f 100644
--- a/tests/maint_notifications/test_cluster_maint_notifications_handling.py
+++ b/tests/maint_notifications/test_cluster_maint_notifications_handling.py
@@ -1,50 +1,62 @@
-from typing import cast
+from dataclasses import dataclass
+from typing import List, Optional, cast
 
 from redis import ConnectionPool, RedisCluster
 from redis.cluster import ClusterNode
 from redis.connection import (
     BlockingConnectionPool,
 )
-from redis.maint_notifications import MaintNotificationsConfig
+from redis.maint_notifications import MaintNotificationsConfig, MaintenanceState
 from redis.cache import CacheConfig
 from tests.maint_notifications.proxy_server_helpers import (
     ProxyInterceptorHelper,
     RespTranslator,
 )
 
+NODE_PORT_1 = 15379
+NODE_PORT_2 = 15380
+NODE_PORT_3 = 15381
+
 # Initial cluster node configuration for proxy-based tests
 PROXY_CLUSTER_NODES = [
-    ClusterNode("127.0.0.1", 15379),
-    ClusterNode("127.0.0.1", 15380),
-    ClusterNode("127.0.0.1", 15381),
+    ClusterNode("127.0.0.1", NODE_PORT_1),
+    ClusterNode("127.0.0.1", NODE_PORT_2),
+    ClusterNode("127.0.0.1", NODE_PORT_3),
 ]
 
 
-class TestClusterMaintNotificationsConfig:
-    """Test the maint_notifications_config parameter of RedisCluster."""
+class TestClusterMaintNotificationsBase:
+    """Base class for cluster maintenance notifications handling tests."""
 
-    # Helper methods
     def _create_cluster_client(
         self,
+        pool_class=ConnectionPool,
+        enable_cache=False,
+        max_connections=10,
         maint_config=None,
-        connection_pool_class=None,
-        cache_config=None,
-        skip_full_coverage_check=True,
-    ):
-        """Create a RedisCluster instance with real cluster nodes."""
-        kwargs = {
-            "startup_nodes": PROXY_CLUSTER_NODES,
-            "protocol": 3,
-            "skip_full_coverage_check": skip_full_coverage_check,
-        }
-        if maint_config is not None:
-            kwargs["maint_notifications_config"] = maint_config
-        if connection_pool_class is not None:
-            kwargs["connection_pool_class"] = connection_pool_class
-        if cache_config is not None:
-            kwargs["cache_config"] = cache_config
-
-        return RedisCluster(**kwargs)
+    ) -> RedisCluster:
+        """Create a RedisCluster instance with mocked sockets."""
+        if maint_config is None and hasattr(self, "config") and self.config is not None:
+            maint_config = self.config
+
+        kwargs = {}
+        if enable_cache:
+            kwargs = {"cache_config": CacheConfig()}
+
+        test_redis_client = RedisCluster(
+            protocol=3,
+            startup_nodes=PROXY_CLUSTER_NODES,
+            maint_notifications_config=maint_config,
+            connection_pool_class=pool_class,
+            max_connections=max_connections,
+            **kwargs,
+        )
+
+        return test_redis_client
+
+
+class TestClusterMaintNotificationsConfig(TestClusterMaintNotificationsBase):
+    """Test the maint_notifications_config parameter of RedisCluster."""
 
     def _validate_maint_config_on_nodes_manager(
         self,
@@ -159,7 +171,7 @@ def test_config_with_blocking_connection_pool(self):
 
         cluster = self._create_cluster_client(
             maint_config=maint_config,
-            connection_pool_class=BlockingConnectionPool,
+            pool_class=BlockingConnectionPool,
         )
 
         try:
@@ -183,11 +195,10 @@ def test_config_with_cache_enabled(self):
         maint_config = MaintNotificationsConfig(
             enabled=False, proactive_reconnect=True, relaxed_timeout=15
         )
-        cache_config = CacheConfig()
 
         cluster = self._create_cluster_client(
             maint_config=maint_config,
-            cache_config=cache_config,
+            enable_cache=True,
         )
 
         try:
@@ -274,7 +285,7 @@ def test_config_with_pipeline_operations(self):
             cluster.close()
 
 
-class TestClusterMaintNotificationsHandlingBase:
+class TestClusterMaintNotificationsHandlingBase(TestClusterMaintNotificationsBase):
     """Base class for maintenance notifications handling tests."""
 
     def setup_method(self):
@@ -287,39 +298,128 @@ def setup_method(self):
         )
         self.cluster = self._create_cluster_client(maint_config=self.config)
 
-    def _create_cluster_client(
-        self,
-        pool_class=ConnectionPool,
-        enable_cache=False,
-        max_connections=10,
-        maint_config=None,
-    ) -> RedisCluster:
-        """Create a RedisCluster instance with mocked sockets."""
-        config = maint_config if maint_config is not None else self.config
-        kwargs = {}
-        if enable_cache:
-            kwargs = {"cache_config": CacheConfig()}
-
-        test_redis_client = RedisCluster(
-            protocol=3,
-            startup_nodes=PROXY_CLUSTER_NODES,
-            maint_notifications_config=config,
-            connection_pool_class=pool_class,
-            max_connections=max_connections,
-            **kwargs,
-        )
-
-        return test_redis_client
-
     def teardown_method(self):
         """Clean up test fixtures."""
         self.cluster.close()
         self.proxy_helper.cleanup_interceptors()
 
 
+@dataclass
+class ConnectionStateExpectation:
+    """Data class to hold connection state details for validation."""
+
+    node_port: int
+    changed_connections_count: int = 0
+    state: MaintenanceState = MaintenanceState.NONE
+    relaxed_timeout: Optional[int] = None
+
+
 class TestClusterMaintNotificationsHandling(TestClusterMaintNotificationsHandlingBase):
     """Test maintenance notifications handling with RedisCluster."""
 
+    def _get_expected_node_state(
+        self, expectations_list: List[ConnectionStateExpectation], node_port: int
+    ) -> Optional[ConnectionStateExpectation]:
+        """Get the expected state for a node."""
+        for expectation in expectations_list:
+            if expectation.node_port == node_port:
+                return expectation
+        return None
+
+    def _validate_connections_states(
+        self,
+        cluster: RedisCluster,
+        expected_states: List[ConnectionStateExpectation],
+    ):
+        """Validate connections states."""
+        default_maint_state = MaintenanceState.NONE
+        default_timeout = None
+        nodes = list(cluster.nodes_manager.nodes_cache.values())
+        for node in nodes:
+            cluster_node = cast(ClusterNode, node)
+            assert cluster_node.redis_connection is not None
+            connection_pool = cluster_node.redis_connection.connection_pool
+            assert connection_pool is not None
+            expected_state = self._get_expected_node_state(
+                expected_states, cluster_node.port
+            )
+            if expected_state is None:
+                # No expectation for this node
+                continue
+            changed_connections_count = 0
+            for conn in (
+                *connection_pool._get_in_use_connections(),
+                *connection_pool._get_free_connections(),
+            ):
+                if (
+                    conn.maintenance_state != default_maint_state
+                    and conn.maintenance_state == expected_state.state
+                ) or (
+                    conn.socket_timeout != default_timeout
+                    and conn.socket_timeout == expected_state.relaxed_timeout
+                ):
+                    changed_connections_count += 1
+            assert changed_connections_count == expected_state.changed_connections_count
+
+    def test_receive_oss_maintenance_notification(self):
+        """Test receiving an OSS maintenance notification."""
+        # get three connections from each node
+        for node in self.cluster.nodes_manager.nodes_cache.values():
+            node_connections = []
+            for _ in range(3):
+                node_connections.append(
+                    node.redis_connection.connection_pool.get_connection()
+                )
+            for conn in node_connections:
+                node.redis_connection.connection_pool.release(conn)
+
+            node_connections.clear()
+
+        # send a notification to node 1
+        notification = RespTranslator.smigrating_to_resp(
+            "SMIGRATING 12 TO 127.0.0.1:15380 <123,456,5000-7000>"
+        )
+        self.proxy_helper.send_notification(NODE_PORT_1, notification)
+
+        # validate no timeout is relaxed on any connection
+        self._validate_connections_states(
+            self.cluster,
+            [
+                ConnectionStateExpectation(
+                    node_port=NODE_PORT_1, changed_connections_count=0
+                ),
+                ConnectionStateExpectation(
+                    node_port=NODE_PORT_2, changed_connections_count=0
+                ),
+                ConnectionStateExpectation(
+                    node_port=NODE_PORT_3, changed_connections_count=0
+                ),
+            ],
+        )
+
+        # execute a command that will receive the notification
+        res = self.cluster.set("anyprefix:{3}:k", "VAL")
+        assert res is True
+
+        # validate the timeout was relaxed on just one connection for the node
+        self._validate_connections_states(
+            self.cluster,
+            [
+                ConnectionStateExpectation(
+                    node_port=NODE_PORT_1,
+                    changed_connections_count=1,
+                    state=MaintenanceState.MAINTENANCE,
+                    relaxed_timeout=self.config.relaxed_timeout,
+                ),
+                ConnectionStateExpectation(
+                    node_port=NODE_PORT_2, changed_connections_count=0
+                ),
+                ConnectionStateExpectation(
+                    node_port=NODE_PORT_3, changed_connections_count=0
+                ),
+            ],
+        )
+
     def test_receive_maint_notification(self):
         """Test receiving a maintenance notification."""
         self.cluster.set("test", "VAL")
diff --git a/tests/maint_notifications/test_maint_notifications.py b/tests/maint_notifications/test_maint_notifications.py
index 85aa671390..4ae5ae60a5 100644
--- a/tests/maint_notifications/test_maint_notifications.py
+++ b/tests/maint_notifications/test_maint_notifications.py
@@ -11,6 +11,8 @@
     NodeMigratedNotification,
     NodeFailingOverNotification,
     NodeFailedOverNotification,
+    OSSNodeMigratingNotification,
+    OSSNodeMigratedNotification,
     MaintNotificationsConfig,
     MaintNotificationsPoolHandler,
     MaintNotificationsConnectionHandler,
@@ -381,6 +383,222 @@ def test_equality_and_hash(self):
         assert hash(notification1) != hash(notification3)
 
 
+class TestOSSNodeMigratingNotification:
+    """Test the OSSNodeMigratingNotification class."""
+
+    def test_init_with_defaults(self):
+        """Test OSSNodeMigratingNotification initialization with default values."""
+        with patch("time.monotonic", return_value=1000):
+            notification = OSSNodeMigratingNotification(id=1)
+            assert notification.id == 1
+            assert notification.ttl == OSSNodeMigratingNotification.DEFAULT_TTL
+            assert notification.creation_time == 1000
+            assert notification.slots is None
+
+    def test_init_with_all_parameters(self):
+        """Test OSSNodeMigratingNotification initialization with all parameters."""
+        with patch("time.monotonic", return_value=1000):
+            slots = [1, 2, 3, 4, 5]
+            notification = OSSNodeMigratingNotification(
+                id=1,
+                slots=slots,
+            )
+            assert notification.id == 1
+            assert notification.ttl == OSSNodeMigratingNotification.DEFAULT_TTL
+            assert notification.creation_time == 1000
+            assert notification.slots == slots
+
+    def test_default_ttl(self):
+        """Test that DEFAULT_TTL is used correctly."""
+        assert OSSNodeMigratingNotification.DEFAULT_TTL == 30
+        notification = OSSNodeMigratingNotification(id=1)
+        assert notification.ttl == 30
+
+    def test_repr(self):
+        """Test OSSNodeMigratingNotification string representation."""
+        with patch("time.monotonic", return_value=1000):
+            notification = OSSNodeMigratingNotification(
+                id=1,
+                slots=[1, 2, 3],
+            )
+
+        with patch("time.monotonic", return_value=1005):  # 5 seconds later
+            repr_str = repr(notification)
+            assert "OSSNodeMigratingNotification" in repr_str
+            assert "id=1" in repr_str
+            assert "ttl=30" in repr_str
+            assert "remaining=25.0s" in repr_str
+            assert "expired=False" in repr_str
+
+    def test_equality_same_id_and_type(self):
+        """Test equality for notifications with same id and type."""
+        notification1 = OSSNodeMigratingNotification(
+            id=1,
+            slots=[1, 2, 3],
+        )
+        notification2 = OSSNodeMigratingNotification(
+            id=1,
+            slots=[4, 5, 6],
+        )
+        # Should be equal because id and type are the same
+        assert notification1 == notification2
+
+    def test_equality_different_id(self):
+        """Test inequality for notifications with different id."""
+        notification1 = OSSNodeMigratingNotification(id=1)
+        notification2 = OSSNodeMigratingNotification(id=2)
+        assert notification1 != notification2
+
+    def test_equality_different_type(self):
+        """Test inequality for notifications of different types."""
+        notification1 = OSSNodeMigratingNotification(id=1)
+        notification2 = NodeMigratingNotification(id=1, ttl=30)
+        assert notification1 != notification2
+
+    def test_hash_same_id_and_type(self):
+        """Test hash for notifications with same id and type."""
+        notification1 = OSSNodeMigratingNotification(
+            id=1,
+            slots=[1, 2, 3],
+        )
+        notification2 = OSSNodeMigratingNotification(
+            id=1,
+            slots=[4, 5, 6],
+        )
+        # Should have same hash because id and type are the same
+        assert hash(notification1) == hash(notification2)
+
+    def test_hash_different_id(self):
+        """Test hash for notifications with different id."""
+        notification1 = OSSNodeMigratingNotification(id=1)
+        notification2 = OSSNodeMigratingNotification(id=2)
+        assert hash(notification1) != hash(notification2)
+
+    def test_in_set(self):
+        """Test that notifications can be used in sets."""
+        notification1 = OSSNodeMigratingNotification(id=1)
+        notification2 = OSSNodeMigratingNotification(id=1)
+        notification3 = OSSNodeMigratingNotification(id=2)
+        notification4 = OSSNodeMigratingNotification(id=2)
+
+        notification_set = {notification1, notification2, notification3, notification4}
+        assert (
+            len(notification_set) == 2
+        )  # notification1 and notification2 should be the same
+
+
+class TestOSSNodeMigratedNotification:
+    """Test the OSSNodeMigratedNotification class."""
+
+    def test_init_with_defaults(self):
+        """Test OSSNodeMigratedNotification initialization with default values."""
+        with patch("time.monotonic", return_value=1000):
+            notification = OSSNodeMigratedNotification(id=1)
+            assert notification.id == 1
+            assert notification.ttl == OSSNodeMigratedNotification.DEFAULT_TTL
+            assert notification.creation_time == 1000
+            assert notification.node_address is None
+            assert notification.slots is None
+
+    def test_init_with_all_parameters(self):
+        """Test OSSNodeMigratedNotification initialization with all parameters."""
+        with patch("time.monotonic", return_value=1000):
+            slots = [1, 2, 3, 4, 5]
+            notification = OSSNodeMigratedNotification(
+                id=1,
+                node_address="127.0.0.1:6380",
+                slots=slots,
+            )
+            assert notification.id == 1
+            assert notification.ttl == OSSNodeMigratedNotification.DEFAULT_TTL
+            assert notification.creation_time == 1000
+            assert notification.node_address == "127.0.0.1:6380"
+            assert notification.slots == slots
+
+    def test_default_ttl(self):
+        """Test that DEFAULT_TTL is used correctly."""
+        assert OSSNodeMigratedNotification.DEFAULT_TTL == 30
+        notification = OSSNodeMigratedNotification(id=1)
+        assert notification.ttl == 30
+
+    def test_repr(self):
+        """Test OSSNodeMigratedNotification string representation."""
+        with patch("time.monotonic", return_value=1000):
+            notification = OSSNodeMigratedNotification(
+                id=1,
+                node_address="127.0.0.1:6380",
+                slots=[1, 2, 3],
+            )
+
+        with patch("time.monotonic", return_value=1010):  # 10 seconds later
+            repr_str = repr(notification)
+            assert "OSSNodeMigratedNotification" in repr_str
+            assert "id=1" in repr_str
+            assert "ttl=30" in repr_str
+            assert "remaining=20.0s" in repr_str
+            assert "expired=False" in repr_str
+
+    def test_equality_same_id_and_type(self):
+        """Test equality for notifications with same id and type."""
+        notification1 = OSSNodeMigratedNotification(
+            id=1,
+            node_address="127.0.0.1:6380",
+            slots=[1, 2, 3],
+        )
+        notification2 = OSSNodeMigratedNotification(
+            id=1,
+            node_address="127.0.0.1:6381",
+            slots=[4, 5, 6],
+        )
+        # Should be equal because id and type are the same
+        assert notification1 == notification2
+
+    def test_equality_different_id(self):
+        """Test inequality for notifications with different id."""
+        notification1 = OSSNodeMigratedNotification(id=1)
+        notification2 = OSSNodeMigratedNotification(id=2)
+        assert notification1 != notification2
+
+    def test_equality_different_type(self):
+        """Test inequality for notifications of different types."""
+        notification1 = OSSNodeMigratedNotification(id=1)
+        notification2 = NodeMigratedNotification(id=1)
+        assert notification1 != notification2
+
+    def test_hash_same_id_and_type(self):
+        """Test hash for notifications with same id and type."""
+        notification1 = OSSNodeMigratedNotification(
+            id=1,
+            node_address="127.0.0.1:6380",
+            slots=[1, 2, 3],
+        )
+        notification2 = OSSNodeMigratedNotification(
+            id=1,
+            node_address="127.0.0.1:6381",
+            slots=[4, 5, 6],
+        )
+        # Should have same hash because id and type are the same
+        assert hash(notification1) == hash(notification2)
+
+    def test_hash_different_id(self):
+        """Test hash for notifications with different id."""
+        notification1 = OSSNodeMigratedNotification(id=1)
+        notification2 = OSSNodeMigratedNotification(id=2)
+        assert hash(notification1) != hash(notification2)
+
+    def test_in_set(self):
+        """Test that notifications can be used in sets."""
+        notification1 = OSSNodeMigratedNotification(id=1)
+        notification2 = OSSNodeMigratedNotification(id=1)
+        notification3 = OSSNodeMigratedNotification(id=2)
+        notification4 = OSSNodeMigratedNotification(id=2)
+
+        notification_set = {notification1, notification2, notification3, notification4}
+        assert (
+            len(notification_set) == 2
+        )  # notification1 and notification2 should be the same
+
+
 class TestMaintNotificationsConfig:
     """Test the MaintNotificationsConfig class."""