dashboard: messages run as they left; a link breathes red for those lost

A message between this node and a subscribed one ran from this node's
counts both ways: one the subscribed node sent ran once this node had
counted it received, as it arrived. Each message now runs from its
sender where the page reads the sender (this node's counts, a
subscribed node's messages), so it runs the way it went, also where it
never arrived; only one from a node the page doesn't read runs as it
arrived. The console lists them as before.

Where the page reads both ends of a link, what one sent the other
should count received: a message sent and not received within 5 s
makes the link breathe red once (`breatheLink`), its tooltip saying
how many. Kinds the paths and identity layers count under addresses
are left out, as this node's counts leave them out; a subscribed
node's tick that came short (the latest 500 of more) starts the count
afresh, so what the page didn't see isn't taken as lost.

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