fix: decode a batch of delta messages as a chain

A protocol message can carry several messages, each a delta from the one
before it, but the decoding context only advanced once per protocol
message, so any batch of two or more failed on the second message with
"previous message not available" and the whole batch was dropped.

- Advance the context's last message id as each message decodes, so
  deltas chain within a batch as well as across batches, as ably-js does
- Restore the context if any message in a batch fails, so the batch the
  server replays decodes against the right base payload rather than
  against the output of its own first messages
- Skip the delta check when there is no decoding context, as the REST
  paths pass, instead of raising AttributeError

Tests cover chaining within and across batches, out-of-order rejection,
the context being left untouched after both failure modes, the
context-less paths, and a channel receiving a three-message delta batch.

Refs #692

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: ttypic

ably/realtime/channel.py

@@ -739,7 +739,6 @@ def _on_message(self, proto_msg: dict) -> None:
             try:
                 messages = Message.from_encoded_array(proto_msg.get('messages'),
                                                       cipher=self.cipher, 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

ably/types/message.py

@@ -300,12 +300,17 @@ def from_encoded(obj, cipher=None, context=None):
         version = obj.get('version', None)
 
         delta_extra = DeltaExtras(extras)
-        if delta_extra.from_id and delta_extra.from_id != context.last_message_id:
+        if context and 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)
 
+        # A protocol message can carry several messages, each a delta from the one before it, so
+        # the base for the next delta is this message rather than the last one of the batch.
+        if context:
+            context.last_message_id = id
+
         if action is not None:
             try:
                 action = MessageAction(action)

ably/types/mixins.py

@@ -127,4 +127,16 @@ def decode(data, encoding='', cipher=None, context=None):
 
     @classmethod
     def from_encoded_array(cls, objs, cipher=None, context=None):
-        return [cls.from_encoded(obj, cipher=cipher, context=context) for obj in objs]
+        if context is None:
+            return [cls.from_encoded(obj, cipher=cipher) for obj in objs]
+
+        # Decoding a message advances the context, and a batch that fails part way through is
+        # discarded whole and replayed by the server, so the context is only kept if every
+        # message in it decoded. Otherwise the replayed batch would decode against the base
+        # payload left behind by its own first messages.
+        base_payload, last_message_id = context.base_payload, context.last_message_id
+        try:
+            return [cls.from_encoded(obj, cipher=cipher, context=context) for obj in objs]
+        except Exception:
+            context.base_payload, context.last_message_id = base_payload, last_message_id
+            raise

pyproject.toml

