Mobile OS + Updates to Datagram service

MeshService/mesh/transport.py

@@ -20,6 +20,20 @@
 * ``seen_msg_ids`` is paired with a FIFO ``_seen_order`` deque and bounded
   at 4096 entries so a long-running peer doesn't leak as chatter accumulates.
   All de-dup writes go through ``_remember_msg_id``.
+
+ACK protocol:
+  When a node receives a message, it sends back an ACK packet to the
+  original sender. The sender tracks pending ACKs and retries up to
+  MAX_RETRIES times with ACK_TIMEOUT_S between attempts.
+
+  ACK packet format:
+  {
+      "type": "ack",
+      "ack_msg_id": "<msg_id being acknowledged>",
+      "from_id": "<node that received the message>",
+      "from_name": "<name of acknowledging node>",
+      "mesh_id": "<mesh_id>"
+  }
 """
 
 import hashlib
@@ -37,6 +51,9 @@
 SEEN_MSG_ID_CAP = 4096
 DEFAULT_MESH_ID = "beacon-default"
 
+ACK_TIMEOUT_S = 2.0   # seconds to wait before retry
+MAX_RETRIES = 3        # total send attempts (1 original + 2 retries)
+
 
 def _autogen_port(node_id: str) -> int:
     """Deterministic per-node-id port in [MSG_PORT_BASE, MSG_PORT_BASE+1000).
@@ -57,17 +74,19 @@ def __init__(self, node_id, name, port=None, *, mesh_id=DEFAULT_MESH_ID):
         self.mesh_id = mesh_id
         self.peers = {}  # node_id -> {"name": str, "addr": (ip, port), "last_seen": float}
         self.on_message = None  # callback(src_id, src_name, dst_id, text, hop_count, msg_id)
+        self.on_ack = None      # callback(msg_id, from_id, from_name)
         self.running = False
 
-        # Bounded de-dup ring. The set is the membership index; the deque
-        # is the FIFO eviction order. Both are maintained in lockstep via
-        # ``_remember_msg_id`` under ``_seen_lock`` so the listener thread
-        # and the sender thread can't race them out of sync.
+        # Bounded de-dup ring.
         self.seen_msg_ids: set[str] = set()
         self._seen_order: deque[str] = deque()
         self._seen_lock = threading.Lock()
 
-        # message socket — receives direct messages
+        # ACK tracking: msg_id -> {"packet": bytes, "attempts": int, "time": float, "acked_by": set}
+        self._pending_acks: dict[str, dict] = {}
+        self._ack_lock = threading.Lock()
+
+        # message socket — receives direct messages + ACKs
         self.msg_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
         self.msg_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
@@ -78,13 +97,7 @@ def __init__(self, node_id, name, port=None, *, mesh_id=DEFAULT_MESH_ID):
         self.bc_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
 
     def _remember_msg_id(self, mid: str) -> None:
-        """Record a msg_id in the bounded de-dup ring.
-
-        Idempotent — re-adding an already-known id is a no-op (the entry
-        is not re-ordered to the back of the FIFO; we deliberately do
-        *not* implement LRU semantics, the goal here is leak prevention,
-        not optimal cache hit rate).
-        """
+        """Record a msg_id in the bounded de-dup ring."""
         with self._seen_lock:
             if mid in self.seen_msg_ids:
                 return
@@ -102,6 +115,7 @@ def start(self):
         threading.Thread(target=self._listen_messages, daemon=True).start()
         threading.Thread(target=self._listen_discovery, daemon=True).start()
         threading.Thread(target=self._send_heartbeats, daemon=True).start()
+        threading.Thread(target=self._retry_unacked, daemon=True).start()
 
         # announce ourselves immediately
         self._broadcast_discovery()
@@ -128,10 +142,26 @@ def send(self, dst_id, text, msg_id=None, hop_count=0, src_id=None, max_hops=3):
             "mesh_id": self.mesh_id,
         }).encode()
 
-        # send to all known peers (flood-based mesh relay)
+        # Track for ACK (only for messages we originate, not relays)
+        if src_id == self.node_id and hop_count == 0:
+            with self._ack_lock:
+                self._pending_acks[msg_id] = {
+                    "packet": packet,
+                    "attempts": 1,
+                    "time": time.time(),
+                    "dst_id": dst_id,
+                    "acked_by": set(),
+                }
+
+        self._flood_packet(packet)
+
+    def broadcast(self, text):
+        self.send("^all", text)
+
+    def _flood_packet(self, packet: bytes):
+        """Send packet to all known peers."""
         for peer_id, peer in self.peers.items():
             addr = peer["addr"]
-            # try both the discovered IP and localhost (for same-machine nodes)
             targets = [addr]
             if addr[0] != "127.0.0.1":
                 targets.append(("127.0.0.1", addr[1]))
@@ -141,8 +171,87 @@ def send(self, dst_id, text, msg_id=None, hop_count=0, src_id=None, max_hops=3):
                 except OSError:
                     pass
 
