1 files changed, 126 insertions, 11 deletions
diff --git a/src/api/dashboard/dashboard.html b/src/api/dashboard/dashboard.html
index d872e9b..05c233c 100644
--- a/src/api/dashboard/dashboard.html
+++ b/src/api/dashboard/dashboard.html
@@ -3076,7 +3076,11 @@ function trafficArrived() {
         received: messages(peer.traffic.received) });
     }
   }
-  if (lastTraffic && dots) addFlows(self, current, lastTraffic, () => false);
+  // What a subscribed node sent this one runs from its own messages, as it left
+  // (`remoteMessagesArrived`), not from what this one counted received.
+  if (lastTraffic && dots) addFlows(self, current, lastTraffic,
+    (way, id) => way === "received" && logging(id));
+  if (lastTraffic) deliveredHere(self, current, lastTraffic);
   measureRates(current);
   lastTraffic = current;
 }
@@ -3128,20 +3132,35 @@ function remoteMessagesArrived(id, entry, list) {
   const self = state.node.id;
   const wall = Date.now(), gap = wall - (entry.messagesAt || 0);
   entry.messagesAt = wall;
-  const kept = [];
+  // A tick's messages come whole, those since the last, up to the latest 500: one
+  // missing between (a busy tick) leaves what was delivered unknown for now.
+  if (list.length > 0) {
+    if (entry.lastSeq !== undefined && list[0].seq !== entry.lastSeq + 1) forgetDeliveries(id);
+    entry.lastSeq = list[list.length - 1].seq;
+  }
+  // For the console: its messages with others than this node (this node's own log has
+  // those), each message once. For the map: each message as it left, from its sender
+  // where the page reads the sender's messages (this node's own counts, or a
+  // subscribed node's), so it runs the way it went; else as it arrived.
+  const kept = [], running = [];
   for (const message of list) {
-    if (message.way === "refused" || message.peer === self) continue;
-    if (message.way === "received" && message.peer !== id && subscriptions.has(message.peer) &&
-        (subscribed.get(message.peer) || {}).messagesLive) continue;
+    if (message.way === "refused") continue;
     const [from, to] = message.way === "sent" ? [id, message.peer] : [message.peer, id];
-    kept.push({ ...message, seq: id + ":" + message.seq, node: id, from, to,
-      related: message.related ? id + ":" + message.related : null });
+    const made = { ...message, seq: id + ":" + message.seq, node: id, from, to,
+      related: message.related ? id + ":" + message.related : null };
+    if (message.way === "sent") noteDelivery(from, to, message.counted, "sent");
+    else noteDelivery(from, to, message.counted, "got");
+    if (message.peer !== self && !(message.way === "received" && message.peer !== id &&
+        logging(message.peer))) kept.push(made);
+    if (message.way === "sent" || !(message.peer === self || logging(message.peer))) {
+      running.push(made);
+    }
   }
-  if (kept.length === 0) return;
-  if (dotsNow(gap)) {
+  checkDeliveries();
+  if (dotsNow(gap) && running.length > 0) {
     const now = performance.now();
     const links = new Map();
-    for (const message of kept) {
+    for (const message of running) {
       const transport = schemeOf(message.address);
       const key = message.from + " " + message.to + " " + (transport || "");
       const link = links.get(key) ||
@@ -3161,7 +3180,103 @@ function remoteMessagesArrived(id, entry, list) {
     }
     if (flows.length > 0 && flowFrame === null) flowFrame = requestAnimationFrame(drawFlows);
   }
-  messagesArrived(kept);
+  if (kept.length > 0) messagesArrived(kept);
+}
+// Whether the page reads `id`'s messages one by one: a subscribed node's, while they
+// come.
+const logging = id => subscriptions.has(id) && (subscribed.get(id) || {}).messagesLive === true;
+
+// --- Messages that didn't arrive: where the page reads both ends of a link (this node's
+// counts, or a subscribed node's messages), what one sent the other should count
+// received. Sent but not received within `DELIVERY_GRACE_MS` (both come every 2 s, each
+// on its own tick), the link breathes red (`breatheLink`). By message, of every kind but
+// those the paths and identity layers count under the address rather than the node
+// (this node's counts leave them out). Kept from when both ends are read; forgotten
+// when either isn't any more, or a tick of either came short. ---
+const DELIVERY_GRACE_MS = 5000, UNCOUNTED = new Set(["paths", "identity"]);
+// By `from to`: `sent` and `got`, each a queue of [when, how many], the oldest first.
+const deliveries = new Map();
+const readHere = id => state !== null && (id === state.node.id || logging(id));
+function noteDelivery(from, to, kind, which, count = 1) {
+  if (UNCOUNTED.has(kind) || count <= 0) return;
+  const key = from + " " + to;
+  if (!flowsShown || !readHere(from) || !readHere(to)) { deliveries.delete(key); return; }
+  const record = deliveries.get(key) || { from, to, sent: [], got: [] };
+  record[which].push([performance.now(), count]);
+  deliveries.set(key, record);
+}
+function forgetDeliveries(id) {
+  for (const [key, record] of deliveries) {
+    if (record.from === id || record.to === id) deliveries.delete(key);
+  }
+}
+// This node's counts, as they grew since the last update: by peer, what it sent and
+// received, each kind counted.
+function deliveredHere(self, current, before) {
+  for (const [id, counts] of current) {
+    const was = before.get(id);
+    if (!was) continue;
+    for (const [way, from, to, which] of [["sent", self, id, "sent"], ["received", id, self, "got"]]) {
+      for (const [kind, count] of Object.entries(counts[way])) {
+        const earlier = was[way][kind] || { messages: 0 };
+        noteDelivery(from, to, kind, which, count.messages - earlier.messages);
+      }
+    }
+  }
+  checkDeliveries();
+}
+// Received messages settle the oldest sent first; what was sent longer ago than the
+// grace and is still unsettled didn't arrive. Received with nothing sent to settle (the
+// sender's tick not in yet) waits as long again, then goes.
+function checkDeliveries() {
+  const now = performance.now();
+  for (const [key, record] of deliveries) {
+    const { sent, got } = record;
+    while (sent.length > 0 && got.length > 0) {
+      const settled = Math.min(sent[0][1], got[0][1]);
+      sent[0][1] -= settled;
+      got[0][1] -= settled;
+      if (sent[0][1] === 0) sent.shift();
+      if (got[0][1] === 0) got.shift();
+    }
+    let lost = 0;
+    while (sent.length > 0 && sent[0][0] <= now - DELIVERY_GRACE_MS) lost += sent.shift()[1];
+    while (got.length > 0 && got[0][0] <= now - 2 * DELIVERY_GRACE_MS) got.shift();
+    if (lost > 0) breatheLink(record.from, record.to, lost);
+    if (sent.length === 0 && got.length === 0) deliveries.delete(key);
+  }
+}
+// A red halo along the link `from`–`to` (its line in use), in and out once: `lost`
+// messages sent over it didn't arrive. Below the flashes' layer's other breaths.
+const LOST_MS = 1200;
+function breatheLink(from, to, lost) {
+  if (document.hidden || !flowsShown) return;
+  const layer = document.getElementById("flashes");
+  const halo = svg("path", { fill: "none", stroke: "var(--bad)", "stroke-width": 10,
+    "stroke-linecap": "round", "pointer-events": "none" });
+  const title = svg("title");
+  title.textContent = lost + (lost === 1 ? " message " : " messages ") + name(from) +
+    " sent " + name(to) + " didn't arrive";
+  halo.append(title);
+  layer.prepend(halo);
+  const started = performance.now();
+  const step = now => {
+    const line = route(from, to, null);
+    const t = Math.min(1, Math.max(0, (now - started) / LOST_MS));
+    if (!line || t >= 1) {
+      halo.remove();
+      return;
+    }
+    const points = [];
+    for (let k = 0; k <= 12; k++) {
+      const point = line.at(k / 12);
+      points.push(point.x.toFixed(1) + " " + point.y.toFixed(1));
+    }
+    halo.setAttribute("d", "M" + points.join("L"));
+    halo.setAttribute("stroke-opacity", (.5 * Math.sin(Math.PI * t)).toFixed(2));
+    requestAnimationFrame(step);
+  };
+  requestAnimationFrame(step);
 }
 
 // --- The console's rates: messages and bytes per second sent and received, per peer