Merge pull request #3012 from yueguobin/fix/marker-replay-remote-computes

fix: fetch remote-compute marker pcaps in tag replay

docs/features/marker-tag-replay.md

@@ -41,11 +41,16 @@ graph TB
     TMP["/tmp scratch copies<br/>(hardened-profile workaround)"]
     SK["sharkd -<br/>(resident JSON-RPC on stdio)"]
 
+    subgraph Remote["Remote compute"]
+        RFS[("its markers dir")]
+    end
+
     UI -->|"GET range / frames [?filter=&link=]"| GATE
     GATE --> SCAN
     SCAN -->|"columns / filter matches"| SESS
     SESS --> TMP --> SK
     SCAN -->|"hex: raw bytes"| FS
+    RFS -->|"HTTP fetch → local cache"| SCAN
     UI -->|"GET frame detail (lazy)"| SESS
 ```
 
@@ -76,6 +81,8 @@ and the user's home even though the server process can read both. The pcap copy
 off the event loop, and server shutdown kills every session so no scratch copy leaks
 into `/tmp` across restarts.
 
+Capture-file locality: a marker's pcap is written by uBridge on the compute hosting the capture node, so only local markers' pcaps sit in the controller's project directory. A remote marker's pcap is fetched through the compute's project-files API on every acquire into a bounded cache (32 idle files under `/tmp`, wiped at shutdown) — a paused marker's pcap holds matched packets only, so the refetch stays cheap. The cached copy is replaced only when the bytes changed, so an unchanged refetch keeps the previous file's mtime and the resident sharkd session keyed by it survives; for remote sources the per-request `(mtime, size)` validation runs against the cache copy. A compute that cannot be reached flags its source (`count: 0` plus an `error` message in `sources`) instead of silently shrinking the timeline; the frame-detail endpoint then answers 409 with the fetch failure.
+
 ### Resource bounds
 
 Three independent bounds, easily confused:
@@ -171,7 +178,7 @@ without it.
   "frame_count": 29,
   "sources": [
     { "node_id": "47703cad…", "link_id": "2697a7c6…", "marker": "global-ospf",
-      "data_link_type": "DLT_EN10MB", "count": 4 }
+      "data_link_type": "DLT_EN10MB", "count": 4, "error": null }
   ],
   "frames": [
     { "ts": "1788369209.406812", "len": 114,
@@ -218,7 +225,10 @@ design:
 
 - **`sources` is the stable inventory of the tag**: every capture source is always
   listed, with engine-free **total** counts, unaffected by `link` / `filter` — a
-  source dropdown must not shrink when the view narrows.
+  source dropdown must not shrink when the view narrows. A source whose capture file
+  could not be fetched from its compute keeps its entry with `count: 0` and an `error`
+  message (compute unreachable) — a partial timeline must be visible as partial, never
+  mistaken for the whole capture.
 - **An unknown `link_id` matches nothing**: `frame_count: 0`, `start: null`, empty
   `frames` — the same shape as a zero-match display filter, deliberately not a 404.
 
@@ -308,7 +318,7 @@ All error bodies are `{"message": "…"}` (the app's unified format).
 | 400 | Invalid display filter (sharkd's filter rejection only; message carries its original text) or filter longer than 2000 chars |
 | 401 | Not authenticated |
 | 404 | Tag has no markers in the project; detail source unknown; `ts`/`frame_number` do not match the file (the capture may have been rebuilt); or the pcap vanished mid-request |
-| 409 | Tag gate: a marker under the tag is still `enabled: true` (the response lists them) |
+| 409 | Tag gate: a marker under the tag is still `enabled: true` (the response lists them); detail endpoint only — the compute hosting the capture file could not be reached (message names the compute) |
 | 501 | sharkd not installed / unavailable — replay is unavailable, no degraded mode |
 | 502 | sharkd failed, timed out (10 s per RPC), or answered out of sync — the session is killed and re-spawned on the next request |
 
@@ -348,8 +358,4 @@ All error bodies are `{"message": "…"}` (the app's unified format).
   directory — nothing accumulates across restarts.
 - **Tag type.** REST and the `marker.match` WS event both carry `tag` as `int` (the
   listener normalizes); replay keys on that int value.
-- **Follow-ups.** Remote-compute support via the existing capture-file proxy pattern;
-  convenience APIs (`GET …/markers/tags` to list tags, `POST …/markers/tags/{tag}/pause`
-  to batch-pause — a one-call path to the replayable state); uBridge-side
-  `pcap_dump_open_append` (with a linktype-header check on the existing file) so capture
-  history survives node restarts instead of being truncated on every reinstall.
+- **Follow-ups.** Convenience APIs (`GET …/markers/tags` to list tags, `POST …/markers/tags/{tag}/pause` to batch-pause — a one-call path to the replayable state); uBridge-side `pcap_dump_open_append` (with a linktype-header check on the existing file) so capture history survives node restarts instead of being truncated on every reinstall.

gns3server/controller/marker_replay.py

@@ -51,9 +51,17 @@ merge sorts by ``(ts, source file, frame number)`` and index structures must
 never use ts alone as a dict key, or same-microsecond frames silently
 overwrite each other. The canonical ts strings travel to clients verbatim
 and must be round-tripped verbatim.
+
+Capture files live on the compute hosting the capture node — only a local
+marker's pcap sits in the controller's project directory. A remote marker's
+pcap is fetched from its compute over HTTP on every acquire into a bounded
+cache (paused marker pcaps hold matched packets only, so they stay small);
+the previous copy is kept when the bytes are identical so resident sharkd
+sessions (validated by ``(mtime, size)``) survive unchanged refetches.
 """
 
 import asyncio
+import filecmp
 import json
 import logging
 import os
@@ -63,7 +71,9 @@ import tempfile
 import time
 from contextlib import asynccontextmanager
 
-from .controller_error import ControllerBadRequestError, ControllerError, ControllerNotFoundError
+import aiofiles
+
+from .controller_error import ComputeError, ControllerBadRequestError, ControllerError, ControllerNotFoundError
 
 log = logging.getLogger(__name__)
 
@@ -202,6 +212,173 @@ def read_frame_bytes(path, frame_number):
         return f.read(incl_len).hex()
 
 
+# ---------------------------------------------------------------------------
+# Remote capture sources (fetched from their compute over HTTP)
+# ---------------------------------------------------------------------------
+
+# Bounded number of remote marker pcaps kept on the controller. Only idle
+# entries are ever evicted (see _remote_pcap), so a request's own sources
+# cannot be pulled out from under it.
+_REMOTE_CACHE_MAX = 32
+
+# (project_id, compute_id, rel_path) -> {"path": str, "last_used": float, "_uses": int}
+_remote_cache: dict[tuple[str, str, str], dict] = {}
+_remote_locks: dict[tuple[str, str, str], asyncio.Lock] = {}
+_remote_cache_dir: str | None = None
+
+
+def _cache_path(project, compute, rel_path):
+    """Flat, collision-free cache path: project/compute ids are uuids and
+    basename() strips any path component out of the file name."""
+
+    global _remote_cache_dir
+    if _remote_cache_dir is None:
+        _remote_cache_dir = tempfile.mkdtemp(prefix="gns3-marker-replay-")
+    os.makedirs(_remote_cache_dir, exist_ok=True)
+    return os.path.join(_remote_cache_dir, f"{project.id}_{compute.id}_{os.path.basename(rel_path)}")
+
+
+def _drop_cached_file(key):
+    """Forget a cache entry and delete its file (marker removed on the compute)."""
+
+    entry = _remote_cache.pop(key, None)
+    if entry and entry["path"]:
+        try:
+            os.remove(entry["path"])
+        except OSError:
+            pass
+
+
+def _evict_idle_remote_cache():
+    """Trim the cache down to the cap, least-recently-used first. Only idle
+    entries are evictable; dict access is synchronous, so this cannot
+    interleave with a concurrent acquire's refcount bump."""
+
+    while len(_remote_cache) > _REMOTE_CACHE_MAX:
+        idle = [key for key, entry in _remote_cache.items() if entry["_uses"] == 0]
+        if not idle:
+            break
+        victim = min(idle, key=lambda key: _remote_cache[key]["last_used"])
+        entry = _remote_cache.pop(victim)
+        if entry["path"]:
+            try:
+                os.remove(entry["path"])
+            except OSError:
+                pass
+
+
+async def _fetch_to_cache(project, compute, rel_path, path):
+    """
+    Fetch the pcap from its compute, replacing the cached copy only when the
+    bytes changed — an unchanged copy keeps its mtime, so the resident sharkd
+    session keyed by that path stays valid. The file is refetched on every
+    acquire: the gate already pins replay to paused markers, and a marker
+    resumed and re-paused between two requests rebuilds its pcap.
+
+    :returns: the cache path, or None when the compute has no such file (the
+        marker never matched anything — not an error).
+    :raises ComputeError: the compute could not be reached or refused the file.
+    """
+
+    try:
+        response = await compute.http_query("GET", f"/projects/{project.id}/files/{rel_path}", stream=True)
+    except ControllerNotFoundError:
+        return None
+    except ControllerError as e:
+        raise ComputeError(f"Could not fetch capture file from compute '{compute.id}': {e}") from e
+
+    tmp = f"{path}.tmp"
+    try:
+        async with aiofiles.open(tmp, "wb") as f:
+            async for chunk in response.content.iter_any():
+                if chunk:
+                    await f.write(chunk)
+        await response.release()
+        if os.path.exists(path) and filecmp.cmp(path, tmp, shallow=False):
+            os.remove(tmp)  # unchanged: keep the old copy, mtime intact
+        else:
+            os.replace(tmp, path)
+    finally:
+        if os.path.exists(tmp):
+            try:
+                os.remove(tmp)
+            except OSError:
+                pass
+    return path
+
+
+@asynccontextmanager
+async def _remote_pcap(project, compute, rel_path):
+    """
+    A controller-local copy of a capture file living on a remote compute.
+    Acquires of the same file are serialized per key; the entry is
+    refcounted while yielded so eviction only ever removes idle files.
+
+    Yields the local path, or None when the compute has no such file.
+    """
+
+    key = (project.id, compute.id, rel_path)
+    lock = _remote_locks.setdefault(key, asyncio.Lock())
+    async with lock:
+        entry = _remote_cache.get(key)
+        if entry is None:
+            entry = {"path": _cache_path(project, compute, rel_path), "last_used": 0.0, "_uses": 0}
+            _remote_cache[key] = entry
+        entry["_uses"] += 1
+        try:
+            path = await _fetch_to_cache(project, compute, rel_path, entry["path"])
+            entry["last_used"] = time.monotonic()
+            if path is None:
+                _drop_cached_file(key)
+            yield path
+        finally:
+            entry["_uses"] -= 1
+    _evict_idle_remote_cache()
+
+
+def _capture_node_compute(project, node_id):
+    """
+    The compute hosting a marker's capture node, or None when the capture is
+    local — or the node is gone, in which case only the controller's own
+    markers directory can hold a copy and the caller has already checked it.
+    """
+
+    get_node = getattr(project, "get_node", None)
+    if get_node is None:
+        return None
+    try:
+        compute = get_node(node_id).compute
+    except ControllerError:
+        return None
+    if compute is None or compute.id == "local":
+        return None
+    return compute
+
+
+@asynccontextmanager
+async def _source_pcap(project, node_id, link_id, marker):
+    """
+    Resolve a marker's capture file to a controller-local path.
+
+    Yields None when no capture exists (no packet ever matched on that
+    marker). Local markers resolve straight to the project's markers
+    directory; a marker whose capture node runs on another compute is
+    fetched from that compute (see _remote_pcap).
+    """
+
+    filename = f"{node_id}_{link_id}_{marker}.pcap"
+    path = os.path.join(project.markers_directory, filename)
+    if os.path.exists(path):
+        yield path
+        return
+    compute = _capture_node_compute(project, node_id)
+    if compute is None:
+        yield None
+        return
+    async with _remote_pcap(project, compute, f"project-files/markers/{filename}") as fetched:
+        yield fetched
+
+
 # ---------------------------------------------------------------------------
 # sharkd: process environment and scratch copies
 # ---------------------------------------------------------------------------
@@ -526,13 +703,19 @@ def _get_manager():
 
 async def close_sessions():
     """Server shutdown hook: kill every resident session and drop its scratch
-    copy, so nothing leaks into /tmp across restarts."""
+    copy, and remove the remote-capture cache, so nothing leaks into /tmp
+    across restarts."""
 
     global _manager
+    global _remote_cache_dir
     manager = _manager
     if manager is not None:
         await manager.close_all()
     _manager = None
+    if _remote_cache_dir is not None:
+        shutil.rmtree(_remote_cache_dir, ignore_errors=True)
+        _remote_cache_dir = None
+    _remote_cache.clear()
 
 
 # ---------------------------------------------------------------------------
@@ -655,45 +838,68 @@ async def _merged_frames(project, entries, filter_expr=None, link_id=None):
     ``sources`` is the stable inventory of the tag: EVERY capture source is
     listed with engine-free total counts, unaffected by ``link_id`` /
     ``filter_expr`` — the inventory must not shrink when the view narrows.
+    A capture file that cannot be fetched from its compute keeps its source
+    entry, with ``count: 0`` and an ``error`` message, instead of silently
+    vanishing from the timeline.
     """
 
-    markers_dir = project.markers_directory
     merged = []
     sources = []
     for entry in entries:
-        pcap = os.path.join(markers_dir, f"{entry['node_id']}_{entry['link_id']}_{entry['marker']}.pcap")
-        frames = scan_pcap_frames(pcap) if os.path.exists(pcap) else []
-        # Inventory first: every source, engine-free totals.
-        sources.append(
-            {**{k: entry[k] for k in ("node_id", "link_id", "marker", "data_link_type")}, "count": len(frames)}
-        )
-        if link_id and entry["link_id"] != link_id:
-            continue  # link narrows the stream before any engine work
-        if frames:
-            columns = await _columns_for(pcap, filter_expr)
-        else:
-            columns = {}
-        source_key = f"{entry['node_id']}_{entry['link_id']}_{entry['marker']}"
-        for frame_number, (sec, usec, incl_len) in enumerate(frames, start=1):
-            if filter_expr is not None and frame_number not in columns:
-                continue
-            cols = columns.get(frame_number, {})
-            merged.append(
+        try:
+            async with _source_pcap(project, entry["node_id"], entry["link_id"], entry["marker"]) as pcap:
+                frames = scan_pcap_frames(pcap) if pcap else []
+                # Inventory first: every source, engine-free totals.
+                sources.append(
+                    {
+                        **{k: entry[k] for k in ("node_id", "link_id", "marker", "data_link_type")},
+                        "count": len(frames),
+                    }
+                )
+                if link_id and entry["link_id"] != link_id:
+                    continue  # link narrows the stream before any engine work
+                if frames:
+                    columns = await _columns_for(pcap, filter_expr)
+                else:
+                    columns = {}
+                source_key = f"{entry['node_id']}_{entry['link_id']}_{entry['marker']}"
+                for frame_number, (sec, usec, incl_len) in enumerate(frames, start=1):
+                    if filter_expr is not None and frame_number not in columns:
+                        continue
+                    cols = columns.get(frame_number, {})
+                    merged.append(
+                        {
+                            "ts": _format_ts(sec, usec),
+                            "ts_us": sec * 1_000_000 + usec,
+                            "_source": source_key,
+                            "len": incl_len,
+                            "node_id": entry["node_id"],
+                            "link_id": entry["link_id"],
+                            "marker": entry["marker"],
+                            "frame_number": frame_number,
+                            "src": cols.get("src"),
+                            "dst": cols.get("dst"),
+                            "proto": cols.get("proto"),
+                            "info": cols.get("info"),
+                            "bg": cols.get("bg"),
+                            "fg": cols.get("fg"),
+                        }
+                    )
+        except ComputeError as e:
+            # The capture file lives on a compute that could not be reached:
+            # keep the source in the inventory, flagged, so a partial timeline
+            # is never mistaken for the whole capture.
+            log.warning(
+                "Capture file of marker '%s' on link %s could not be fetched: %s",
+                entry["marker"],
+                entry["link_id"],
+                e,
+            )
+            sources.append(
                 {
-                    "ts": _format_ts(sec, usec),
-                    "ts_us": sec * 1_000_000 + usec,
-                    "_source": source_key,
-                    "len": incl_len,
-                    "node_id": entry["node_id"],
-                    "link_id": entry["link_id"],
-                    "marker": entry["marker"],
-                    "frame_number": frame_number,
-                    "src": cols.get("src"),
-                    "dst": cols.get("dst"),
-                    "proto": cols.get("proto"),
-                    "info": cols.get("info"),
-                    "bg": cols.get("bg"),
-                    "fg": cols.get("fg"),
+                    **{k: entry[k] for k in ("node_id", "link_id", "marker", "data_link_type")},
+                    "count": 0,
+                    "error": str(e),
                 }
             )
     merged.sort(key=lambda f: (f["ts_us"], f["_source"], f["frame_number"]))
@@ -761,7 +967,10 @@ async def decode_frame(project, tag, ts, node_id, link_id, marker, frame_number=
     round-tripped ts still matches the file (guards a rebuild between the
     timeline view and this click), read the raw bytes for the hex view
     straight from the pcap, and rename the sharkd protocol tree into the
-    REST contract (closed key set, values untouched).
+    REST contract (closed key set, values untouched). A marker captured on
+    another compute decodes from its fetched cache copy; a compute that
+    cannot be reached raises ComputeError — distinct from the 404s below,
+    which mean the marker never matched anything or the capture was rebuilt.
 
     ``ts`` is not unique within one pcap (same-microsecond frames are kept
     deliberately); an explicit ``frame_number`` from the frame list entry
@@ -778,40 +987,40 @@ async def decode_frame(project, tag, ts, node_id, link_id, marker, frame_number=
     if entry is None:
         raise ControllerNotFoundError(f"No marker '{marker}' with tag {tag} on link {link_id} captured by {node_id}")
 
-    pcap = os.path.join(project.markers_directory, f"{node_id}_{link_id}_{marker}.pcap")
-    if not os.path.exists(pcap):
-        raise ControllerNotFoundError(f"No capture file for marker '{marker}' (nothing ever matched)")
+    async with _source_pcap(project, node_id, link_id, marker) as pcap:
+        if pcap is None:
+            raise ControllerNotFoundError(f"No capture file for marker '{marker}' (nothing ever matched)")
 
-    frames = scan_pcap_frames(pcap)
-    rebuilt_message = f"No frame at ts {ts} in marker '{marker}' (the capture may have been rebuilt)"
-    if frame_number is None:
-        # The ts must be the exact string the timeline returned; find the frame
-        # it identifies rather than trusting any position hint from the client.
-        frame_number = next(
-            (i for i, (sec, usec, _len) in enumerate(frames, start=1) if _format_ts(sec, usec) == ts),
-            None,
-        )
+        frames = scan_pcap_frames(pcap)
+        rebuilt_message = f"No frame at ts {ts} in marker '{marker}' (the capture may have been rebuilt)"
         if frame_number is None:
-            raise ControllerNotFoundError(rebuilt_message)
-    else:
-        if not 1 <= frame_number <= len(frames):
-            raise ControllerNotFoundError(rebuilt_message)
-        sec, usec, _len = frames[frame_number - 1]
-        if _format_ts(sec, usec) != ts:
+            # The ts must be the exact string the timeline returned; find the frame
+            # it identifies rather than trusting any position hint from the client.
+            frame_number = next(
+                (i for i, (sec, usec, _len) in enumerate(frames, start=1) if _format_ts(sec, usec) == ts),
+                None,
+            )
+            if frame_number is None:
+                raise ControllerNotFoundError(rebuilt_message)
+        else:
+            if not 1 <= frame_number <= len(frames):
+                raise ControllerNotFoundError(rebuilt_message)
+            sec, usec, _len = frames[frame_number - 1]
+            if _format_ts(sec, usec) != ts:
+                raise ControllerNotFoundError(rebuilt_message)
+
+        raw_hex = read_frame_bytes(pcap, frame_number)
+        if raw_hex is None:
+            # The header scan listed it but the bytes are gone — mid-write tail.
             raise ControllerNotFoundError(rebuilt_message)
 
-    raw_hex = read_frame_bytes(pcap, frame_number)
-    if raw_hex is None:
-        # The header scan listed it but the bytes are gone — mid-write tail.
-        raise ControllerNotFoundError(rebuilt_message)
-
-    async with _get_manager().session(pcap) as session:
-        try:
-            result = await session.rpc("frame", {"frame": frame_number, "proto": True})
-        except _SharkdRpcError as e:
-            # The frame range was validated against the file above, so an
-            # engine error here is a real fault (502), not a client 404.
-            raise SharkdError(f"sharkd frame failed: {e.message}")
+        async with _get_manager().session(pcap) as session:
+            try:
+                result = await session.rpc("frame", {"frame": frame_number, "proto": True})
+            except _SharkdRpcError as e:
+                # The frame range was validated against the file above, so an
+                # engine error here is a real fault (502), not a client 404.
+                raise SharkdError(f"sharkd frame failed: {e.message}")
 
     tree = _rename_value(result.get("tree", []))
     return {

gns3server/schemas/controller/marker_replay.py

@@ -29,6 +29,9 @@ class ReplaySource(BaseModel):
     marker: str
     data_link_type: str
     count: int = Field(..., description="Total frames in the capture, unaffected by filters")
+    error: Optional[str] = Field(
+        None, description="Set when the capture file could not be fetched from the compute hosting it"
+    )
 
 
 class ReplayFrame(BaseModel):

tests/api/routes/controller/test_response_models.py

@@ -458,7 +458,14 @@ class TestReplayFrameDetail:
             "end": None,
             "frame_count": 0,
             "sources": [
-                {"node_id": "n1", "link_id": link.id, "marker": "icmp", "data_link_type": "DLT_EN10MB", "count": 0}
+                {
+                    "node_id": "n1",
+                    "link_id": link.id,
+                    "marker": "icmp",
+                    "data_link_type": "DLT_EN10MB",
+                    "count": 0,
+                    "error": None,
+                }
             ],
             "frames": [],
         }

tests/controller/test_marker_replay.py

@@ -30,7 +30,13 @@ Unit tests for the tag-keyed aggregate replay module (sharkd edition):
   invalidation, LRU bound (idle-only eviction), single spawn under
   concurrent acquires, timeout/stale-reply aborts — against the real sharkd
   where installed, plus the frame detail end-to-end (hex + renamed tree +
-  filter expressions, same-microsecond disambiguation).
+  filter expressions, same-microsecond disambiguation)
+* remote capture sources — markers whose capture node runs on another
+  compute are fetched over HTTP into the bounded cache: merged timelines
+  span both computes, a compute with no file is a zero-count source (not an
+  error), an unreachable compute flags its source instead of shrinking the
+  inventory, unchanged refetches keep the cached file identity (resident
+  sharkd sessions survive), and the local compute is never queried.
 """
 
 import asyncio
@@ -38,6 +44,7 @@ import os
 import shutil
 import struct
 import sys
+from contextlib import asynccontextmanager
 from types import SimpleNamespace
 from unittest.mock import patch
 
@@ -45,6 +52,7 @@ import pytest
 
 from gns3server.controller import marker_replay
 from gns3server.controller.controller_error import (
+    ComputeError,
     ControllerBadRequestError,
     ControllerError,
     ControllerNotFoundError,
@@ -880,3 +888,254 @@ class TestDecodeFrame:
             assert seen <= known, f"unknown sharkd keys appeared: {seen - known}"
         finally:
             await manager.close_all()
+
+
+# ---------------------------------------------------------------------------
+# remote capture sources (fetch from the compute hosting the capture node)
+# ---------------------------------------------------------------------------
+
+
+def _pcap_bytes(frames, magic=PCAP_MAGIC_US, snaplen=65535):
+    """In-memory twin of _write_pcap: the bytes a compute serves for a marker
+    pcap."""
+
+    buf = bytearray(struct.pack("<IHHiIII", magic, 2, 4, 0, 0, snaplen, 1))
+    for sec, frac, payload in frames:
+        buf += struct.pack("<IIII", sec, frac, len(payload), len(payload))
+        buf += payload
+    return bytes(buf)
+
+
+class _FakeRemoteContent:
+    def __init__(self, chunks):
+        self._chunks = chunks
+
+    def iter_any(self):
+        async def _gen():
+            for chunk in self._chunks:
+                yield chunk
+
+        return _gen()
+
+
+class _FakeRemoteResponse:
+    """Duck-typed aiohttp response: streaming content + release()."""
+
+    def __init__(self, body):
+        self.status = 200
+        self.content = _FakeRemoteContent([body] if body else [])
+
+    async def release(self):
+        pass
+
+
+class _FakeRemoteCompute:
+    """Duck-typed Compute: answers GETs on /projects/{id}/files/{rel} from a
+    dict (missing key → 404), or raises a fixed failure on every call."""
+
+    def __init__(self, compute_id="remote-1", files=None, failure=None):
+        self.id = compute_id
+        self.requests = []
+        self.files = dict(files or {})
+        self.failure = failure
+
+    async def http_query(self, method, path, stream=False, **kwargs):
+        self.requests.append((method, path))
+        if self.failure is not None:
+            raise self.failure
+        rel = path.split("/files/", 1)[1]
+        if rel not in self.files:
+            raise ControllerNotFoundError(f"{method} {path} not found")
+        return _FakeRemoteResponse(self.files[rel])
+
+
+def _remote_project(tmp_path, markers, computes_by_node, markers_dir=None):
+    """_fake_project plus id and get_node resolving to the fake computes."""
+
+    project = _fake_project(tmp_path, markers, markers_dir)
+    project.id = "proj-1"
+
+    def get_node(node_id):
+        try:
+            return SimpleNamespace(compute=computes_by_node[node_id])
+        except KeyError:
+            raise ControllerNotFoundError(f"Node ID {node_id} doesn't exist")
+
+    project.get_node = get_node
+    return project
+
+
+_REL = "project-files/markers/remotenode_linkB_icmp.pcap"
+
+
+class TestRemoteSources:
+    @pytest.fixture(autouse=True)
+    def _fresh_cache(self, monkeypatch, tmp_path):
+        cache_dir = tmp_path / "marker-cache"
+        monkeypatch.setattr(marker_replay, "_remote_cache", {})
+        monkeypatch.setattr(marker_replay, "_remote_locks", {})
+        monkeypatch.setattr(marker_replay, "_remote_cache_dir", str(cache_dir))
+        return cache_dir
+
+    def _markers(self, tag=7):
+        return {
+            "linkA/icmp": _marker_entry(tag=tag, enabled=False, node_id="localnode"),
+            "linkB/icmp": _marker_entry(tag=tag, enabled=False, node_id="remotenode"),
+        }
+
+    async def test_remote_source_fetched_and_merged(self, tmp_path, monkeypatch):
+        _fake_columns(monkeypatch, {})
+        _write_pcap(tmp_path / "localnode_linkA_icmp.pcap", [(1693472000, 100000, _icmp_frame())])
+        compute = _FakeRemoteCompute(files={_REL: _pcap_bytes([(1693472000, 200000, _icmp_frame())])})
+        project = _remote_project(tmp_path, self._markers(), {"remotenode": compute})
+
+        timeline = await build_timeline(project, tag=7)
+
+        assert timeline["frame_count"] == 2
+        assert [f["node_id"] for f in timeline["frames"]] == ["localnode", "remotenode"]  # merged by ts
+        assert all(s["count"] == 1 for s in timeline["sources"])
+        assert not any("error" in s for s in timeline["sources"])
+        # the remote pcap was fetched through the compute project-files API
+        assert compute.requests == [("GET", f"/projects/proj-1/files/{_REL}")]
+
+    async def test_remote_missing_file_is_zero_count_without_error(self, tmp_path):
+        compute = _FakeRemoteCompute()  # no files: the marker never matched anything
+        project = _remote_project(tmp_path, self._markers(), {"remotenode": compute})
+
+        timeline = await build_timeline(project, tag=7)
+
+        by_node = {s["node_id"]: s for s in timeline["sources"]}
+        assert by_node["remotenode"]["count"] == 0
+        assert "error" not in by_node["remotenode"]
+        assert timeline["frame_count"] == 0
+
+    async def test_unreachable_compute_flags_source(self, tmp_path, monkeypatch):
+        _fake_columns(monkeypatch, {})
+        _write_pcap(tmp_path / "localnode_linkA_icmp.pcap", [(1693472000, 100000, _icmp_frame())])
+        compute = _FakeRemoteCompute(failure=ComputeError("Cannot connect to compute 'remote-1'"))
+        project = _remote_project(tmp_path, self._markers(), {"remotenode": compute})
+
+        timeline = await build_timeline(project, tag=7)
+
+        by_node = {s["node_id"]: s for s in timeline["sources"]}
+        assert by_node["localnode"]["count"] == 1
+        assert "error" not in by_node["localnode"]
+        assert by_node["remotenode"]["count"] == 0
+        assert "remote-1" in by_node["remotenode"]["error"]
+        # the timeline still carries the reachable sources' frames
+        assert timeline["frame_count"] == 1
+
+    async def test_local_compute_is_never_queried(self, tmp_path, monkeypatch):
+        _fake_columns(monkeypatch, {})
+        _write_pcap(tmp_path / "localnode_linkA_icmp.pcap", [(1693472000, 0, _icmp_frame())])
+        compute = _FakeRemoteCompute(compute_id="local")
+        markers = {
+            "linkA/icmp": _marker_entry(tag=7, enabled=False, node_id="localnode"),  # file on disk
+            "linkB/icmp": _marker_entry(tag=7, enabled=False, node_id="localnode2"),  # no file ever
+        }
+        project = _remote_project(tmp_path, markers, {"localnode": compute, "localnode2": compute})
+
+        timeline = await build_timeline(project, tag=7)
+
+        assert compute.requests == []
+        by_link = {s["link_id"]: s for s in timeline["sources"]}
+        assert by_link["linkA"]["count"] == 1
+        assert by_link["linkB"]["count"] == 0
+        assert "error" not in by_link["linkB"]
+
+    async def test_unchanged_refetch_preserves_cache_file(self, tmp_path, monkeypatch):
+        _fake_columns(monkeypatch, {})
+        compute = _FakeRemoteCompute(files={_REL: _pcap_bytes([(1693472000, 0, _icmp_frame())])})
+        markers = {"linkB/icmp": _marker_entry(tag=7, enabled=False, node_id="remotenode")}
+        project = _remote_project(tmp_path, markers, {"remotenode": compute})
+
+        await build_timeline(project, tag=7)
+        entry = marker_replay._remote_cache[("proj-1", "remote-1", _REL)]
+        first = os.stat(entry["path"])
+
+        await build_timeline(project, tag=7)  # refetch with identical bytes
+        assert os.stat(entry["path"]).st_ino == first.st_ino  # same file: resident sessions stay valid
+
+        compute.files[_REL] = _pcap_bytes([(1693473000, 0, _icmp_frame())])  # rebuilt capture
+        await build_timeline(project, tag=7)
+        assert os.stat(entry["path"]).st_ino != first.st_ino  # replaced copy
+
+    async def test_cache_evicts_idle_entries(self, tmp_path, monkeypatch):
+        _fake_columns(monkeypatch, {})
+        monkeypatch.setattr(marker_replay, "_REMOTE_CACHE_MAX", 2)
+        compute = _FakeRemoteCompute(
+            files={
+                f"project-files/markers/remotenode_link{i}_icmp.pcap": _pcap_bytes([(1693472000 + i, 0, _icmp_frame())])
+                for i in range(3)
+            }
+        )
+        markers = {f"link{i}/icmp": _marker_entry(tag=7, enabled=False, node_id="remotenode") for i in range(3)}
+        project = _remote_project(tmp_path, markers, {"remotenode": compute})
+
+        await build_timeline(project, tag=7)
+
+        assert len(marker_replay._remote_cache) == 2
+        assert all(os.path.exists(entry["path"]) for entry in marker_replay._remote_cache.values())
+
+    async def test_decode_remote_missing_is_not_matched_404(self, tmp_path):
+        compute = _FakeRemoteCompute()
+        project = _remote_project(tmp_path, self._markers(), {"remotenode": compute})
+
+        with pytest.raises(ControllerNotFoundError, match="nothing ever matched"):
+            await decode_frame(
+                project, tag=7, ts="1693472000.000000", node_id="remotenode", link_id="linkB", marker="icmp"
+            )
+
+    async def test_decode_remote_unreachable_raises_compute_error(self, tmp_path):
+        compute = _FakeRemoteCompute(failure=ComputeError("Cannot connect to compute 'remote-1'"))
+        project = _remote_project(tmp_path, self._markers(), {"remotenode": compute})
+
+        with pytest.raises(ComputeError, match="Could not fetch capture file"):
+            await decode_frame(
+                project, tag=7, ts="1693472000.000000", node_id="remotenode", link_id="linkB", marker="icmp"
+            )
+
+    async def test_decode_remote_frame_uses_cached_copy(self, tmp_path, monkeypatch):
+        compute = _FakeRemoteCompute(files={_REL: _pcap_bytes([(1693472000, 123456, _icmp_frame())])})
+        project = _remote_project(tmp_path, self._markers(), {"remotenode": compute})
+
+        class _FakeManager:
+            def __init__(self):
+                self.seen_pcap = None
+
+            @asynccontextmanager
+            async def session(self, pcap):
+                self.seen_pcap = pcap
+
+                async def rpc(method, params):
+                    assert (method, params) == ("frame", {"frame": 1, "proto": True})
+                    return {"tree": []}
+
+                yield SimpleNamespace(rpc=rpc)
+
+        manager = _FakeManager()
+        monkeypatch.setattr(marker_replay, "_get_manager", lambda: manager)
+
+        detail = await decode_frame(
+            project, tag=7, ts="1693472000.123456", node_id="remotenode", link_id="linkB", marker="icmp"
+        )
+
+        assert detail["hex"] == _icmp_frame().hex()
+        cached = {entry["path"] for entry in marker_replay._remote_cache.values()}
+        assert manager.seen_pcap in cached
+
+    async def test_close_sessions_removes_cache_dir(self, tmp_path, monkeypatch):
+        _fake_columns(monkeypatch, {})
+        compute = _FakeRemoteCompute(files={_REL: _pcap_bytes([(1693472000, 0, _icmp_frame())])})
+        project = _remote_project(
+            tmp_path, {"linkB/icmp": _marker_entry(tag=7, enabled=False, node_id="remotenode")}, {"remotenode": compute}
+        )
+
+        await build_timeline(project, tag=7)
+        cache_dir = marker_replay._remote_cache_dir
+        assert os.path.isdir(cache_dir)
+
+        await marker_replay.close_sessions()
+
+        assert not os.path.exists(cache_dir)
+        assert marker_replay._remote_cache == {}