-    def broadcast(self, text):
-        self.send("^all", text)
+    def _send_ack(self, msg_id: str, src_id: str):
+        """Send an ACK back to the message originator."""
+        ack_packet = json.dumps({
+            "type": "ack",
+            "ack_msg_id": msg_id,
+            "from_id": self.node_id,
+            "from_name": self.name,
+            "mesh_id": self.mesh_id,
+        }).encode()
+
+        # Send to the source peer if we know them
+        if src_id in self.peers:
+            addr = self.peers[src_id]["addr"]
+            targets = [addr]
+            if addr[0] != "127.0.0.1":
+                targets.append(("127.0.0.1", addr[1]))
+            for target in targets:
+                try:
+                    self.msg_sock.sendto(ack_packet, target)
+                except OSError:
+                    pass
+        else:
+            # Don't know the peer directly — flood the ACK
+            self._flood_packet(ack_packet)
+
+    def _handle_ack(self, packet: dict):
+        """Process an incoming ACK."""
+        ack_msg_id = packet.get("ack_msg_id")
+        from_id = packet.get("from_id")
+        from_name = packet.get("from_name", from_id)
+
+        if not ack_msg_id:
+            return
+
+        with self._ack_lock:
+            if ack_msg_id in self._pending_acks:
+                entry = self._pending_acks[ack_msg_id]
+                entry["acked_by"].add(from_id)
+                dst_id = entry["dst_id"]
+
+                # For direct messages, one ACK is enough
+                # For broadcasts, we keep collecting but don't retry once we have any
+                if dst_id != "^all" or len(entry["acked_by"]) > 0:
+                    del self._pending_acks[ack_msg_id]
+
+        print(f"\n  [ACK] {from_name} ({from_id}) confirmed msg {ack_msg_id[:12]}...")
+        print("mesh> ", end="", flush=True)
+
+        if self.on_ack:
+            self.on_ack(ack_msg_id, from_id, from_name)
+
+    def _retry_unacked(self):
+        """Background thread: retry messages that haven't been ACKed."""
+        while self.running:
+            time.sleep(1.0)
+            now = time.time()
+            to_retry = []
+            to_drop = []
+
+            with self._ack_lock:
+                for msg_id, entry in self._pending_acks.items():
+                    elapsed = now - entry["time"]
+                    if elapsed >= ACK_TIMEOUT_S:
+                        if entry["attempts"] < MAX_RETRIES:
+                            entry["attempts"] += 1
+                            entry["time"] = now
+                            to_retry.append((msg_id, entry["packet"], entry["attempts"]))
+                        else:
+                            to_drop.append(msg_id)
+
+                for msg_id in to_drop:
+                    del self._pending_acks[msg_id]
+
+            for msg_id, packet, attempt in to_retry:
+                print(f"\n  [RETRY] msg {msg_id[:12]}... attempt {attempt}/{MAX_RETRIES}")
+                print("mesh> ", end="", flush=True)
+                self._flood_packet(packet)
+
+            for msg_id in to_drop:
+                print(f"\n  [FAIL] msg {msg_id[:12]}... no ACK after {MAX_RETRIES} attempts")
+                print("mesh> ", end="", flush=True)
 
     def _broadcast_discovery(self):
         packet = json.dumps({
@@ -179,9 +288,6 @@ def _listen_discovery(self):
                     continue
                 if packet.get("node_id") == self.node_id:
                     continue
-                # Charlie filter: drop foreign meshes. A packet without the
-                # field is treated as "beacon-default" for backward-compat
-                # with pre-Wave-1 peers.
                 if packet.get("mesh_id", DEFAULT_MESH_ID) != self.mesh_id:
                     continue
                 peer_id = packet["node_id"]
@@ -202,11 +308,21 @@ def _listen_messages(self):
             try:
                 data, addr = self.msg_sock.recvfrom(BUFFER_SIZE)
                 packet = json.loads(data.decode())
-                if packet.get("type") != "msg":
+                pkt_type = packet.get("type")
+
+                # Handle ACKs
+                if pkt_type == "ack":
+                    if packet.get("mesh_id", DEFAULT_MESH_ID) != self.mesh_id:
+                        continue
+                    if packet.get("from_id") == self.node_id:
+                        continue  # ignore our own ACKs
+                    self._handle_ack(packet)
                     continue
 
-                # Charlie filter on msg packets too — same backward-compat
-                # default as discovery.
+                if pkt_type != "msg":
+                    continue
+
+                # Charlie filter
                 if packet.get("mesh_id", DEFAULT_MESH_ID) != self.mesh_id:
                     continue
 
@@ -216,17 +332,21 @@ def _listen_messages(self):
                 self._remember_msg_id(msg_id)
 
                 dst_id = packet["dst_id"]
+                src_id = packet["src_id"]
+
                 # deliver if for us or broadcast
                 if dst_id == self.node_id or dst_id == "^all":
                     if self.on_message:
                         self.on_message(
-                            packet["src_id"],
+                            src_id,
                             packet["src_name"],
                             dst_id,
                             packet["text"],
                             packet["hop_count"],
                             msg_id,
                         )
+                    # Send ACK back to originator
+                    self._send_ack(msg_id, src_id)
 
                 # relay if hops remain
                 if packet["hop_count"] < packet["max_hops"]:
@@ -235,7 +355,7 @@ def _listen_messages(self):
                         text=packet["text"],
                         msg_id=msg_id,
                         hop_count=packet["hop_count"] + 1,
-                        src_id=packet["src_id"],
+                        src_id=src_id,
                         max_hops=packet["max_hops"],
                     )
             except OSError: