diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index 99c8bbe..f65d8df 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -56,6 +56,29 @@ Single message payload: - RF: `snr` (float|nil), `rssi` (int|nil), `hop_limit` (int|nil) - Meta: `channel_name` (string; only when not encrypted and known), `ingestor` (canonical host id), `lora_freq`, `modem_preset` +**Cross-ingestor deduplication.** The `id` field is the sole dedup key — the server collapses repeat POSTs on the `messages.id` PRIMARY KEY. Protocols that lack a firmware-assigned packet ID MUST derive a stable, sender-side fingerprint so that the same physical transmission heard by multiple ingestors produces the same `id`. The id MUST fit in 53 bits (`0 <= id <= (1 << 53) - 1`) to round-trip through the JavaScript frontend without precision loss. + +For MeshCore the canonical fingerprint is: + +``` +v1:::: +``` + +hashed with SHA-256 and truncated to 53 bits (first 7 bytes, masked). Components: + +- `sender_identity` — for channel messages, the lowercased+stripped sender name parsed from a leading `SenderName:` prefix in the message text (split on the first colon, surrounding whitespace stripped); for direct messages, the sender's `pubkey_prefix` from the MeshCore event payload. Empty string when unavailable — when the channel-message text lacks any `SenderName:` prefix the dedup degrades and two distinct senders sharing timestamp + channel + text collide. In practice MeshCore clients always prefix the name; the residual risk is anonymous/malformed transmissions. +- `sender_timestamp` — Unix seconds from the sender's clock (identical across receivers). +- `discriminator` — `c` for channel messages on channel `N`, `dm` for direct messages. +- `text` — the message text exactly as transmitted. + +The `v1:` prefix lets the format evolve (e.g. add a channel-secret hash) without colliding with previously-written ids. + +**Known limitations of the v1 fingerprint:** + +- *Format-string ambiguity around `:`.* Components are joined with literal colons and not length-prefixed, so a colon embedded in `sender_identity` or `text` shifts the boundary between fields. In theory two distinct triples (e.g. `sender_identity="a:b"` vs `sender_identity="a"` with a leading `b:` in `text`) can produce the same fingerprint. In practice this is vanishingly rare — MeshCore sender names rarely contain colons and even then both senders would have to land on the same timestamp/channel — but a `v2` revision should switch to a delimiter that cannot appear in any component (e.g. `\x00`) or length-prefix each field. +- *meshcore_py text-decoding inconsistency.* The upstream `meshcore_py` reader strips trailing `\0` bytes on the real-time `CHANNEL_MSG_RECV` path but not on the sync-replay path. If the same physical message is heard once in real-time and once via sync-replay, the byte sequences differ → different fingerprints → duplicate row. Out of scope for the ingestor; track upstream. +- *Sender-side clock reset.* MeshCore nodes without an RTC start `sender_timestamp` from `0` after reboot. Two messages from the same sender containing the same text within one second of power-on collapse into a single row. Acceptable trade-off given the alternative (no dedup at all). + #### `POST /api/positions` Single position payload: diff --git a/data/mesh_ingestor/protocols/meshcore.py b/data/mesh_ingestor/protocols/meshcore.py index b1da264..1f3d46b 100644 --- a/data/mesh_ingestor/protocols/meshcore.py +++ b/data/mesh_ingestor/protocols/meshcore.py @@ -123,27 +123,58 @@ _MESHCORE_ADV_TYPE_ROLE: dict[int, str] = { # --------------------------------------------------------------------------- -def _derive_message_id(sender_ts: int, discriminator: str, text: str) -> int: - """Derive a stable 32-bit message ID from available MeshCore fields. +_MESHCORE_ID_BITS = 53 +"""Width of the synthetic MeshCore message ID, in bits. - MeshCore does not assign firmware-side packet IDs. This function - produces a deterministic 32-bit integer so that re-delivered messages - resolve to the same database row via the UPSERT ON CONFLICT path, while - messages that differ in timestamp, channel/peer, or text content produce - distinct IDs. +53 bits keeps the value within :js:data:`Number.MAX_SAFE_INTEGER` +(``2**53 - 1``) so the JSON ID round-trips through the JavaScript frontend +without precision loss, while giving roughly :math:`2^{26.5}` (~95 million) +distinct messages of birthday-collision headroom. +""" + +_MESHCORE_ID_MASK = (1 << _MESHCORE_ID_BITS) - 1 +"""Bitmask applied to the SHA-256 prefix to clamp the id to 53 bits.""" + + +def _derive_message_id( + sender_identity: str, + sender_ts: int, + discriminator: str, + text: str, +) -> int: + """Derive a stable 53-bit message ID from sender-side MeshCore fields. + + MeshCore does not assign firmware-side packet IDs. This function produces + a deterministic 53-bit integer fingerprint of a physical transmission so + that the same packet heard by multiple ingestors collapses to a single + ``messages`` row via the ``messages.id`` PRIMARY KEY upsert path. Every + component of the fingerprint is sender-side, ensuring two receivers with + different clocks or roster state still compute the same value. Parameters: - sender_ts: Unix timestamp from the sender's clock. - discriminator: Channel index (``"c"`` for channel messages) or - pubkey prefix (for direct messages) to separate messages with - the same timestamp. - text: Message text. + sender_identity: Stable sender identifier shared across receivers. + For channel messages this is the lowercased+stripped sender name + parsed from the message text via :func:`_parse_sender_name`; for + direct messages it is the sender's MeshCore ``pubkey_prefix``. + Must be a string (use ``""`` when unavailable). + sender_ts: Unix timestamp from the sender's clock (identical across + receivers regardless of receiver-side clock skew). + discriminator: Namespace tag separating message classes that could + otherwise collide. ``"c"`` is reserved for channel messages + on channel ``N``; ``"dm"`` is reserved for direct messages. + text: Message text exactly as transmitted by the sender. Returns: - A non-negative 32-bit integer suitable for the ``id`` column. + A non-negative 53-bit integer suitable for the ``id`` column. The + value is bounded by ``0 <= id <= (1 << 53) - 1`` so it survives the + JSON → JavaScript number round-trip without precision loss. """ - data = f"{sender_ts}:{discriminator}:{text}".encode("utf-8", errors="replace") - return int.from_bytes(hashlib.sha256(data).digest()[:4], "big") + # The ``v1:`` prefix lets us evolve the fingerprint format (e.g. add a + # channel-secret hash) by bumping to ``v2:`` without colliding with + # existing ids written under the v1 scheme. + fingerprint = f"v1:{sender_identity}:{sender_ts}:{discriminator}:{text}" + digest = hashlib.sha256(fingerprint.encode("utf-8", errors="replace")).digest() + return int.from_bytes(digest[:7], "big") & _MESHCORE_ID_MASK def _meshcore_node_id(public_key_hex: str | None) -> str | None: @@ -904,8 +935,18 @@ def _make_event_handlers(iface: _MeshcoreInterface, target: str | None) -> dict: ) iface._synthetic_node_ids.add(mention_id) + # The dedup fingerprint uses the parsed sender name (lowercased and + # stripped) rather than ``from_id``: each ingestor independently + # resolves Alice to either her real ``!aabbccdd`` (when she is in its + # contact roster) or to a synthetic id derived from her name; the + # parsed name lives in the message text itself, so it is identical + # across all receivers regardless of roster state. + sender_identity = (sender_name or "").strip().lower() + packet = { - "id": _derive_message_id(sender_ts, f"c{channel_idx}", text), + "id": _derive_message_id( + sender_identity, sender_ts, f"c{channel_idx}", text + ), "rxTime": rx_time, "rx_time": rx_time, "from_id": from_id, @@ -941,8 +982,12 @@ def _make_event_handlers(iface: _MeshcoreInterface, target: str | None) -> dict: pubkey_prefix = payload.get("pubkey_prefix", "") from_id = iface.lookup_node_id(pubkey_prefix) + # ``pubkey_prefix`` is already a sender-side stable identifier (the + # first six bytes of the sender's public key); ``"dm"`` namespaces + # direct messages so they cannot collide with channel messages that + # happen to share the other components. packet = { - "id": _derive_message_id(sender_ts, pubkey_prefix or "", text), + "id": _derive_message_id(pubkey_prefix or "", sender_ts, "dm", text), "rxTime": rx_time, "rx_time": rx_time, "from_id": from_id, diff --git a/tests/test_provider_unit.py b/tests/test_provider_unit.py index c02ed84..a7a97b7 100644 --- a/tests/test_provider_unit.py +++ b/tests/test_provider_unit.py @@ -1202,46 +1202,108 @@ def test_interface_close_is_idempotent(): def test_derive_message_id_is_deterministic(): """Same inputs must always produce the same ID.""" - assert _derive_message_id(1_000_000, "c0", "hello") == _derive_message_id( - 1_000_000, "c0", "hello" + assert _derive_message_id("alice", 1_000_000, "c0", "hello") == _derive_message_id( + "alice", 1_000_000, "c0", "hello" ) def test_derive_message_id_differs_by_channel(): """Messages on different channels with the same timestamp must not collide.""" - assert _derive_message_id(1_000_000, "c0", "hello") != _derive_message_id( - 1_000_000, "c1", "hello" + assert _derive_message_id("alice", 1_000_000, "c0", "hello") != _derive_message_id( + "alice", 1_000_000, "c1", "hello" ) def test_derive_message_id_differs_by_text(): """Messages with different text must produce different IDs.""" - assert _derive_message_id(1_000_000, "c0", "hello") != _derive_message_id( - 1_000_000, "c0", "world" + assert _derive_message_id("alice", 1_000_000, "c0", "hello") != _derive_message_id( + "alice", 1_000_000, "c0", "world" ) def test_derive_message_id_differs_by_timestamp(): """Messages at different timestamps must produce different IDs.""" - assert _derive_message_id(1_000_000, "c0", "hi") != _derive_message_id( - 1_000_001, "c0", "hi" + assert _derive_message_id("alice", 1_000_000, "c0", "hi") != _derive_message_id( + "alice", 1_000_001, "c0", "hi" ) -def test_derive_message_id_is_32bit(): - """Result must fit in a 32-bit unsigned integer.""" - result = _derive_message_id(1_758_000_000, "aabbccddee11", "some text") - assert 0 <= result <= 0xFFFFFFFF +def test_derive_message_id_is_53bit(): + """Result must fit in JS ``Number.MAX_SAFE_INTEGER`` (2**53 - 1). + + Federation passes the id through JSON, where Number values exceeding + 53 bits lose precision in the JavaScript frontend. Clamping to 53 bits + preserves the value across the round-trip while leaving ample collision + headroom (~95M messages at the 50% birthday bound). + """ + result = _derive_message_id("alice", 1_758_000_000, "c0", "some text") + assert 0 <= result <= (1 << 53) - 1 def test_derive_message_id_distinguishes_long_messages_differing_after_128_chars(): """Messages that share the first 128 characters must still get different IDs.""" prefix = "A" * 128 - id_a = _derive_message_id(1_000_000, "c0", prefix + "AAAAAA") - id_b = _derive_message_id(1_000_000, "c0", prefix + "BBBBBB") + id_a = _derive_message_id("alice", 1_000_000, "c0", prefix + "AAAAAA") + id_b = _derive_message_id("alice", 1_000_000, "c0", prefix + "BBBBBB") assert id_a != id_b +def test_derive_message_id_includes_sender_identity(): + """Two senders posting the same text on the same channel/second must NOT collide. + + Regression test for issue #751: prior to the fix the channel-message + fingerprint omitted the sender entirely, so Alice and Bob both posting + "ack" at the same instant collapsed into a single row. + """ + alice_id = _derive_message_id("alice", 1_000_000, "c0", "ack") + bob_id = _derive_message_id("bob", 1_000_000, "c0", "ack") + assert alice_id != bob_id + + +def test_derive_message_id_channel_vs_dm_disjoint(): + """Channel and direct messages must occupy disjoint id namespaces. + + Without a discriminator that distinguishes the two classes, a channel + message and a DM that happen to share the other components could collide. + """ + channel_id = _derive_message_id("alice", 1_000_000, "c0", "hi") + dm_id = _derive_message_id("alice", 1_000_000, "dm", "hi") + assert channel_id != dm_id + + +def test_derive_message_id_identical_across_receivers(): + """Two ingestors with different roster state must derive the same id. + + The whole point of the fingerprint is that every input is sender-side, so + two physically separate receivers compute the same id and the messages + collapse on the ``messages.id`` PRIMARY KEY upsert. + """ + args = ("alice", 1_758_000_000, "c0", "hello mesh") + assert _derive_message_id(*args) == _derive_message_id(*args) + + +def test_derive_message_id_handles_invalid_utf8(): + """Inputs with surrogate pairs must not raise; ``errors='replace'`` cleans them.""" + bad_text = "before \ud800 after" # lone surrogate is invalid UTF-8 + result = _derive_message_id("alice", 1_000_000, "c0", bad_text) + assert 0 <= result <= (1 << 53) - 1 + + +def test_derive_message_id_anonymous_channel_msgs_still_distinguished_by_other_fields(): + """Anonymous channel msgs (sender_identity="") still differ when text/ts differ. + + The empty sender-identity path is documented as a degraded mode in + CONTRACTS.md (anonymous transmissions cannot be distinguished from each + other when timestamp + channel + text also match). This test pins down + the *non-degraded* behaviour: as long as any of the remaining components + differ, the ids must remain distinct. + """ + base = _derive_message_id("", 1_000_000, "c0", "hi") + assert base != _derive_message_id("", 1_000_001, "c0", "hi") # ts differs + assert base != _derive_message_id("", 1_000_000, "c1", "hi") # channel differs + assert base != _derive_message_id("", 1_000_000, "c0", "hello") # text differs + + # --------------------------------------------------------------------------- # _make_event_handlers — async callbacks # --------------------------------------------------------------------------- @@ -1334,8 +1396,9 @@ def test_on_channel_msg_queues_packet(monkeypatch): assert pkt["from_id"] is None assert pkt["snr"] == 5 assert pkt["rssi"] == -80 - # ID must be the hash-derived value, not the raw timestamp - assert pkt["id"] == _derive_message_id(1_758_000_000, "c2", "hello mesh") + # ID must be the hash-derived value, not the raw timestamp. The text has no + # "Name:" prefix so the sender-identity component is the empty string. + assert pkt["id"] == _derive_message_id("", 1_758_000_000, "c2", "hello mesh") def test_on_channel_msg_resolves_from_id_via_sender_name(monkeypatch): @@ -1535,10 +1598,90 @@ def test_on_contact_msg_queues_packet_with_from_id(monkeypatch): assert pkt["from_id"] == "!aabbccdd" assert pkt["to_id"] == "!deadbeef" assert pkt["id"] == _derive_message_id( - 1_758_000_001, "aabbccddee11", "direct message" + "aabbccddee11", 1_758_000_001, "dm", "direct message" ) +def test_on_channel_msg_id_identical_across_ingestors_with_different_rosters( + monkeypatch, +): + """Two ingestors that hear the same channel message must emit the same id. + + Regression test for issue #751. Ingestor A has Alice in its contact roster + (so ``from_id`` resolves to ``!aabbccdd``); ingestor B does not (so a + synthetic ``from_id`` is created). The dedup id MUST still match because + it is derived from the parsed sender name in the text, not from the + per-ingestor ``from_id`` resolution. + """ + import asyncio + + pub_key = "aabbccdd" + "00" * 28 + payload = { + "sender_timestamp": 1_758_000_999, + "text": "Alice: dedup me", + "channel_idx": 0, + } + + captured_a, _, _, hmap_a = _setup_channel_msg_handlers( + monkeypatch, + contacts=[{"public_key": pub_key, "adv_name": "Alice"}], + ) + asyncio.run(hmap_a["CHANNEL_MSG_RECV"](_FakeEvt(payload))) + + captured_b, _, _, hmap_b = _setup_channel_msg_handlers(monkeypatch) + asyncio.run(hmap_b["CHANNEL_MSG_RECV"](_FakeEvt(payload))) + + assert len(captured_a) == 1 + assert len(captured_b) == 1 + # Different ingestors → different from_id resolution, but the dedup id is + # identical because it comes from the parsed sender name and the + # sender-side timestamp/text. + assert captured_a[0]["from_id"] != captured_b[0]["from_id"] + assert captured_a[0]["id"] == captured_b[0]["id"] + + +def test_on_contact_msg_id_identical_across_ingestors_with_different_rosters( + monkeypatch, +): + """Two ingestors that hear the same DM must emit the same id. + + Direct messages already carry the sender's ``pubkey_prefix`` in the event + payload, so the dedup id is identical regardless of contact-roster state. + """ + import asyncio + import data.mesh_ingestor as _mesh_pkg + import data.mesh_ingestor.protocols.meshcore as _mod + + pub_key = "aabbccddee11" + "00" * 26 + payload = { + "sender_timestamp": 1_758_000_998, + "text": "private hello", + "pubkey_prefix": "aabbccddee11", + } + + def _run(with_contact: bool): + captured: list = [] + stub = _make_stub_handlers_module() + stub.store_packet_dict = lambda pkt: captured.append(pkt) + monkeypatch.setattr(_mod.config, "_debug_log", lambda *_a, **_k: None) + monkeypatch.setattr(_mesh_pkg, "handlers", stub) + + iface = _MeshcoreInterface(target=None) + iface.host_node_id = "!deadbeef" + if with_contact: + iface._update_contact({"public_key": pub_key, "adv_name": "Alice"}) + hmap = _make_event_handlers(iface, "/dev/ttyUSB0") + asyncio.run(hmap["CONTACT_MSG_RECV"](_FakeEvt(payload))) + return captured + + captured_a = _run(with_contact=True) + captured_b = _run(with_contact=False) + + assert len(captured_a) == 1 + assert len(captured_b) == 1 + assert captured_a[0]["id"] == captured_b[0]["id"] + + def test_on_channel_msg_skips_empty_text(monkeypatch): """on_channel_msg must not queue a packet when text is absent.""" import asyncio