Fix silent message loss: globally unique Nats-Msg-Id (2.2.1 hotfix)

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "serverish"
-version = "2.2.0"
+version = "2.2.1"
 description = "helpers for server alike projects"
 authors = ["Mikołaj Kałuszyński", "MMME team"]
 readme = "README.md"

serverish/messenger/messenger.py

@@ -24,7 +24,7 @@
 from serverish.base import dt_utcnow_array, dt_from_array, Task, create_task, dt_ensure_array
 from serverish.base.collector import Collector
 from serverish.connection.connection_jets import ConnectionJetStream
-from serverish.base.idmanger import gen_id
+from serverish.base.idmanger import gen_id, gen_uid
 from serverish.base.manageable import Manageable
 from serverish.messenger.msgvalidator import MsgValidator
 from serverish.base.singleton import Singleton
@@ -51,6 +51,13 @@ class Messenger(Singleton):
     def __init__(self, name: str = None, parent: Collector = None, **kwargs) -> None:
         self.validator = MsgValidator()
         self.opener_task: Task | None = None
+        # Random per-process token, part of the Nats-Msg-Id dedup header.
+        # meta.id (gen_id) is only process-locally unique - two processes (or one
+        # process restarted within the stream's duplicate window) both emit
+        # 'msg-1', 'msg-2', ..., and JetStream would silently drop the later
+        # publishes as duplicates. The instance prefix makes the dedup identity
+        # global while retries still reuse the exact same header.
+        self.instance_id: str = gen_uid('mi', 12)
         super().__init__(name, parent, **kwargs)
 
     @property

serverish/messenger/msg_publisher.py

@@ -75,7 +75,11 @@ async def publish(self, data: dict | None = None, meta: dict | None = None, **kw
             raise e
         self.messenger.log_msg_trace(msg['data'], msg['meta'], f"PUB to {self.subject}")
         headers = dict(kwargs.pop('headers', None) or {})
-        headers.setdefault('Nats-Msg-Id', msg['meta']['id'])  # server-side dedup, makes ack-timeout retries safe
+        # Server-side dedup id: makes ack-timeout retries safe. Must be globally
+        # unique, hence the per-process instance prefix - meta.id alone is only
+        # process-locally unique and would collide across processes/restarts,
+        # silently dropping messages within the stream's duplicate window.
+        headers.setdefault('Nats-Msg-Id', f"{self.messenger.instance_id}:{msg['meta']['id']}")
         try:
             await self._publish_with_ack_retries(bdata, headers, msg['meta']['id'], **kwargs)
             # Track successful publish

tests/test_messenger_publish_ack.py

@@ -11,16 +11,50 @@
 
 @pytest.mark.nats_js
 async def test_publish_sets_msg_id_header(messenger, unique_subject):
-    """Every publish carries Nats-Msg-Id equal to meta.id."""
+    """Every publish carries Nats-Msg-Id of the form <instance>:<meta.id>.
+
+    The per-process instance prefix is essential: meta.id (gen_id) is only
+    process-locally unique, so a bare meta.id header would collide across
+    processes (and restarts) on a shared stream and JetStream would silently
+    drop the later messages as duplicates.
+    """
     pub = get_publisher(unique_subject)
+    await pub.open()
     msg = await pub.publish(data={'v': 1})
     await pub.close()
 
     js = Messenger().connection.js
     stream = await js.find_stream_name_by_subject(unique_subject)
     raw = await js.get_last_msg(stream, unique_subject)
     assert raw.headers is not None
-    assert raw.headers.get('Nats-Msg-Id') == msg['meta']['id']
+    assert raw.headers.get('Nats-Msg-Id') == f"{Messenger().instance_id}:{msg['meta']['id']}"
+
+
+@pytest.mark.nats_js
+async def test_publish_no_cross_process_dedup_collision(messenger, unique_subject):
+    """A message with the same meta.id as one published by a *different* process
+    (different instance prefix) must NOT be swallowed by stream deduplication."""
+    import uuid
+    js = Messenger().connection.js
+    # Simulate another process publishing the same meta.id: different instance
+    # prefix. Both the prefix and the forced meta.id are unique per test run -
+    # stream-wide dedup would legitimately drop exact repeats (same process
+    # naturally emits 'msg-N' ids; an earlier suite test or a re-run within
+    # the duplicate window would collide, which is this very bug).
+    shared_id = f"msg-x{uuid.uuid4().hex[:8]}"
+    await js.publish(unique_subject,
+                     f'{{"data":{{}},"meta":{{"id":"{shared_id}"}}}}'.encode(),
+                     headers={'Nats-Msg-Id': f'otherproc-{uuid.uuid4().hex[:8]}:{shared_id}'})
+
+    pub = get_publisher(unique_subject)
+    await pub.open()
+    msg = await pub.publish(data={'v': 1}, meta={'id': shared_id})
+    await pub.close()
+
+    reader = get_reader(unique_subject, deliver_policy='all', nowait=True)
+    received = [meta['nats']['seq'] async for _data, meta in reader]
+    await reader.close()
+    assert len(received) == 2, "same meta.id from different processes must both be stored"
 
 
 @pytest.mark.nats_js
@@ -64,7 +98,7 @@ async def flaky_publish(*args, **kwargs):
         await pub.close()
 
     assert len(attempts) == 2, "one failed attempt + one successful retry"
-    assert attempts[0] == attempts[1] == msg['meta']['id'], \
+    assert attempts[0] == attempts[1] == f"{Messenger().instance_id}:{msg['meta']['id']}", \
         "retry must reuse the same Nats-Msg-Id for dedup"