feat: Enhance vnode with unique packet handling and update examples for clarity

Author: pdxlocations

README.md

@@ -82,6 +82,10 @@ Supported `VirtualNode` calls:
 - MeshInterface-compatible sends: `sendText()`, `sendAlert()`, `sendData()`, `sendPosition()`
 - Packet helpers: `is_direct_message_for_me()`, `is_text_message()`, `get_text_message()`, `reply_to_packet()`
 
+`receive(callback)` is deduplicated. Repeated multicast copies stay on `mudp`'s wire-level
+topics, while `vnode` publishes `meshtastic.receive` from `mesh.rx.unique_packet` so
+application callbacks run once per logical packet.
+
 ## Examples
 
 Use these as small runnable references for common tasks:
@@ -93,7 +97,7 @@ Use these as small runnable references for common tasks:
 - `examples/serial_or_vnode.py`: try a serial Meshtastic node first, then fall back to `VirtualNode`
 - `examples/send_dm.py`: minimal one-shot direct-message sender
 - `examples/library_embed.py`: minimal application-style embedding example using `VirtualNode` directly
-- `examples/watch_reliability.py`: watcher for ACK, NAK, retry, and retransmit-failure events
+- `examples/watch_reliability.py`: watcher for unique ACK, NAK, retry, and retransmit-failure events
 
 ```bash
 .venv/bin/python examples/autoresponder.py

examples/README.md

@@ -16,7 +16,8 @@ It skips duplicates, ignores replies, and alternates between an emoji and plain-
 
 Shows the minimal `node.receive(callback)` / `node.unreceive(callback)` pattern using
 Meshtastic-style `(packet, interface)` callbacks. Edit `VNODE_FILE` in the script if you
-want to use a config path other than `node.json`.
+want to use a config path other than `node.json`. The callback is fed from `mudp`'s
+unique-packet topic, so duplicates are already suppressed before app code sees them.
 
 ```bash
 .venv/bin/python examples/basic_subscriptions.py
@@ -72,8 +73,9 @@ callback shape.
 
 ## Watch ACK and retry events
 
