fix: stop a stream on close, count only real users, record where a node is

**Closing the viewer left the node working.** Nothing told it to stop:
the player dropped its handlers, which only made the browser deaf. ffmpeg
kept running and held one of the node's two transcode slots until the
credit timeout expired two minutes later — which is why the next video
answered "server busy". `stream_stop` ends it at once, and the viewer
also drops its queue, ends the MediaSource and revokes the object URL on
the way out, any of which could be holding megabytes of decrypted video.

While there: `file_chunk` replies were matched to their requests by
arrival order, which was true by luck rather than by construction. The
reply now names the file it belongs to and is matched on that and the
chunk index; a chunk nobody is waiting for is dropped instead of being
handed to whatever request happens to be oldest.

**The administration panel counted its own history.** A deleted account
is tombstoned so the connection log stays readable, and every count and
list treated that row as a user — including a group's member count, and
the member list of the group itself. They do not any more.

**Where a node is.** `endpoint_hint` is what a node believes its address
to be, learned from a STUN server and sent to us: useful for reaching it,
and a claim. The announcement that carries it is signed with the node key
over a fresh timestamp, so the address that request *arrives from* is the
address of whoever holds that key — that is now recorded on the node row
and shown in a Nodes tab, next to the hint, with the difference spelled
out. Clients get the same treatment: `webrtc_offer` is logged with the
address the hub saw when a browser starts a peer connection.

Verified against the live deployment: the node's row reads 90.112.206.172
after a restart, and in e2e a stopped stream goes quiet in one message
and the next one starts immediately instead of being refused.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

14 files changed, 448 insertions, 16 deletions

diff --git a/docs/USERGUIDE.md b/docs/USERGUIDE.md
index 7242dc2..8a55928 100644
--- a/docs/USERGUIDE.md
+++ b/docs/USERGUIDE.md
@@ -742,6 +742,11 @@ player consumes it a segment at a time. Segments are 256 KB and the web client
 keeps 24 outstanding, so roughly 6 MB is in flight whatever the film's length. A
 client that sends no credit count gets the old unpaced behaviour.
 
+**Closing the viewer stops the stream.** `stream_stop` tells the node nobody is
+watching, so ffmpeg is killed and its transcode slot released at once. There are
+two slots; before this, leaving a video held one for the two-minute credit
+timeout, which is what made the next video answer "server busy".
+
 **The player drops what has been watched.** A SourceBuffer is not a file: browsers
 cap it and refuse the append that goes past, so anything more than a minute behind
 the playhead is evicted. A segment refused for want of room is retried rather than
@@ -826,6 +831,29 @@ If the keystore password is not stored on the node (`unlock_mode = "secure"`), a
 
 ---
 
+### Where a node is, in the administration panel
+
+The **Nodes** tab lists every registered node with two addresses, and the
+difference between them matters:
+
+- **Seen from** is the address the node's announcement arrived from. That request
+  carries an Ed25519 signature over a fresh timestamp made with the node key, so
+  the address belongs to whoever holds that key. It is IPv4 or IPv6, whichever
+  the node connected over, and it is the one to answer a question with.
+- **Announced hint** is what the node believes its own address to be, discovered
+  through a STUN server and sent to us. It is useful for reaching the node
+  directly and it is a claim, not evidence.
+
+Clients are recorded the same way: `webrtc_offer` in the log is written when a
+browser starts a peer connection, with the address the hub saw it come from.
+Whatever the two peers then discover through STUN is theirs to negotiate and does
+not belong in a log.
+
+Deleted accounts are not counted or listed anywhere in the panel. The tombstone
+row exists so the connection log stays readable (§2) and is not a user.
+
+---
+
 ## 9. Moderation and Legal
 
 ### Who is the legal host
diff --git a/packages/meshbay-common/src/meshbay_common/protocol.py b/packages/meshbay-common/src/meshbay_common/protocol.py
index 2972d58..b5a54ff 100644
--- a/packages/meshbay-common/src/meshbay_common/protocol.py
+++ b/packages/meshbay-common/src/meshbay_common/protocol.py
@@ -46,6 +46,7 @@ class MNP:
     STREAM_DATA      = "stream_data"       # node sends encrypted fMP4 segment
     STREAM_END       = "stream_end"        # node signals end of stream
     STREAM_MORE      = "stream_more"       # client → node: room for N more segments