@@ -40,7 +40,7 @@ dependencies = [
 [project.optional-dependencies]
 oldcrypto = ["pycrypto>=2.6.1,<3.0.0"]
 crypto = ["pycryptodome"]
-vcdiff = ["vcdiff-decoder>=0.1.0,<0.2.0"]
+vcdiff = ["vcdiff-decoder>=0.2.0,<0.3.0"]
 dev = [
   "pytest>=7.1,<8.0",
   "pytest-asyncio>=0.21.0,<0.23.0; python_version=='3.7'",
@@ -55,7 +55,7 @@ dev = [
   "pytest-timeout>=2.1.0,<3.0.0",
   "async-case>=10.1.0,<11.0.0; python_version=='3.7'",
   "tokenize_rt",
-  "vcdiff-decoder>=0.1.0a1",
+  "vcdiff-decoder>=0.2.0",
 ]
 
 [project.scripts]

test/ably/realtime/deltadecoding_test.py

@@ -0,0 +1,179 @@
+"""
+Unit tests for delta (vcdiff) message decoding.
+
+A protocol message can carry several messages, each one a delta from the message before it, so the
+decoding context is chained per message and a batch that fails part way through leaves it untouched.
+"""
+
+import base64
+
+import pytest
+
+from ably import AblyRealtime
+from ably.types.channelstate import ChannelState
+from ably.types.message import Message
+from ably.types.mixins import DecodingContext
+from ably.types.options import VCDiffDecoder
+from ably.util.exceptions import AblyException
+from test.ably.utils import BaseAsyncTestCase
+
+
+class AppendingDecoder(VCDiffDecoder):
+    """A decoder whose output depends on the base, so a wrong base is visible in the result"""
+
+    def __init__(self):
+        self.number_of_calls = 0
+
+    def decode(self, delta: bytes, base: bytes) -> bytes:
+        self.number_of_calls += 1
+        return base + delta
+
+
+class FailingDecoder(VCDiffDecoder):
+
+    def decode(self, delta: bytes, base: bytes) -> bytes:
+        raise Exception("Failed to decode delta.")
+
+
+def full_message(id, data):
+    return {'id': id, 'name': id, 'data': data, 'encoding': 'utf-8'}
+
+
+def delta_message(id, delta, from_id):
+    """A message carrying `delta` as a vcdiff delta from the message `from_id`"""
+    return {
+        'id': id,
+        'name': id,
+        'data': base64.b64encode(delta.encode()).decode(),
+        'encoding': 'utf-8/vcdiff/base64',
+        'extras': {'delta': {'format': 'vcdiff', 'from': from_id}},
+    }
+
+
+class TestDeltaBatchDecoding(BaseAsyncTestCase):
+    """Message.from_encoded_array over a protocol message's messages"""
+
+    def test_batch_of_chained_deltas_decodes(self):
+        decoder = AppendingDecoder()
+        context = DecodingContext(vcdiff_decoder=decoder)
+
+        messages = Message.from_encoded_array([
+            full_message('m:0', 'a'),
+            delta_message('m:1', 'b', from_id='m:0'),
+            delta_message('m:2', 'c', from_id='m:1'),
+        ], context=context)
+
+        assert [message.data for message in messages] == ['a', 'ab', 'abc']
+        assert decoder.number_of_calls == 2
+        assert context.last_message_id == 'm:2'
+
+    def test_delta_chains_from_the_previous_batch(self):
+        context = DecodingContext(vcdiff_decoder=AppendingDecoder())
+
+        Message.from_encoded_array([full_message('m:0', 'a')], context=context)
+        messages = Message.from_encoded_array([delta_message('m:1', 'b', from_id='m:0')],
+                                              context=context)
+
+        assert [message.data for message in messages] == ['ab']
+
+    def test_delta_from_an_unknown_message_is_rejected(self):
+        context = DecodingContext(vcdiff_decoder=AppendingDecoder())
+        Message.from_encoded_array([full_message('m:0', 'a')], context=context)
+
+        with pytest.raises(AblyException) as excinfo:
+            Message.from_encoded_array([delta_message('m:9', 'b', from_id='m:8')], context=context)
+
+        assert excinfo.value.code == 40018
+
+    def test_a_batch_that_fails_leaves_the_context_unchanged(self):
+        # RTL18: the batch is discarded and replayed by the server from the last channel serial,
+        # so the base payload has to still be the one the replayed batch decodes against
+        context = DecodingContext(vcdiff_decoder=AppendingDecoder())
+        Message.from_encoded_array([full_message('m:0', 'a')], context=context)
+        base_payload, last_message_id = context.base_payload, context.last_message_id
+
+        with pytest.raises(AblyException):
+            Message.from_encoded_array([
+                delta_message('m:1', 'b', from_id='m:0'),
+                delta_message('m:2', 'c', from_id='out-of-order'),
+            ], context=context)
+
+        assert context.base_payload == base_payload
+        assert context.last_message_id == last_message_id
+
+    def test_a_batch_whose_decoder_fails_leaves_the_context_unchanged(self):
+        context = DecodingContext(vcdiff_decoder=FailingDecoder())
+        Message.from_encoded_array([full_message('m:0', 'a')], context=context)
+        base_payload, last_message_id = context.base_payload, context.last_message_id
+
+        with pytest.raises(AblyException) as excinfo:
+            Message.from_encoded_array([delta_message('m:1', 'b', from_id='m:0')], context=context)
+
+        assert excinfo.value.code == 40018
+        assert context.base_payload == base_payload
+        assert context.last_message_id == last_message_id
+
+    def test_messages_decode_without_a_decoding_context(self):
+        # the REST paths (history, for one) decode without a context
+        messages = Message.from_encoded_array([full_message('m:0', 'a')])
+
+        assert [message.data for message in messages] == ['a']
+
+    def test_a_delta_without_a_decoding_context_reports_the_missing_decoder(self):
+        # and not a failure to look up the previous message id in a context that isn't there
+        with pytest.raises(AblyException) as excinfo:
+            Message.from_encoded_array([delta_message('m:1', 'b', from_id='m:0')])
+
+        assert excinfo.value.code == 40019
+
+
+class TestChannelDeltaBatch(BaseAsyncTestCase):
+    """RealtimeChannel handling of a protocol message carrying several deltas"""
+
+    def setup_channel(self, decoder):
+        ably = AblyRealtime(key='not_a.real:key', auto_connect=False, vcdiff_decoder=decoder)
+        channel = ably.channels.get('delta')
+        received = []
+        # subscribe() would also wait for the channel to attach, which this client cannot do
+        channel._RealtimeChannel__message_emitter.on(lambda message: received.append(message))
+        return ably, channel, received
+
+    async def test_every_message_of_a_delta_batch_is_emitted(self):
+        ably, channel, received = self.setup_channel(AppendingDecoder())
+
+        channel._on_message({
+            'action': 15,
+            'channel': 'delta',
+            'channelSerial': 'serial-1',
+            'messages': [
+                full_message('m:0', 'a'),
+                delta_message('m:1', 'b', from_id='m:0'),
+                delta_message('m:2', 'c', from_id='m:1'),
+            ],
+        })
+
+        assert [message.data for message in received] == ['a', 'ab', 'abc']
+        assert channel._RealtimeChannel__channel_serial == 'serial-1'
+        await ably.close()
+
+    async def test_a_batch_that_fails_to_decode_starts_recovery(self):
+        ably, channel, received = self.setup_channel(FailingDecoder())
+        channel._on_message({
+            'action': 15, 'channel': 'delta', 'channelSerial': 'serial-1',
+            'messages': [full_message('m:0', 'a')],
+        })
+
+        channel._on_message({
+            'action': 15,
+            'channel': 'delta',
+            'channelSerial': 'serial-2',
+            'messages': [delta_message('m:1', 'b', from_id='m:0')],
+        })
+
+        # RTL18b/RTL18c: the batch is dropped and the channel reattaches from the serial of the
+        # last batch it decoded
+        assert [message.data for message in received] == ['a']
+        assert channel._RealtimeChannel__channel_serial == 'serial-1'
+        assert channel.state == ChannelState.ATTACHING
+        assert channel.error_reason.code == 40018
+        await ably.close()