-Subscribes to `mudp` reliability events and prints ACK, NAK, retry, and max-retransmit updates.
-Optionally sends a startup DM so you can watch the full reliability lifecycle immediately.
+Subscribes to `mudp` reliability events and prints unique ACK, NAK, retry, and
+max-retransmit updates. Optionally sends a startup DM so you can watch the full
+reliability lifecycle immediately.
 
 ```bash
 .venv/bin/python examples/watch_reliability.py --vnode-file node.json --to '!1234abcd' --message 'hello'

examples/watch_reliability.py

@@ -72,7 +72,8 @@ def main() -> int:
     node.start()
 
     # These pubsub topics come from mudp, not the example itself.
-    # They let app code observe ACK resolution and retry behavior without reimplementing transport logic.
+    # ACK/NAK are emitted once per logical packet on mudp's unique-only routing topics.
+    # Use mesh.rx.packet / mesh.rx.duplicate instead if you need every wire observation.
     pub.subscribe(on_ack, "mesh.rx.ack")
     pub.subscribe(on_nak, "mesh.rx.nak")
     pub.subscribe(on_retry, "mesh.tx.retry")

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "vnode"
-version = "0.1.7"
+version = "0.1.8"
 description = "Virtual Meshtastic node using mudp transport and meshdb storage."
 authors = [
     {name = "Ben Lipsey",email = "ben@pdxlocations.com"}
@@ -12,7 +12,7 @@ dependencies = [
   "cryptography>=43,<45",
   "meshdb>=0.1.9",
   "meshtastic>=2.7,<3.0",
-  "mudp>=1.1.0",
+  "mudp>=1.1.5",
 ]
 
 [project.urls]

vnode/tests/test_crypto.py

@@ -866,6 +866,44 @@ def on_receive(packet, interface) -> None:
             self.assertEqual(packet["decoded"]["text"], "hello")
             self.assertIs(packet["raw"], inbound)
 
+    def test_receive_callback_runs_once_when_duplicate_wire_copy_is_not_unique(self) -> None:
+        with tempfile.TemporaryDirectory() as temp_dir:
+            root = Path(temp_dir)
+            _public_key, private_key = generate_keypair()
+            config_path = self._write_temp_config(root, b64_encode(private_key))
+            node = VirtualNode(config_path)
+
+            received: list[tuple[dict[str, object], object]] = []
+
+            def on_receive(packet, interface) -> None:
+                received.append((packet, interface))
+
+            node.receive(on_receive)
+            try:
+                inbound = mesh_pb2.MeshPacket()
+                inbound.id = 5152
+                setattr(inbound, "from", int("12345678", 16))
+                inbound.to = node.node_num
+                inbound.channel = 0
+                inbound.hop_limit = 3
+                inbound.hop_start = 3
+                inbound.decoded.portnum = portnums_pb2.PortNum.TEXT_MESSAGE_APP
+                inbound.decoded.payload = b"hello once"
+
+                node._handle_raw_packet(inbound)
+                node._handle_unique_packet(inbound)
+
+                duplicate = mesh_pb2.MeshPacket()
+                duplicate.CopyFrom(inbound)
+                node._handle_raw_packet(duplicate)
+            finally:
+                node.unreceive(on_receive)
+
+            self.assertEqual(len(received), 1)
+            packet, interface = received[0]
+            self.assertIs(interface, node)
+            self.assertEqual(packet["decoded"]["text"], "hello once")
+
     def test_sendText_compat_returns_meshpacket(self) -> None:
         with tempfile.TemporaryDirectory() as temp_dir:
             root = Path(temp_dir)

vnode/vnode/runtime.py

@@ -63,7 +63,10 @@ def resolve_role(value: Union[str, int]) -> int:
 
 
 class VirtualNode:
+    RAW_PACKET_TOPIC = "mesh.rx.packet"
     PACKET_TOPIC = "mesh.rx.unique_packet"
+    ACK_TOPIC = "mesh.rx.ack"
+    NAK_TOPIC = "mesh.rx.nak"
     DUPLICATE_TOPIC = "mesh.rx.duplicate"
     RECEIVE_TOPIC = "meshtastic.receive"
 
@@ -142,11 +145,11 @@ def _configure_mudp_globals(self) -> None:
     def start(self) -> None:
         if self.stream is not None:
             return
-        pub.subscribe(self._handle_raw_packet, "mesh.rx.packet")
+        pub.subscribe(self._handle_raw_packet, self.RAW_PACKET_TOPIC)
         pub.subscribe(self._handle_unique_packet, self.PACKET_TOPIC)
         pub.subscribe(self._handle_compat_response_packet, self.RECEIVE_TOPIC)
-        pub.subscribe(self._handle_compat_ack, "mesh.rx.ack")
-        pub.subscribe(self._handle_compat_nak, "mesh.rx.nak")
+        pub.subscribe(self._handle_compat_ack, self.ACK_TOPIC)
+        pub.subscribe(self._handle_compat_nak, self.NAK_TOPIC)
         self.stream = UDPPacketStream(
             self.config.udp.mcast_group,
             int(self.config.udp.mcast_port),
@@ -169,7 +172,7 @@ def stop(self) -> None:
             self.stream.stop()
             self.stream = None
         try:
-            pub.unsubscribe(self._handle_raw_packet, "mesh.rx.packet")
+            pub.unsubscribe(self._handle_raw_packet, self.RAW_PACKET_TOPIC)
         except KeyError:
             pass
         try:
@@ -178,8 +181,8 @@ def stop(self) -> None:
             pass
         for topic, listener in (
             (self.RECEIVE_TOPIC, self._handle_compat_response_packet),
-            ("mesh.rx.ack", self._handle_compat_ack),
-            ("mesh.rx.nak", self._handle_compat_nak),
+            (self.ACK_TOPIC, self._handle_compat_ack),
+            (self.NAK_TOPIC, self._handle_compat_nak),
         ):
             try:
                 pub.unsubscribe(listener, topic)