Merge pull request #620 from ably/ECO-5456/vcdiff-support

[ECO-5456] feat: add VCDiff support for delta message decoding

Author: ttypic

ably/__init__.py

@@ -5,9 +5,10 @@
 from ably.types.capability import Capability
 from ably.types.channelsubscription import PushChannelSubscription
 from ably.types.device import DeviceDetails
-from ably.types.options import Options
+from ably.types.options import Options, VCDiffDecoder
 from ably.util.crypto import CipherParams
 from ably.util.exceptions import AblyException, AblyAuthException, IncompatibleClientIdException
+from ably.vcdiff.default_vcdiff_decoder import AblyVCDiffDecoder
 
 import logging
 

ably/realtime/realtime_channel.py

@@ -1,13 +1,15 @@
 from __future__ import annotations
+
 import asyncio
 import logging
 from typing import Optional, TYPE_CHECKING, Dict, Any
 from ably.realtime.connection import ConnectionState
-from ably.transport.websockettransport import ProtocolMessageAction
 from ably.rest.channel import Channel, Channels as RestChannels
+from ably.transport.websockettransport import ProtocolMessageAction
 from ably.types.channelstate import ChannelState, ChannelStateChange
 from ably.types.flags import Flag, has_flag
 from ably.types.message import Message
+from ably.types.mixins import DecodingContext
 from ably.util.eventemitter import EventEmitter
 from ably.util.exceptions import AblyException
 from ably.util.helper import Timer, is_callable_or_coroutine