+    STREAM_STOP      = "stream_stop"       # client → node: nobody is watching any more
     EPHEMERAL_STREAM = "ephemeral_stream"  # reserved — mobile live push
     HANDSHAKE_CHALLENGE  = "handshake_challenge"   # node → client: GEK proof nonce
     HANDSHAKE_RESPONSE   = "handshake_response"    # client → node: HMAC(GEK, nonce)
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/admin.py b/packages/meshbay-hub/src/meshbay_hub/api/admin.py
index 785731d..7ee05ca 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/admin.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/admin.py
@@ -15,7 +15,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from meshbay_hub.auth import decrypt_email
 from meshbay_hub.api.deps import require_admin, require_moderator
-from meshbay_hub.api.revocation import get_connected_node_count
+from meshbay_hub.api.revocation import get_connected_node_count, is_node_connected
 from meshbay_hub.db.engine import get_db
 from meshbay_hub.db.models import Group, GroupMember, IPLog, Node, User
 
@@ -42,8 +42,18 @@ async def admin_stats(
     current_user: User = Depends(require_moderator),
     db: AsyncSession = Depends(get_db),
 ):
-    user_count = (await db.execute(select(func.count()).select_from(User))).scalar_one()
-    group_count = (await db.execute(select(func.count()).select_from(Group))).scalar_one()
+    # Deleted accounts are tombstoned rather than dropped, so that the
+    # connection log stays readable. They are not users any more and must not be
+    # counted as any: a hub whose user count only ever rises is measuring its
+    # own history, not its population.
+    user_count = (await db.execute(
+        select(func.count()).select_from(User)
+        .where(User.status != "deleted"))).scalar_one()
+    # Groups are not tombstoned — deleting one removes the row — so every group
+    # here is a group. A revoked one is suspended by moderation and still shown
+    # in the list, so counting it keeps the two consistent.
+    group_count = (await db.execute(
+        select(func.count()).select_from(Group))).scalar_one()
     node_count = (await db.execute(select(func.count()).select_from(Node))).scalar_one()
     return {
         "users": user_count,
@@ -63,14 +73,16 @@ async def admin_list_users(
     offset: int = 0,
     limit: int = Query(default=50, le=200),
 ):
-    query = select(User).order_by(User.created_at.desc())
+    query = (select(User).where(User.status != "deleted")
+             .order_by(User.created_at.desc()))
     if q:
         query = query.where(User.username.ilike(f"%{q}%"))
     query = query.offset(offset).limit(limit)
     result = await db.execute(query)
     users = result.scalars().all()
 
-    total_query = select(func.count()).select_from(User)
+    total_query = (select(func.count()).select_from(User)
+                   .where(User.status != "deleted"))
     if q:
         total_query = total_query.where(User.username.ilike(f"%{q}%"))
     total = (await db.execute(total_query)).scalar_one()
@@ -222,9 +234,13 @@ async def admin_list_groups(
     query = (
         select(
             Group,
-            func.count(GroupMember.user_id).label("member_count"),
+            func.count(User.id).label("member_count"),
         )
+        # Members, not rows: a deleted account's membership is removed with it,
+        # but joining through User keeps the count honest if one ever survives.
         .outerjoin(GroupMember, Group.id == GroupMember.group_id)
+        .outerjoin(User, (User.id == GroupMember.user_id)
+                   & (User.status != "deleted"))
         .group_by(Group.id)
         .order_by(Group.created_at.desc())
         .offset(offset)
@@ -288,6 +304,44 @@ async def admin_patch_group(
 
 # ── IP Audit Logs ────────────────────────────────────────────────────────────
 
+@router.get("/nodes")
+async def admin_list_nodes(
+    current_user: User = Depends(require_moderator),
+    db: AsyncSession = Depends(get_db),
+    limit: int = Query(default=100, le=200),
+):
+    """
+    Registered nodes, with the address the hub saw them announce from.
+
+    `observed_ip` is the one to answer a question with: it comes from the
+    connection that carried a valid Ed25519 signature over a fresh timestamp, so
+    it is the address of whoever holds the node key. `endpoint_hint` is what the
+    node believes its own address to be, discovered through a STUN server and
+    sent to us — useful for reaching it, and not evidence of anything.
+    """
+    rows = (await db.execute(
+        select(Node, User.username)
+        .outerjoin(User, User.id == Node.user_id)
+        .order_by(Node.announced_at.desc())
+        .limit(limit))).all()
+    return {
+        "nodes": [
+            {
+                "id": n.id,
+                "user_id": n.user_id,
+                "username": uname or "",
+                "pk_node": n.pk_node,
+                "observed_ip": n.observed_ip or "",
+                "endpoint_hint": n.endpoint_hint or "",
+                "last_seen": n.last_seen.isoformat() if n.last_seen else "",
+                "announced_at": n.announced_at.isoformat(),
+                "online": is_node_connected(n.id),
+            }
+            for n, uname in rows
+        ],
+    }
+
+
 @router.get("/logs")
 async def admin_list_logs(
     current_user: User = Depends(require_moderator),
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
index 74c6c9e..8e3197c 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py
@@ -207,7 +207,7 @@ async def group_members(
     result = await db.execute(
         select(User.id, User.username)
         .join(GroupMember, User.id == GroupMember.user_id)
-        .where(GroupMember.group_id == group_id)
+        .where(GroupMember.group_id == group_id, User.status != "deleted")
     )
     members = [{"user_id": uid, "username": uname} for uid, uname in result.all()]
     return {
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/nodes.py b/packages/meshbay-hub/src/meshbay_hub/api/nodes.py
index 0770148..67e65f2 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/nodes.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/nodes.py
@@ -1,5 +1,6 @@
 """Node endpoints — /v1/nodes/*"""
 
+from datetime import datetime, timezone
 import base64
 import time
 
@@ -130,10 +131,18 @@ async def announce_node(
         select(Node).where(Node.user_id == current_user.id,
                            Node.pk_node == body.pk_node))
     node = existing.scalar_one_or_none()
+    # The address is taken from the connection, never from the body: the
+    # signature above proves who is announcing, and this is where they are
+    # announcing from. What the node believes its address to be — endpoint_hint,
+    # learned from a STUN server — is kept separately and is not evidence.
+    seen_from = client_ip(request)
+
     if node is not None:
         node.endpoint_hint = body.endpoint_hint
+        node.observed_ip = seen_from
+        node.last_seen = datetime.now(timezone.utc)
         db.add(IPLog(user_id=current_user.id, event="node_announce",
-                     ip_address=client_ip(request), detail=body.endpoint_hint))
+                     ip_address=seen_from, detail=body.endpoint_hint))
         await db.commit()
         return {"node_id": node.id}
 
@@ -141,12 +150,14 @@ async def announce_node(
         user_id=current_user.id,
         pk_node=body.pk_node,
         endpoint_hint=body.endpoint_hint,
+        observed_ip=seen_from,
+        last_seen=datetime.now(timezone.utc),
     )
     db.add(node)
     db.add(IPLog(
         user_id=current_user.id,
         event="node_announce",
-        ip_address=client_ip(request),
+        ip_address=seen_from,
         detail=body.endpoint_hint,
     ))
     await db.commit()
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
index 58ebf50..1f6d7f0 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py
@@ -55,6 +55,10 @@ _node_groups: dict[str, list[str]] = {}        # node_id → [group_id, ...]
 _punch_events: dict[str, asyncio.Event] = {}  # node_id → signaling event
 
 
+def is_node_connected(node_id: str) -> bool:
+    return node_id in _connected_nodes
+
+
 def get_connected_node_count() -> int:
     return len(_connected_nodes)
 
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
index 8f84163..cb00a67 100644
--- a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
+++ b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py
@@ -25,7 +25,8 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from meshbay_hub.api.deps import get_current_user
 from meshbay_hub.api.middleware import limiter
 from meshbay_hub.db.engine import get_db
-from meshbay_hub.db.models import Group, GroupMember, User
+from meshbay_hub.api.netutil import client_ip
+from meshbay_hub.db.models import Group, GroupMember, IPLog, User
 
 log = logging.getLogger(__name__)
 
@@ -77,6 +78,14 @@ async def webrtc_offer(
     if len(body.sdp) > MAX_SDP_BYTES:
         raise HTTPException(status_code=413, detail="SDP too large")
 
+    # Logged here because this is the moment a browser starts a peer connection,
+    # and the address it starts it from is this one — the hub's own view of the
+    # TCP connection. Whatever address the peers then discover through STUN is
+    # theirs to negotiate and is not what a log should record.
+    db.add(IPLog(user_id=current_user.id, event="webrtc_offer",
+                 ip_address=client_ip(request), detail=node_id[:8]))
+    await db.commit()
+
     ws = _connected_nodes.get(node_id)
     if not ws:
         raise HTTPException(status_code=404, detail="Node not connected")
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/d1f47a90c3b2_node_observed_ip.py b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/d1f47a90c3b2_node_observed_ip.py
new file mode 100644
index 0000000..c720cab
--- /dev/null
+++ b/packages/meshbay-hub/src/meshbay_hub/db/migrations/versions/d1f47a90c3b2_node_observed_ip.py
@@ -0,0 +1,35 @@
+"""node_observed_ip
+
+Where a node was actually seen, as opposed to where it says it is.
+
+`endpoint_hint` is discovered by the node through a STUN server and sent to the
+hub in its announcement — useful for reaching it, and a claim. `observed_ip` is
+the address the announcement itself arrived from, on a request carrying a valid
+Ed25519 signature over a fresh timestamp, so it is the address of whoever holds
+the node key. That is the one an administrator's question is about.
+
+Revision ID: d1f47a90c3b2
+Revises: c5e93b1a2f60
+Create Date: 2026-08-15
+
+"""
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = 'd1f47a90c3b2'
+down_revision: Union[str, Sequence[str], None] = 'c5e93b1a2f60'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column('nodes', sa.Column('observed_ip', sa.String(45), nullable=True))
+    op.add_column('nodes',
+                  sa.Column('last_seen', sa.DateTime(timezone=True), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column('nodes', 'last_seen')
+    op.drop_column('nodes', 'observed_ip')
diff --git a/packages/meshbay-hub/src/meshbay_hub/db/models.py b/packages/meshbay-hub/src/meshbay_hub/db/models.py
index 2e36989..7d5a3e3 100644
--- a/packages/meshbay-hub/src/meshbay_hub/db/models.py
+++ b/packages/meshbay-hub/src/meshbay_hub/db/models.py
@@ -72,6 +72,13 @@ class Node(Base):
     user_id:       Mapped[str] = mapped_column(ForeignKey("users.id"), nullable=False)
     pk_node:       Mapped[str] = mapped_column(String(64), nullable=False)  # Ed25519 b64
     endpoint_hint: Mapped[str | None] = mapped_column(String(128))          # "ip:port" or null
+    # The address the hub saw this node connect from, recorded on an announce
+    # that carried a valid Ed25519 signature over a fresh timestamp. Unlike
+    # endpoint_hint — which the node discovers through STUN and sends us — this
+    # is observed rather than claimed, and it is the one to answer questions
+    # with. IPv4 or IPv6.
+    observed_ip:   Mapped[str | None] = mapped_column(String(45))
+    last_seen:     Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
     announced_at:  Mapped[datetime] = mapped_column(DateTime(timezone=True), default=_now)
 
     user: Mapped["User"] = relationship(back_populates="nodes")
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/app.js b/packages/meshbay-hub/src/meshbay_hub/static/app.js
index 18ce83a..cf33d72 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/app.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/app.js
@@ -2587,10 +2587,25 @@ function VideoPlayer({ entry, transportRef, gekRef, onClose }) {
         videoRef.current.removeEventListener('seeking', onSeeking);
       }
       if (transport) {
+        // Tell the node first: dropping the handlers only makes us deaf, and a
+        // stream nobody is listening to still occupies a transcode slot.
+        transport.stopStream();
         transport.onStreamInit = null;
         transport.onStreamData = null;
         transport.onStreamEnd = null;
       }
+      // The queue can hold several megabytes of decrypted video.
+      queueRef.current = [];
+      const ms = msRef.current;
+      if (ms && ms.readyState === 'open') {
+        try { ms.endOfStream(); } catch { /* already ended */ }
+      }
+      if (blobUrlRef.current) {
+        URL.revokeObjectURL(blobUrlRef.current);
+        blobUrlRef.current = null;
+      }
+      sbRef.current = null;
+      msRef.current = null;
     };
   }, [entry, flushQueue]);
 
@@ -3056,6 +3071,7 @@ function AdminPage({ token }) {
   const [logEvent, setLogEvent] = useState('');
   const [logOffset, setLogOffset] = useState(0);
   const [blocklist, setBlocklist] = useState([]);
+  const [nodes, setNodes] = useState([]);
   const [detailUser, setDetailUser] = useState(null);
   const [error, setError] = useState('');
 
@@ -3105,6 +3121,10 @@ function AdminPage({ token }) {
     if (tab === 'stats') loadStats();
     else if (tab === 'users') loadUsers(userSearch);
     else if (tab === 'groups') loadGroups();
+    else if (tab === 'nodes') {
+      hubFetch('/v1/admin/nodes', { token })
+        .then(d => setNodes(d.nodes || [])).catch(e => setError(e.message));
+    }
     else if (tab === 'logs') { setLogOffset(0); loadLogs(logEvent, 0); }
     else if (tab === 'blocklist') loadBlocklist();
   }, [tab]);
@@ -3157,7 +3177,7 @@ function AdminPage({ token }) {
     } catch (e) { setError(e.message); }
   }, [token]);
 
-  const TABS = ['stats', 'users', 'groups', 'logs', 'blocklist'];
+  const TABS = ['stats', 'users', 'groups', 'nodes', 'logs', 'blocklist'];
 
   return html`
     <div>
@@ -3267,6 +3287,41 @@ function AdminPage({ token }) {
         </table>
       `}
 
+      ${tab === 'nodes' && html`
+        <p class="settings-hint" style="margin-bottom:10px">
+          ${t('admin.nodes_hint')}
+        </p>
+        <table class="admin-table">
+          <thead><tr>
+            <th>${t('admin.col_username')}</th>
+            <th>${t('admin.col_observed_ip')}</th>
+            <th>${t('admin.col_hint')}</th>
+            <th>${t('admin.col_last_seen')}</th>
+            <th>${t('admin.col_status')}</th>
+          </tr></thead>
+          <tbody>
+            ${nodes.length === 0 && html`
+              <tr><td colspan="5" class="admin-empty">${t('admin.no_nodes')}</td></tr>
+            `}
+            ${nodes.map(n => html`
+              <tr key=${n.id}>
+                <td>${n.username || n.user_id.slice(0, 8)}</td>
+                <td style="font-family:monospace">${n.observed_ip || '—'}</td>
+                <td style="font-family:monospace;color:var(--text-dim)">
+                  ${n.endpoint_hint || '—'}
+                </td>
+                <td>${n.last_seen ? new Date(n.last_seen).toLocaleString() : '—'}</td>
+                <td>
+                  <span class="badge ${n.online ? 'badge-ok' : ''}">
+                    ${n.online ? t('admin.node_online') : t('admin.node_offline')}
+                  </span>
+                </td>
+              </tr>
+            `)}
+          </tbody>
+        </table>
+      `}
+
       ${tab === 'logs' && html`
         <div class="admin-toolbar">
           <select class="admin-select" value=${logEvent} onChange=${e => {
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/i18n.js b/packages/meshbay-hub/src/meshbay_hub/static/i18n.js
index 71f6cce..affa7c8 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/i18n.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/i18n.js
@@ -195,7 +195,19 @@ const en = {
   'admin.tab_stats': 'Stats',
   'admin.tab_users': 'Users',
   'admin.tab_groups': 'Groups',
+  'admin.tab_nodes': 'Nodes',
   'admin.tab_logs': 'Logs',
+  'admin.col_observed_ip': 'Seen from',
+  'admin.col_hint': 'Announced hint',
+  'admin.col_last_seen': 'Last announce',
+  'admin.node_online': 'online',
+  'admin.node_offline': 'offline',
+  'admin.no_nodes': 'No nodes registered',
+  'admin.nodes_hint': '"Seen from" is the address the announcement arrived from, '
+    + 'on a request signed with the node key — that is the one to answer a '
+    + 'question with. The hint is what the node believes its own address to be, '
+    + 'learned from a STUN server and sent to us; it is useful for reaching the '
+    + 'node and is not evidence of anything.',
   'admin.tab_blocklist': 'Blocklist',
 
   // Admin stats
diff --git a/packages/meshbay-hub/src/meshbay_hub/static/transport.js b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
index 844a201..df2abb0 100644
--- a/packages/meshbay-hub/src/meshbay_hub/static/transport.js
+++ b/packages/meshbay-hub/src/meshbay_hub/static/transport.js
@@ -562,6 +562,18 @@ class MeshBayTransport {
   }
 
   /**
+   * Nobody is watching any more.
+   *
+   * Closing the viewer used to say nothing to the node, which went on
+   * transcoding and holding one of its two slots until the credit timeout — so
+   * the next video answered "server busy".
+   */
+  stopStream() {
+    if (!this._connected) return;
+    try { this._send({ type: 'stream_stop', v: '0.1' }); } catch { /* gone */ }
+  }
+
+  /**
    * Push a whole file, several chunks in flight at once.
    *
    * One chunk per round trip is 48 KB of throughput per RTT no matter how much
@@ -778,6 +790,11 @@ class MeshBayTransport {
       }, 30000);
       this._pending.set(id, {
         _reqType: obj.type,
+        // Chunks are the one request that runs several at a time and can be
+        // interleaved with anything else on the channel. Matching them by
+        // arrival order was only ever true by luck; this makes it true.
+        _key: obj.type === 'file_req'
+          ? `chunk:${obj.file_id}:${obj.chunk_index}` : null,
         resolve: (msg) => { clearTimeout(timeout); this._pending.delete(id); resolve(msg); },
         reject: (err) => { clearTimeout(timeout); this._pending.delete(id); reject(err); },
       });
@@ -857,6 +874,24 @@ class MeshBayTransport {
       return;
     }
 
+    if (msg.type === 'file_chunk') {
+      const key = `chunk:${msg.file_id}:${msg.chunk_index}`;
+      for (const [, handler] of this._pending) {
+        // A node from before the reply carried a file_id: fall back to the
+        // index, which is still better than the oldest pending request.
+        const match = msg.file_id
+          ? handler._key === key
+          : handler._key && handler._key.endsWith(`:${msg.chunk_index}`);
+        if (match) {
+          handler.resolve(msg);
+          return;
+        }
+      }
+      // Nobody asked for it any more — a cancelled download, most likely. It
+      // must not be handed to whatever request happens to be waiting.
+      return;
+    }
+
     const oldest = this._pending.entries().next();
     if (!oldest.done) {
       const [, handler] = oldest.value;
diff --git a/packages/meshbay-hub/tests/test_admin_views.py b/packages/meshbay-hub/tests/test_admin_views.py
new file mode 100644
index 0000000..29f41d8
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_admin_views.py
@@ -0,0 +1,150 @@
+"""
+What the administration panel counts and lists.
+
+Deleting an account leaves a tombstone so the connection log stays readable
+(see test_account_deletion). That row is not a user any more, and every place
+that counts or lists people has to agree — a hub whose user count only ever
+rises is measuring its own history rather than its population.
+"""
+
+import base64
+import hashlib
+
+import pytest
+from sqlalchemy import select
+
+from meshbay_hub.db.models import User
+
+
+def _auth_key(password: str, username: str) -> str:
+    salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
+    return base64.b64encode(
+        hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()
+
+
+async def _user(client, username, password="a-long-enough-passphrase"):
+    await client.post("/v1/users/register", json={
+        "username": username, "email": f"{username}@example.com",
+        "auth_key": _auth_key(password, username)})
+    r = await client.post("/v1/users/login", json={
+        "username": username, "auth_key": _auth_key(password, username)})
+    return {"Authorization": f"Bearer {r.json()['access_token']}"}
+
+
+async def _admin(client, db_session, username="root"):
+    headers = await _user(client, username)
+    user = (await db_session.execute(
+        select(User).where(User.username == username))).scalar_one()
+    user.role = "admin"
+    await db_session.commit()
+    return headers
+
+
+@pytest.mark.asyncio
+async def test_a_deleted_account_stops_being_counted(client, db_session):
+    admin = await _admin(client, db_session)
+    leaver = await _user(client, "ghost")
+
+    before = (await client.get("/v1/admin/stats", headers=admin)).json()["users"]
+    await client.request("DELETE", "/v1/users/me", headers=leaver,
+                         json={"auth_key": _auth_key(
+                             "a-long-enough-passphrase", "ghost")})
+    after = (await client.get("/v1/admin/stats", headers=admin)).json()["users"]
+
+    assert after == before - 1, "a tombstone is still being counted as a user"
+
+
+@pytest.mark.asyncio
+async def test_a_deleted_account_is_not_listed(client, db_session):
+    admin = await _admin(client, db_session, "root2")
+    leaver = await _user(client, "vanishing")
+
+    await client.request("DELETE", "/v1/users/me", headers=leaver,
+                         json={"auth_key": _auth_key(
+                             "a-long-enough-passphrase", "vanishing")})
+
+    r = await client.get("/v1/admin/users", headers=admin)
+    names = [u["username"] for u in r.json()["users"]]
+    assert "vanishing" not in names
+    assert not any(n.startswith("deleted-") for n in names), \
+        "the tombstone is showing under its placeholder name"
+    assert r.json()["total"] == len(names)
+
+
+@pytest.mark.asyncio
+async def test_the_member_list_of_a_group_skips_them(client, db_session):
+    admin = await _admin(client, db_session, "root3")
+    owner = await _user(client, "host3")
+    leaver = await _user(client, "quitter")
+
+    g = await client.post("/v1/groups", json={"name": "party"}, headers=owner)
+    gid = g.json()["group_id"]
+    await client.post(f"/v1/groups/{gid}/members/quitter", json={}, headers=owner)
+
+    await client.request("DELETE", "/v1/users/me", headers=leaver,
+                         json={"auth_key": _auth_key(
+                             "a-long-enough-passphrase", "quitter")})
+
+    members = (await client.get(f"/v1/groups/{gid}/members",
+                                headers=owner)).json()["members"]
+    assert [m["username"] for m in members] == ["host3"]
+
+    groups = (await client.get("/v1/admin/groups", headers=admin)).json()["groups"]
+    party = next(g for g in groups if g["name"] == "party")
+    assert party["member_count"] == 1, \
+        f"the group still counts its departed member ({party['member_count']})"
+
+
+# ── Where a node actually is ────────────────────────────────────────────────
+
+@pytest.mark.asyncio
+async def test_a_node_is_recorded_at_the_address_it_announced_from(
+        client, db_session):
+    """
+    The hub sees the connection; the node only tells us what a STUN server told
+    it. The observed address is the evidence, and the announcement carries an
+    Ed25519 signature over a fresh timestamp, so it is the address of whoever
+    holds the node key.
+    """
+    import base64
+    import time
+
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+    from cryptography.hazmat.primitives import serialization
+
+    from meshbay_hub.db.models import Node
+
+    admin = await _admin(client, db_session, "root4")
+    owner = await _user(client, "nodeowner")
+    uid = (await db_session.execute(
+        select(User.id).where(User.username == "nodeowner"))).scalar_one()
+
+    sk = Ed25519PrivateKey.generate()
+    pk_b64 = base64.b64encode(sk.public_key().public_bytes(
+        encoding=serialization.Encoding.Raw,
+        format=serialization.PublicFormat.Raw)).decode()
+    ts = int(time.time())
+    message = f"meshbay:node_announce:{uid}:{pk_b64}:{ts}".encode()
+
+    r = await client.post("/v1/nodes/announce", headers=owner, json={
+        "pk_node": pk_b64,
+        # A claim, and a deliberately implausible one.
+        "endpoint_hint": "203.0.113.42:19001",
+        "timestamp": ts,
+        "signature": base64.b64encode(sk.sign(message)).decode(),
+    })
+    assert r.status_code in (200, 201), r.text
+
+    node = (await db_session.execute(
+        select(Node).where(Node.pk_node == pk_b64))).scalar_one()
+    assert node.observed_ip, "no address was recorded for the announcement"
+    assert node.observed_ip != "203.0.113.42", \
+        "the hub recorded what the node claimed instead of what it saw"
+    assert node.last_seen is not None
+
+    listed = (await client.get("/v1/admin/nodes", headers=admin)).json()["nodes"]
+    row = next(n for n in listed if n["pk_node"] == pk_b64)
+    assert row["observed_ip"] == node.observed_ip
+    assert row["endpoint_hint"] == "203.0.113.42:19001", \
+        "the claim is still shown, just not as the answer"
+    assert row["username"] == "nodeowner"
diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
index eab1cac..a892be2 100644
--- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
+++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py
@@ -297,6 +297,7 @@ class WebRTCPeerSession:
         # Flow control for video: how many segments the client says it can take.
         self._stream_credit = 0
         self._stream_credit_evt = asyncio.Event()
+        self._stream_stopped = False
         self._gek_challenge: bytes | None = None
         # Same value as the GEK challenge, but kept for the life of the connection:
         # a join_request is signed over it, and it must stay verifiable after the
@@ -384,6 +385,8 @@ class WebRTCPeerSession:
                 asyncio.ensure_future(self._stream_video(msg))
             elif mtype == MNP.STREAM_MORE:
                 self._grant_stream_credit(msg)
+            elif mtype == MNP.STREAM_STOP:
+                self._stop_stream()
             else:
                 log.warning("Unknown MNP message type on DataChannel: %s", mtype)
         except Exception as e:
@@ -1203,6 +1206,7 @@ class WebRTCPeerSession:
             file_path,
             chunk_index,
             file_hash,
+            entry.id,
         )
         self._send(chunk_data)
         if chunk_index == 0:
@@ -1721,6 +1725,18 @@ class WebRTCPeerSession:
         self._stream_credit += max(0, min(n, STREAM_MAX_CREDIT))
         self._stream_credit_evt.set()
 
+    def _stop_stream(self) -> None:
+        """
+        The viewer was closed. Stop transcoding and let go of the slot.
+
+        Without this the only thing that ended a stream was the credit timeout,
+        so ffmpeg kept running and held one of the node's two transcode slots
+        for two minutes after nobody was watching — which is how closing a video
+        made the next one answer "server busy".
+        """
+        self._stream_stopped = True
+        self._stream_credit_evt.set()
+
     async def _await_stream_credit(self) -> bool:
         """
         Block until the client has room. False if it stopped asking.
@@ -1730,6 +1746,8 @@ class WebRTCPeerSession:
         JavaScript array while MediaSource consumes it a segment at a time.
         """
         while self._stream_credit <= 0:
+            if self._stream_stopped:
+                return False
             self._stream_credit_evt.clear()
             try:
                 await asyncio.wait_for(self._stream_credit_evt.wait(),
@@ -1738,6 +1756,8 @@ class WebRTCPeerSession:
                 log.info("Stream stalled: no credit from peer=%s",
                          (self._user_id or "?")[:8])
                 return False
+            if self._stream_stopped:
+                return False
             if self._channel is None or self._channel.readyState != "open":
                 return False
         self._stream_credit -= 1
@@ -1809,12 +1829,17 @@ class WebRTCPeerSession:
         except (TypeError, ValueError):
             self._stream_credit = 0
         paced = self._stream_credit > 0
+        self._stream_stopped = False
 
         index = 0
         try:
             while True:
                 if paced and not await self._await_stream_credit():
                     break
+                if self._stream_stopped:
+                    log.info("Stream stopped by peer=%s after %d segments",
+                             (self._user_id or "?")[:8], index)
+                    break
                 data = await proc.stdout.read(STREAM_SEGMENT_SIZE)
                 if not data:
                     break
@@ -1840,11 +1865,12 @@ class WebRTCPeerSession:
                 pass
             await proc.wait()
 
-        self._send({
-            "type": MNP.STREAM_END,
-            "v": MNP_VERSION,
-            "file_id": file_id,
-        })
+        if not self._stream_stopped:
+            self._send({
+                "type": MNP.STREAM_END,
+                "v": MNP_VERSION,
+                "file_id": file_id,
+            })
         log.info("Streamed %s: %d segments", entry.name, index)
         self._audit("stream_video", entry.name)
 
@@ -1868,6 +1894,7 @@ def _read_and_encrypt(
     file_path: Path,
     chunk_index: int,
     file_hash: bytes,
+    file_id: str = "",
 ) -> dict:
     with open(file_path, "rb") as f:
         f.seek(chunk_index * CHUNK_SIZE)
@@ -1879,6 +1906,10 @@ def _read_and_encrypt(
     return {
         "type": MNP.FILE_CHUNK,
         "v": MNP_VERSION,
+        # Named so a client running several downloads at once can tell whose
+        # reply this is. It used to carry only the index, which made matching a
+        # reply to its request a question of arrival order.
+        "file_id": file_id,
         "chunk_index": chunk_index,
         "plaintext_size": len(plaintext),
         "nonce": nonce,