diff --git a/pyproject.toml b/pyproject.toml index d8deb27..7694a7e 100644 --- a/pyproject.toml +++ b/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" diff --git a/serverish/messenger/messenger.py b/serverish/messenger/messenger.py index 81870d9..d626ef8 100644 --- a/serverish/messenger/messenger.py +++ b/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 diff --git a/serverish/messenger/msg_publisher.py b/serverish/messenger/msg_publisher.py index ac57659..95dacf4 100644 --- a/serverish/messenger/msg_publisher.py +++ b/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 diff --git a/tests/test_messenger_publish_ack.py b/tests/test_messenger_publish_ack.py index 8555d65..9c1a642 100644 --- a/tests/test_messenger_publish_ack.py +++ b/tests/test_messenger_publish_ack.py @@ -11,8 +11,15 @@ @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 :. + + 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() @@ -20,7 +27,34 @@ async def test_publish_sets_msg_id_header(messenger, unique_subject): 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"