@@ -123,6 +125,11 @@ def __init__(self, realtime: AblyRealtime, name: str, channel_options: Optional[
         self.__channel_options = channel_options or ChannelOptions()
         self.__params: Optional[Dict[str, str]] = None
 
+        # Delta-specific fields for RTL19/RTL20 compliance
+        vcdiff_decoder = self.__realtime.options.vcdiff_decoder if self.__realtime.options.vcdiff_decoder else None
+        self.__decoding_context = DecodingContext(vcdiff_decoder=vcdiff_decoder)
+        self.__decode_failure_recovery_in_progress = False
+
         # Used to listen to state changes internally, if we use the public event emitter interface then internals
         # will be disrupted if the user called .off() to remove all listeners
         self.__internal_state_emitter = EventEmitter()
@@ -415,8 +422,16 @@ def _on_message(self, proto_msg: dict) -> None:
             else:
                 self._request_state(ChannelState.ATTACHING)
         elif action == ProtocolMessageAction.MESSAGE:
-            messages = Message.from_encoded_array(proto_msg.get('messages'))
-            self.__channel_serial = channel_serial
+            messages = []
+            try:
+                messages = Message.from_encoded_array(proto_msg.get('messages'), context=self.__decoding_context)
+                self.__decoding_context.last_message_id = messages[-1].id
+                self.__channel_serial = channel_serial
+            except AblyException as e:
+                if e.code == 40018:  # Delta decode failure - start recovery
+                    self._start_decode_failure_recovery(e)
+                else:
+                    log.error(f"Message processing error {e}. Skip messages {proto_msg.get('messages')}")
             for message in messages:
                 self.__message_emitter._emit(message.name, message)
         elif action == ProtocolMessageAction.ERROR:
@@ -458,6 +473,9 @@ def _notify_state(self, state: ChannelState, reason: Optional[AblyException] = N
         if state in (ChannelState.DETACHED, ChannelState.SUSPENDED, ChannelState.FAILED):
             self.__channel_serial = None
 
+        if state != ChannelState.ATTACHING:
+            self.__decode_failure_recovery_in_progress = False
+
         state_change = ChannelStateChange(self.__state, state, resumed, reason=reason)
 
         self.__state = state
@@ -554,6 +572,24 @@ def params(self) -> Dict[str, str]:
         """Get channel parameters"""
         return self.__params
 
+    def _start_decode_failure_recovery(self, error: AblyException) -> None:
+        """Start RTL18 decode failure recovery procedure"""
+
+        if self.__decode_failure_recovery_in_progress:
+            log.info('VCDiff recovery process already started, skipping')
+            return
+
+        self.__decode_failure_recovery_in_progress = True
+
+        # RTL18a: Log error with code 40018
+        log.error(f'VCDiff decode failure: {error}')
+
+        # RTL18b: Message is already discarded by not processing it
+
+        # RTL18c: Send ATTACH with previous channel serial and transition to ATTACHING
+        self._notify_state(ChannelState.ATTACHING, reason=error)
+        self._check_pending_state()
+
 
 class Channels(RestChannels):
     """Creates and destroys RealtimeChannel objects.

ably/types/message.py

@@ -3,7 +3,7 @@
 import logging
 
 from ably.types.typedbuffer import TypedBuffer
-from ably.types.mixins import EncodeDataMixin
+from ably.types.mixins import EncodeDataMixin, DeltaExtras
 from ably.util.crypto import CipherData
 from ably.util.exceptions import AblyException
 
@@ -178,7 +178,7 @@ def as_dict(self, binary=False):
         return request_body
 
     @staticmethod
-    def from_encoded(obj, cipher=None):
+    def from_encoded(obj, cipher=None, context=None):
         id = obj.get('id')
         name = obj.get('name')
         data = obj.get('data')
@@ -188,7 +188,12 @@ def from_encoded(obj, cipher=None):
         encoding = obj.get('encoding', '')
         extras = obj.get('extras', None)
 
-        decoded_data = Message.decode(data, encoding, cipher)
+        delta_extra = DeltaExtras(extras)
+        if delta_extra.from_id and delta_extra.from_id != context.last_message_id:
+            raise AblyException(f"Delta message decode failure - previous message not available. "
+                                f"Message id = {id}", 400, 40018)
+
+        decoded_data = Message.decode(data, encoding, cipher, context)
 
         return Message(
             id=id,

ably/types/mixins.py

@@ -3,10 +3,29 @@
 import logging
 
 from ably.util.crypto import CipherData
+from ably.util.exceptions import AblyException
 
 
 log = logging.getLogger(__name__)
 
+ENC_VCDIFF = "vcdiff"
+
+
+class DeltaExtras:
+    def __init__(self, extras):
+        self.from_id = None
+        if extras and 'delta' in extras:
+            delta_info = extras['delta']
+            if isinstance(delta_info, dict):
+                self.from_id = delta_info.get('from')
+
+
+class DecodingContext:
+    def __init__(self, base_payload=None, last_message_id=None, vcdiff_decoder=None):
+        self.base_payload = base_payload
+        self.last_message_id = last_message_id
+        self.vcdiff_decoder = vcdiff_decoder
+
 
 class EncodeDataMixin:
 
@@ -25,10 +44,12 @@ def encoding(self, encoding):
             self._encoding_array = encoding.strip('/').split('/')
 
     @staticmethod
-    def decode(data, encoding='', cipher=None):
+    def decode(data, encoding='', cipher=None, context=None):
         encoding = encoding.strip('/')
         encoding_list = encoding.split('/')
 
+        last_payload = data
+
         while encoding_list:
             encoding = encoding_list.pop()
             if not encoding:
@@ -46,10 +67,43 @@ def decode(data, encoding='', cipher=None):
                 if isinstance(data, list) or isinstance(data, dict):
                     continue
                 data = json.loads(data)
-            elif encoding == 'base64' and isinstance(data, bytes):
-                data = bytearray(base64.b64decode(data))
             elif encoding == 'base64':
-                data = bytearray(base64.b64decode(data.encode('utf-8')))
+                data = bytearray(base64.b64decode(data)) if isinstance(data, bytes) \
+                    else bytearray(base64.b64decode(data.encode('utf-8')))
+                if not encoding_list:
+                    last_payload = data
+            elif encoding == ENC_VCDIFF:
+                if not context or not context.vcdiff_decoder:
+                    log.error('Message cannot be decoded as no VCDiff decoder available')
+                    raise AblyException('VCDiff decoder not available', 40019, 40019)
+
+                if not context.base_payload:
+                    log.error('VCDiff decoding requires base payload')
+                    raise AblyException('VCDiff decode failure', 40018, 40018)
+
+                try:
+                    # Convert base payload to bytes if it's a string
+                    base_data = context.base_payload
+                    if isinstance(base_data, str):
+                        base_data = base_data.encode('utf-8')
+                    else:
+                        base_data = bytes(base_data)
+
+                    # Convert delta to bytes if needed
+                    delta_data = data
+                    if isinstance(delta_data, (bytes, bytearray)):
+                        delta_data = bytes(delta_data)
+                    else:
+                        delta_data = str(delta_data).encode('utf-8')
+
+                    # Decode with VCDiff
+                    data = bytearray(context.vcdiff_decoder.decode(delta_data, base_data))
+                    last_payload = data
+
+                except Exception as e:
+                    log.error(f'VCDiff decode failed: {e}')
+                    raise AblyException('VCDiff decode failure', 40018, 40018)
+
             elif encoding.startswith('%s+' % CipherData.ENCODING_ID):
                 if not cipher:
                     log.error('Message cannot be decrypted as the channel is '
@@ -67,9 +121,11 @@ def decode(data, encoding='', cipher=None):
                 encoding_list.append(encoding)
                 break
 
+        if context:
+            context.base_payload = last_payload
         encoding = '/'.join(encoding_list)
         return {'encoding': encoding, 'data': data}
 
     @classmethod
-    def from_encoded_array(cls, objs, cipher=None):
-        return [cls.from_encoded(obj, cipher=cipher) for obj in objs]
+    def from_encoded_array(cls, objs, cipher=None, context=None):
+        return [cls.from_encoded(obj, cipher=cipher, context=context) for obj in objs]

ably/types/options.py

@@ -1,20 +1,38 @@
 import random
 import logging
+from abc import ABC, abstractmethod
 
 from ably.transport.defaults import Defaults
 from ably.types.authoptions import AuthOptions
 
 log = logging.getLogger(__name__)
 
 
+class VCDiffDecoder(ABC):
+    """
+    The VCDiffDecoder class defines the interface for delta decoding operations.
+
+    This class serves as an abstract base class for implementing delta decoding
+    algorithms, which are used to generate target bytes from compressed delta
+    bytes and base bytes. Subclasses of this class should implement the decode
+    method to handle the specifics of delta decoding. The decode method typically
+    takes a delta bytes and base bytes as input and returns the decoded output.
+
+    """
+    @abstractmethod
+    def decode(self, delta: bytes, base: bytes) -> bytes:
+        pass
+
+
 class Options(AuthOptions):
     def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realtime_host=None, port=0,
                  tls_port=0, use_binary_protocol=True, queue_messages=False, recover=False, environment=None,
                  http_open_timeout=None, http_request_timeout=None, realtime_request_timeout=None,
                  http_max_retry_count=None, http_max_retry_duration=None, fallback_hosts=None,
                  fallback_retry_timeout=None, disconnected_retry_timeout=None, idempotent_rest_publishing=None,
                  loop=None, auto_connect=True, suspended_retry_timeout=None, connectivity_check_url=None,
-                 channel_retry_timeout=Defaults.channel_retry_timeout, add_request_ids=False, **kwargs):
+                 channel_retry_timeout=Defaults.channel_retry_timeout, add_request_ids=False,
+                 vcdiff_decoder: VCDiffDecoder = None, **kwargs):
 
         super().__init__(**kwargs)
 
@@ -77,6 +95,7 @@ def __init__(self, client_id=None, log_level=0, tls=True, rest_host=None, realti
         self.__connectivity_check_url = connectivity_check_url
         self.__fallback_realtime_host = None
         self.__add_request_ids = add_request_ids
+        self.__vcdiff_decoder = vcdiff_decoder
 
         self.__rest_hosts = self.__get_rest_hosts()
         self.__realtime_hosts = self.__get_realtime_hosts()
@@ -259,6 +278,10 @@ def fallback_realtime_host(self, value):
     def add_request_ids(self):
         return self.__add_request_ids
 
+    @property
+    def vcdiff_decoder(self):
+        return self.__vcdiff_decoder
+
     def __get_rest_hosts(self):
         """
         Return the list of hosts as they should be tried. First comes the main

ably/types/presence.py

@@ -86,7 +86,7 @@ def extras(self):
         return self.__extras
 
     @staticmethod
-    def from_encoded(obj, cipher=None):
+    def from_encoded(obj, cipher=None, context=None):
         id = obj.get('id')
         action = obj.get('action', PresenceAction.ENTER)
         client_id = obj.get('clientId')

ably/vcdiff/__init__.py

@@ -0,0 +1,82 @@
+"""
+VCDiff Decoder for Ably Python SDK
+
+This module provides a production-ready VCDiff decoder using the vcdiff-decoder library.
+It implements the VCDiffDecoder interface.
+
+Usage:
+    from ably.vcdiff import AblyVCDiffDecoder, AblyRealtime
+
+    # Create VCDiff decoder
+    vcdiff_decoder = AblyVCDiffDecoder()
+
+    # Create client with decoder
+    client = AblyRealtime(key="your-key", vcdiff_decoder=vcdiff_decoder)
+
+    # Get channel with delta enabled
+    channel = client.channels.get("test", ChannelOptions(params={"delta": "vcdiff"}))
+"""
+
+import logging
+
+from ably.types.options import VCDiffDecoder
+from ably.util.exceptions import AblyException
+
+log = logging.getLogger(__name__)
+
+
+class AblyVCDiffDecoder(VCDiffDecoder):
+    """
+    Production VCDiff decoder using Ably's vcdiff-decoder library.
+
+    Raises:
+        ImportError: If vcdiff is not installed
+        AblyException: If VCDiff decoding fails
+    """
+
+    def __init__(self):
+        """Initialize the VCDiff plugin.
+
+        Raises:
+            ImportError: If vcdiff-decoder library is not available
+        """
+        try:
+            import vcdiff_decoder as vcdiff
+            self._vcdiff = vcdiff
+        except ImportError as e:
+            log.error("vcdiff library not found. Install with: pip install ably[vcdiff]")
+            raise ImportError(
+                "VCDiff plugin requires vcdiff library. "
+                "Install with: pip install ably[vcdiff]"
+            ) from e
+
+    def decode(self, delta: bytes, base: bytes) -> bytes:
+        """
+        Decode a VCDiff delta against a base payload.
+
+        Args:
+            delta: The VCDiff-encoded delta data
+            base: The base payload to apply the delta to
+
+        Returns:
+            bytes: The decoded message payload
+
+        Raises:
+            AblyException: If VCDiff decoding fails (error code 40018)
+        """
+        if not isinstance(delta, bytes):
+            raise TypeError("Delta must be bytes")
+        if not isinstance(base, bytes):
+            raise TypeError("Base must be bytes")
+
+        try:
+            # Use the vcdiff library to decode
+            result = self._vcdiff.decode(base, delta)
+            return result
+        except Exception as e:
+            log.error(f"VCDiff decode failed: {e}")
+            raise AblyException(f"VCDiff decode failure: {e}", 40018, 40018) from e
+
+
+# Export for easy importing
+__all__ = ['AblyVCDiffDecoder']