diff --git a/docs/features/marker-tag-replay.md b/docs/features/marker-tag-replay.md index b29273635..51b35d3ac 100644 --- a/docs/features/marker-tag-replay.md +++ b/docs/features/marker-tag-replay.md @@ -41,7 +41,7 @@ graph TB TMP["/tmp scratch copies
(hardened-profile workaround)"] SK["sharkd -
(resident JSON-RPC on stdio)"] - UI -->|"GET range / frames [?filter=]"| GATE + UI -->|"GET range / frames [?filter=&link=]"| GATE GATE --> SCAN SCAN -->|"columns / filter matches"| SESS SESS --> TMP --> SK @@ -60,11 +60,44 @@ Backbone vs engine, deliberately separated: Session lifecycle: each session is validated per request against the source pcap's `(mtime, size)` — a mismatch (e.g. the capture node restarted and uBridge truncated the -pcap while paused) kills and respawns it. Sessions are LRU-bounded (8), each RPC has a -timeout and is serialized by a per-session lock (sharkd serves one request at a time). -sharkd reads a `/tmp` scratch copy of the pcap with a scratch `HOME` — hardened profiles -(AppArmor &c.) deny it the project directory and the user's home even though the server -process can read both. +pcap while paused) kills and respawns it. Sessions are refcounted while a request holds +them and the LRU bound (16) only ever evicts **idle** sessions, so neither a request's +own walk over a tag with more sources than the bound nor a concurrent request can have +its session killed mid-RPC. A single manager lock makes check-spawn atomic (concurrent +requests share one spawn instead of double-spawning a process nobody reaps). Each RPC +is serialized by a per-session lock (sharkd serves one request at a time), carries a +timeout and an **id check** — any transport failure (timeout, dead pipe, malformed or +stale reply, oversized line) kills the session for good, because a desynchronized +session would serve shifted results. sharkd reads a `/tmp` scratch **directory** +containing a copy of the pcap plus pinned Wireshark preferences (the column layout the +frames RPC is parsed against is a contract the server owns, immune to system or user +column customization) — hardened profiles (AppArmor &c.) deny it the project directory +and the user's home even though the server process can read both. The pcap copy runs +off the event loop, and server shutdown kills every session so no scratch copy leaks +into `/tmp` across restarts. + +### Resource bounds + +Three independent bounds, easily confused: + +| Bound | Value | Governs | +|-------|-------|---------| +| `_STREAM_LIMIT_BYTES` | 16 MB | Max **single JSON-RPC reply line** read from sharkd. A full 1000-row page measures ~190 KB against asyncio's 64 KB default — the trigger is the frame count in one pcap (≥ 1000), never the number of pcaps | +| `_FRAMES_PAGE` | 1000 | Rows fetched per `frames` RPC while draining one source's columns | +| `SESSION_MAX` | 16 | Resident **idle** sharkd sessions (LRU eviction) | + +The `range` frame list itself is uncapped by decision — the client owns the +rendering cost of a huge list. + +Process count follows usage, not the pcap inventory. sharkd loads one file per +process, so a session exists per source pcap **actually consulted** — a replay +request only touches its own tag's sources, and a session is either in use +(refcounted, held for a whole pagination drain rather than per RPC) or idle +(no holder — the only state the LRU cap ever evicts). At any moment the pool +holds at most `SESSION_MAX` idle sessions plus one per source being drained by +an in-flight request: a project with a thousand marker pcaps still runs at +most `16 + Σ(in-flight sources)` sharkd processes. Each process holds its pcap +in memory (~file size), which is the quantity the session cap primarily bounds. ## Business Process @@ -79,12 +112,12 @@ sequenceDiagram Note over UI: ③ pause every marker under the tag Note over UI,SK: ④ replay - UI->>C: GET /markers/tags/666/replay/range[?filter=…] + UI->>C: GET /markers/tags/666/replay/range[?filter=…&link=…] C->>C: gate → scan record headers → merge order C->>SK: frames {filter, skip, limit} → columns + matches C-->>UI: {start, end, sources, frames[] with src/dst/proto/info/bg/fg} UI->>C: GET frames?ts=T&window_ms=W (paging — {"frames": []} on a miss) - UI->>C: GET frame/detail?ts=…&node_id=…&link_id=…&marker=… + UI->>C: GET frame/detail?ts=…&node_id=…&link_id=…&marker=…[&frame_number=…] C->>C: hex straight from the pcap C->>SK: frame {frame: N, proto: true} SK-->>C: tree (keys renamed to the REST contract) @@ -126,7 +159,7 @@ without it. |--------|------|-------------| | GET | `/v3/projects/{pid}/markers/tags/{tag}/replay/range[?filter=&link=]` | Timeline metadata + full merged frame list with packet-list columns | | GET | `/v3/projects/{pid}/markers/tags/{tag}/replay/frames?ts=&window_ms=&limit=[&filter=&link=]` | Frames with ts in `[T, T+window]`, merged across sources | -| GET | `/v3/projects/{pid}/markers/tags/{tag}/replay/frame/detail?ts=&node_id=&link_id=&marker=` | Single frame: protocol tree + raw hex (lazy — one call per frame the user opens) | +| GET | `/v3/projects/{pid}/markers/tags/{tag}/replay/frame/detail?ts=&node_id=&link_id=&marker=[&frame_number=]` | Single frame: protocol tree + raw hex (lazy — one call per frame the user opens) | ### `range` — the timeline @@ -136,7 +169,6 @@ without it. "start": "1788369209.406812", "end": "1788369219.249085", "frame_count": 29, - "truncated": false, "sources": [ { "node_id": "47703cad…", "link_id": "2697a7c6…", "marker": "global-ospf", "data_link_type": "DLT_EN10MB", "count": 4 } @@ -152,9 +184,10 @@ without it. } ``` -- `frames` is the **full, merged, time-ordered list** (cap 5000) — one request lays out - the whole timeline. Over the cap, `frames` is omitted and per-second `buckets` are - returned instead with `truncated: true` (check with `'frames' in response`, not null). +- `frames` is the **full, merged, time-ordered list** — one request lays out the whole + timeline, **deliberately uncapped**: rendering a huge list is the client's concern, + and the window endpoint exists for incremental views. There is no `truncated` flag + and no bucket histogram; `frame_count` always equals `len(frames)`. - Every frame entry carries the Wireshark packet-list columns — `src` / `dst` / `proto` / `info` — plus the **coloring-rule hints `bg` / `fg`** (Wireshark's own palette decisions, so the UI can color rows exactly like Wireshark without shipping the @@ -166,12 +199,14 @@ without it. ### Display filter `?filter=` on both `range` and `frames` is a Wireshark display filter, -applied **before** counting and slicing — `start` / `end` / `frame_count` / -`frames` | `buckets` are all computed on the matching frames only. Filtered frames keep +applied **before** counting and slicing — `start` / `end` / `frame_count` / `frames` +are all computed on the matching frames only. Filtered frames keep their original pcap frame numbers. The filter travels as one argv-style element (never through a shell) and is capped at 2000 characters. An invalid expression is a **400** whose message carries sharkd's original error text — suitable for inline display in the -filter bar, and distinct from the 409 gate / 404 unknown-tag semantics. +filter bar, and distinct from the 409 gate / 404 unknown-tag semantics. Only sharkd's +filter-rejection error maps to 400; any other engine failure while a filter is set is a +502 (the filter was fine — the engine was not). ### Capture-source selection @@ -185,8 +220,7 @@ design: listed, with engine-free **total** counts, unaffected by `link` / `filter` — a source dropdown must not shrink when the view narrows. - **An unknown `link_id` matches nothing**: `frame_count: 0`, `start: null`, empty - `frames` / `buckets` — the same shape as a zero-match display filter, deliberately - not a 404. + `frames` — the same shape as a zero-match display filter, deliberately not a 404. ### `frames` — point / window query (paging) @@ -209,7 +243,10 @@ Invoked only when the user opens a frame. The `ts` must be the **exact string re in the timeline/frame list** (round-tripped verbatim — never re-serialized through a float); `node_id + link_id + marker` identify the pcap. The server re-resolves the ts against the file, guarding against a capture rebuilt between the timeline view and this -click. +click. Since ts is **not unique within one pcap** (same-microsecond frames are kept +deliberately), the optional `frame_number` — carried by every frame list entry — +disambiguates them: it must still land on the exact ts, and without it the first ts +match decodes. ```json { @@ -268,12 +305,12 @@ All error bodies are `{"message": "…"}` (the app's unified format). | Status | Description | |--------|-------------| -| 400 | Invalid display filter (message carries sharkd's original text) or filter longer than 2000 chars | +| 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, or ts does not match the file (the capture may have been rebuilt) | +| 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) | | 501 | sharkd not installed / unavailable — replay is unavailable, no degraded mode | -| 502 | sharkd failed or timed out (10 s per RPC) | +| 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 | ## Notes @@ -293,7 +330,16 @@ All error bodies are `{"message": "…"}` (the app's unified format). natural follow-up if list latency ever matters at many-source scale. - **Session invalidation is cheap and total.** Every request stats the source pcap; a rewritten file (mtime/size change) respawns the session — a paused-but-restarted - capture can never serve stale dissect state. + capture can never serve stale dissect state. Transport failures are equally total: a + timeout, dead pipe, malformed or stale reply (id mismatch) kills the session instead + of leaving a poisoned one resident, and concurrent requests for the same pcap share a + single spawn. +- **The packet-list layout is pinned, not assumed.** sharkd runs with a scratch `HOME` + whose Wireshark preferences fix `gui.column.format` to exactly the four columns the + frame entries carry (personal config overrides any `/etc/wireshark` customization), so + the column indexes the server parses are a contract it owns. +- **Server shutdown kills every resident session** and drops its `/tmp` scratch + 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; diff --git a/gns3server/api/routes/controller/projects.py b/gns3server/api/routes/controller/projects.py index 66fe27c2c..26fb7d042 100644 --- a/gns3server/api/routes/controller/projects.py +++ b/gns3server/api/routes/controller/projects.py @@ -253,11 +253,12 @@ async def replay_tag_range( The tag gate applies: every marker under the tag must be paused (``enabled: false``) — 409 otherwise. The response carries the timeline - bounds, per-source stats, and the full merged frame list while under the - frame cap (5000); above it the list is replaced by per-second buckets. + bounds, per-source stats, and the full merged frame list — deliberately + uncapped (rendering a huge list is the client's concern; the window + endpoint exists for incremental views). ``filter`` is an optional Wireshark display filter applied **before** - counting and slicing — start / end / frame_count / frames | buckets are + counting and slicing — start / end / frame_count / frames are all computed on the matching frames only. An invalid expression is a 400 carrying sharkd's original error text (for inline display in the UI filter bar). @@ -321,6 +322,7 @@ async def replay_tag_frame_detail( node_id: str, link_id: str, marker: str, + frame_number: Optional[int] = None, project: Project = Depends(dep_project), ) -> dict: """ @@ -332,14 +334,18 @@ async def replay_tag_frame_detail( ``children``) — values untouched. ``ts`` must be the exact string from the frame list; ``node_id`` + - ``link_id`` + ``marker`` identify the source pcap. The tag gate applies. - Requires sharkd — 501 without it. + ``link_id`` + ``marker`` identify the source pcap. ``frame_number`` + (optional, from the frame list entry) disambiguates same-microsecond + frames on one link — without it the first ts match decodes. The tag gate + applies. Requires sharkd — 501 without it. Required privilege: Project.Audit """ return await _replay_response( - marker_replay.decode_frame(project, tag, ts, node_id, link_id, marker) + marker_replay.decode_frame( + project, tag, ts, node_id, link_id, marker, frame_number=frame_number + ) ) diff --git a/gns3server/controller/marker_replay.py b/gns3server/controller/marker_replay.py index 5a0bdc945..a3b9366e3 100644 --- a/gns3server/controller/marker_replay.py +++ b/gns3server/controller/marker_replay.py @@ -31,13 +31,18 @@ Python, but it is an implementation detail, not an availability promise. Session model — sharkd loads one file at a time, so there is one resident session **per source pcap**: spawned lazily on first use, fed a ``/tmp`` -scratch copy (hardened profiles deny sharkd the project directory; a real -copy, not a symlink, since the profile resolves real paths) with a scratch -``HOME``, validated per request against the original's ``(mtime, size)`` — +scratch directory (hardened profiles deny sharkd the project directory; a +real copy, not a symlink, since the profile resolves real paths) with a +scratch ``HOME`` whose Wireshark preferences pin the packet-list column +layout, validated per request against the original's ``(mtime, size)`` — a mismatch (e.g. the capture node restarted and uBridge truncated the pcap -while paused) kills and respawns the session. A per-pcap asyncio lock -serializes RPCs (sharkd serves one request at a time), each with a timeout. -An LRU cap bounds concurrent sessions. +while paused) kills and respawns the session. A per-session asyncio lock +serializes RPCs (sharkd serves one request at a time), each with a timeout +and an id check; any transport failure (timeout, dead pipe, malformed or +stale reply) kills the session for good — a desynced session must never +serve shifted results. A single manager lock makes check-spawn atomic (no +double spawn under concurrent requests) and sessions are refcounted while +in use, so the LRU cap only ever evicts idle ones. Timestamps are uBridge's userspace ``gettimeofday`` at match time (µs, a value measured after the packet has crossed the kernel twice — the last @@ -56,21 +61,20 @@ import shutil import struct import tempfile import time +from contextlib import asynccontextmanager from .controller_error import ControllerError, ControllerNotFoundError, ControllerBadRequestError log = logging.getLogger(__name__) -# Full frame list is embedded in the range response while under this cap; -# above it the response degrades to start/end + per-second buckets so the -# client is never flooded by a high-traffic BPF. -FRAME_LIST_CAP = 5000 - # One JSON-RPC per sharkd session, serialized by a per-session lock. RPC_TIMEOUT_SECONDS = 10.0 -# Resident sharkd sessions are bounded; least-recently-used evicted first. -SESSION_MAX = 8 +# Resident sharkd sessions are bounded; only IDLE sessions are ever evicted +# (least-recently-used first), so a request's own walk over a tag with more +# sources than the cap never kills a session out from under it — the +# population is trimmed back as uses drop instead. +SESSION_MAX = 16 # Display filters travel as one argv element (never through a shell) and are # capped to keep absurd expressions off the command line. @@ -79,6 +83,16 @@ FILTER_MAX_LENGTH = 2000 # Batch size when draining sharkd's `frames` RPC (columns / filter matches). _FRAMES_PAGE = 1000 +# A full `frames` page (1000 rows of pinned columns) measures ~190 KB against +# the 64 KB default StreamReader limit — over it, readline() raises and clears +# the stream, failing the request AND desyncing the resident session. Raise +# the ceiling well above any page (or single-frame tree) we can produce. +_STREAM_LIMIT_BYTES = 16 * 1024 * 1024 + +# sharkd's JSON-RPC error code for a rejected display filter — the only error +# that belongs to the client (400); everything else is an engine fault (502). +_ERR_INVALID_FILTER = -13002 + class SharkdMissingError(ControllerError): """sharkd is not installed (or not on PATH) — replay is unavailable (501).""" @@ -194,26 +208,47 @@ def read_frame_bytes(path, frame_number): # sharkd: process environment and scratch copies # --------------------------------------------------------------------------- -def _scratch_copy(pcap): +# The packet-list layout is pinned through Wireshark preferences in the +# scratch HOME: personal config overrides any system-wide customization +# (/etc/wireshark &c.), so the column indexes in _columns_for are a contract +# we own rather than an environment default. Exactly the four columns the +# frame entries consume — nothing else rides along in every page. +_PINNED_COLUMNS = ( + 'gui.column.format: "Source", "%s", "Destination", "%d", ' + '"Protocol", "%p", "Info", "%i"\n' +) + + +async def _prepare_scratch(pcap): """ - Copy the pcap to a scratch file under the system temp dir for sharkd to - load. Hardened profiles (AppArmor &c.) can deny it access to the project - directory / the user's home while still allowing /tmp — a real copy, - deliberately not a symlink, since the profile resolves real paths. - Caller must unlink the returned path. + Build a scratch directory under the system temp dir holding the pcap copy + for sharkd to load plus the pinned column preferences. Hardened profiles + (AppArmor &c.) can deny sharkd the project directory / the user's home + while still allowing /tmp — a real copy, deliberately not a symlink, + since the profile resolves real paths. Caller must rmtree the directory. + + :returns: ``(scratch_dir, scratch_pcap_path)`` """ - fd, scratch = tempfile.mkstemp(suffix=".pcap", prefix="gns3-replay-") - os.close(fd) - shutil.copyfile(pcap, scratch) - return scratch + scratch_dir = tempfile.mkdtemp(prefix="gns3-replay-") + scratch = os.path.join(scratch_dir, "capture.pcap") + # A pcap-sized copy has no business stalling the event loop — a 1 GB + # capture must not freeze every other request for the duration. + await asyncio.to_thread(shutil.copyfile, pcap, scratch) + prefs_dir = os.path.join(scratch_dir, ".config", "wireshark") + os.makedirs(prefs_dir, exist_ok=True) + with open(os.path.join(prefs_dir, "preferences"), "w") as f: + f.write(_PINNED_COLUMNS) + return scratch_dir, scratch -def _engine_env(): - """Scratch HOME so sharkd never even tries to read the user's home.""" +def _engine_env(scratch_dir): + """Scratch HOME (and XDG config) so sharkd never even tries to read the + user's home — and always reads OUR pinned preferences instead.""" env = dict(os.environ) - env["HOME"] = tempfile.gettempdir() + env["HOME"] = scratch_dir + env["XDG_CONFIG_HOME"] = os.path.join(scratch_dir, ".config") return env @@ -276,10 +311,13 @@ def _count_tree_nodes(value): class _SharkdSession: """A resident `sharkd -` process with one pcap loaded, addressed through line-oriented JSON-RPC. Serialized by an asyncio lock (sharkd serves one - request at a time).""" + request at a time). Any transport-level failure aborts the session for + good (see _abort) — a timed-out or desynced session must never answer + later requests with shifted results.""" - def __init__(self, pcap, scratch, proc, stat): + def __init__(self, pcap, scratch_dir, scratch, proc, stat): self.pcap = pcap + self.scratch_dir = scratch_dir self.scratch = scratch self.proc = proc self.mtime_ns = stat.st_mtime_ns @@ -287,6 +325,12 @@ class _SharkdSession: self.last_used = time.monotonic() self.lock = asyncio.Lock() self._next_id = 0 + # Refcount of in-use holders (the manager's `session` context); the + # LRU cap only ever evicts sessions with zero uses, and a detached + # session (replaced in the manager, e.g. its pcap was rebuilt) closes + # itself when its last holder releases it. + self._uses = 0 + self._detached = False def matches(self, stat): """True while the source pcap is byte-identical to what was loaded — @@ -303,26 +347,41 @@ class _SharkdSession: async def rpc(self, method, params): async with self.lock: self._next_id += 1 - request = {"jsonrpc": "2.0", "id": self._next_id, "method": method, "params": params} + request_id = self._next_id + request = {"jsonrpc": "2.0", "id": request_id, "method": method, "params": params} try: self.proc.stdin.write((json.dumps(request) + "\n").encode()) await self.proc.stdin.drain() raw = await asyncio.wait_for(self.proc.stdout.readline(), RPC_TIMEOUT_SECONDS) except asyncio.TimeoutError: - raise SharkdError(f"sharkd timed out after {RPC_TIMEOUT_SECONDS:.0f}s on {method!r}") - except OSError as e: - raise SharkdError(f"sharkd session died on {method!r}: {e}") + await self._abort(f"sharkd timed out after {RPC_TIMEOUT_SECONDS:.0f}s on {method!r}") + except (OSError, ValueError) as e: + # ValueError: a reply line over the StreamReader limit — the + # buffer is cleared, so the session is desynced either way. + await self._abort(f"sharkd session died on {method!r}: {e}") if not raw: - raise SharkdError(f"sharkd closed the session during {method!r}") + await self._abort(f"sharkd closed the session during {method!r}") try: response = json.loads(raw) except ValueError as e: - raise SharkdError(f"Malformed sharkd response: {e}") + await self._abort(f"Malformed sharkd response: {e}") + # The echoed id proves this reply belongs to THIS request: a stale + # reply left over from a timed-out predecessor (or any desync) + # must never be served as fresh data. + if response.get("id") != request_id: + await self._abort(f"sharkd reply id mismatch on {method!r} (session desynchronized)") if "error" in response: error = response["error"] raise _SharkdRpcError(error.get("code"), str(error.get("message", ""))) return response.get("result") + async def _abort(self, message): + """Kill the session for good (process + scratch copy) and raise — the + manager replaces it on the next acquire. Idempotent with close().""" + + await self.close() + raise SharkdError(message) + async def close(self): try: if self.proc.returncode is None: @@ -336,56 +395,112 @@ class _SharkdSession: log.warning("sharkd session for %s did not exit after kill", self.pcap) except ProcessLookupError: pass - try: - os.unlink(self.scratch) - except OSError: - pass + shutil.rmtree(self.scratch_dir, ignore_errors=True) class _SharkdManager: - """Resident sharkd sessions keyed by source pcap path, LRU-bounded.""" + """Resident sharkd sessions keyed by source pcap path. + + One asyncio lock covers check-spawn-insert-reap: the check-then-spawn + window spans the pcap copy, the fork and the load RPC, so without it two + concurrent requests for the same pcap double-spawn and the loser leaks + (process + scratch copy) forever. Sessions are refcounted while in use + (the `session` context manager) and the LRU cap only evicts IDLE + sessions — a tag with more sources than the cap, or a concurrent request, + can never have its session killed mid-RPC; the population is trimmed + back to the cap as uses drop (temporary overshoot is allowed).""" def __init__(self): self._sessions = {} + self._mu = asyncio.Lock() - async def session_for(self, pcap): + @asynccontextmanager + async def session(self, pcap): + session = await self._acquire(pcap) + try: + yield session + finally: + await self._release(session) + + async def _acquire(self, pcap): if shutil.which("sharkd") is None: raise SharkdMissingError( "sharkd is not available on this server — marker replay requires sharkd " "(part of the Wireshark package)" ) - stat = os.stat(pcap) - session = self._sessions.get(pcap) - if session is not None and session.matches(stat) and session.alive(): - session.touch() - return session - if session is not None: - await session.close() - del self._sessions[pcap] - # Bound resident sessions: evict the least recently used. - while len(self._sessions) >= SESSION_MAX: - victim = min(self._sessions.values(), key=lambda s: s.last_used) + try: + stat = os.stat(pcap) + except FileNotFoundError: + # Deleted between the directory listing and here (marker removed, + # project cleaned up) — not a server fault. + raise ControllerNotFoundError(f"Capture file {os.path.basename(pcap)} no longer exists") + async with self._mu: + session = self._sessions.get(pcap) + if session is not None and session.matches(stat) and session.alive(): + session._uses += 1 + session.touch() + return session + if session is not None: + # Rebuilt/truncated source or a dead session: replace it. An + # in-use one closes itself when its last holder releases it. + del self._sessions[pcap] + if session._uses == 0: + await session.close() + else: + session._detached = True + spawned = await self._spawn(pcap, stat) + spawned._uses += 1 + self._sessions[pcap] = spawned + victims = self._evict_locked() + for victim in victims: + await victim.close() + return spawned + + def _evict_locked(self): + """Caller holds ``_mu``. Pop (not close — that can block on the reap) + least-recently-used IDLE sessions down to the cap; if everything is + in use, allow the overshoot rather than killing a live request.""" + + victims = [] + while len(self._sessions) > SESSION_MAX: + idle = [s for s in self._sessions.values() if s._uses == 0] + if not idle: + break + victim = min(idle, key=lambda s: s.last_used) + del self._sessions[victim.pcap] + victims.append(victim) + return victims + + async def _release(self, session): + async with self._mu: + session._uses -= 1 + victims = [] + if session._uses == 0: + if session._detached: + # Replaced in the manager while still held — now orphaned. + victims.append(session) + elif not session.alive(): + # Failed mid-use (its RPC aborted it) — drop the corpse. + if self._sessions.get(session.pcap) is session: + del self._sessions[session.pcap] + else: + victims = self._evict_locked() + for victim in victims: await victim.close() - self._sessions.pop(victim.pcap, None) - session = await self._spawn(pcap, stat) - self._sessions[pcap] = session - return session async def _spawn(self, pcap, stat): - scratch = _scratch_copy(pcap) + scratch_dir, scratch = await _prepare_scratch(pcap) try: proc = await asyncio.create_subprocess_exec( "sharkd", "-", stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.DEVNULL, env=_engine_env(), + stderr=asyncio.subprocess.DEVNULL, env=_engine_env(scratch_dir), + limit=_STREAM_LIMIT_BYTES, ) except OSError as e: - try: - os.unlink(scratch) - except OSError: - pass + shutil.rmtree(scratch_dir, ignore_errors=True) raise SharkdError(f"Could not run sharkd: {e}") - session = _SharkdSession(pcap, scratch, proc, stat) + session = _SharkdSession(pcap, scratch_dir, scratch, proc, stat) try: await session.rpc("load", {"file": scratch}) except Exception as e: @@ -394,9 +509,11 @@ class _SharkdManager: return session async def close_all(self): - for session in list(self._sessions.values()): + async with self._mu: + sessions = list(self._sessions.values()) + self._sessions.clear() + for session in sessions: await session.close() - self._sessions.clear() _manager = None @@ -409,6 +526,17 @@ def _get_manager(): return _manager +async def close_sessions(): + """Server shutdown hook: kill every resident session and drop its scratch + copy, so nothing leaks into /tmp across restarts.""" + + global _manager + manager = _manager + if manager is not None: + await manager.close_all() + _manager = None + + # --------------------------------------------------------------------------- # Columns + display filter (sharkd `frames` RPC) # --------------------------------------------------------------------------- @@ -419,42 +547,49 @@ async def _columns_for(pcap, filter_expr): ``{frame_number: {src, dst, proto, info, bg, fg}}``; with a ``filter_expr`` the keys are exactly the matching frame numbers, so one pass both filters and enriches the merge. + + The column layout is pinned by the scratch-HOME preferences (see + _PINNED_COLUMNS): ``c[0]``=src, ``c[1]``=dst, ``c[2]``=proto, ``c[3]``=info + is a layout we own, not an environment default. The whole pagination + loop holds the session so the LRU cap cannot evict it between pages. """ - session = await _get_manager().session_for(pcap) - columns = {} - skip = 0 - while True: - # sharkd rejects skip=0 ("must be a positive integer") — only send it - # once there is actually something to skip. - params = {"limit": _FRAMES_PAGE} - if skip: - params["skip"] = skip - if filter_expr is not None: - params["filter"] = filter_expr - try: - rows = await session.rpc("frames", params) - except _SharkdRpcError as e: + manager = _get_manager() + async with manager.session(pcap) as session: + columns = {} + skip = 0 + while True: + # sharkd rejects skip=0 ("must be a positive integer") — only send + # it once there is actually something to skip. + params = {"limit": _FRAMES_PAGE} + if skip: + params["skip"] = skip if filter_expr is not None: - raise FilterError(f"Invalid display filter: {e.message}") - raise SharkdError(f"sharkd frames failed on {os.path.basename(pcap)}: {e.message}") - for row in rows: + params["filter"] = filter_expr try: - frame_number = int(row.get("num")) - except (TypeError, ValueError): - continue - cols = row.get("c") or [] - columns[frame_number] = { - "src": cols[2] or None if len(cols) > 2 else None, - "dst": cols[3] or None if len(cols) > 3 else None, - "proto": cols[4] or None if len(cols) > 4 else None, - "info": cols[6] or None if len(cols) > 6 else None, - "bg": row.get("bg"), - "fg": row.get("fg"), - } - if len(rows) < _FRAMES_PAGE: - return columns - skip += len(rows) + rows = await session.rpc("frames", params) + except _SharkdRpcError as e: + if filter_expr is not None and e.code == _ERR_INVALID_FILTER: + raise FilterError(f"Invalid display filter: {e.message}") + # Anything else is an engine fault, not the client's filter. + raise SharkdError(f"sharkd frames failed on {os.path.basename(pcap)}: {e.message}") + for row in rows or []: + try: + frame_number = int(row.get("num")) + except (TypeError, ValueError): + continue + cols = row.get("c") or [] + columns[frame_number] = { + "src": cols[0] or None if len(cols) > 0 else None, + "dst": cols[1] or None if len(cols) > 1 else None, + "proto": cols[2] or None if len(cols) > 2 else None, + "info": cols[3] or None if len(cols) > 3 else None, + "bg": row.get("bg"), + "fg": row.get("fg"), + } + if len(rows or []) < _FRAMES_PAGE: + return columns + skip += len(rows or []) # --------------------------------------------------------------------------- @@ -572,39 +707,28 @@ def _validate_filter(filter_expr): ) -async def build_timeline(project, tag, frame_cap=FRAME_LIST_CAP, filter_expr=None, link_id=None): +async def build_timeline(project, tag, filter_expr=None, link_id=None): """ - The ``range`` response: timeline bounds, per-source stats, and (under - ``frame_cap``) the full merged frame list for one-request timeline - layout. Over the cap the list is replaced by per-second buckets. With a - ``filter_expr`` and/or a ``link_id`` every figure is computed on the - matching frames only (``sources`` stays the full tag inventory). + The ``range`` response: timeline bounds, per-source stats, and the full + merged frame list — deliberately uncapped (a flat list is the whole + contract; rendering a huge list is the client's concern, and the window + endpoint exists for incremental views). With a ``filter_expr`` and/or a + ``link_id`` every figure is computed on the matching frames only + (``sources`` stays the full tag inventory). """ _validate_filter(filter_expr) entries = gate_tag(project, tag) frames, sources = await _merged_frames(project, entries, filter_expr, link_id=link_id) - response = { + return { "tag": tag, "start": frames[0]["ts"] if frames else None, "end": frames[-1]["ts"] if frames else None, "frame_count": len(frames), - "truncated": len(frames) > frame_cap, "sources": sources, + "frames": frames, } - if len(frames) <= frame_cap: - response["frames"] = frames - else: - buckets = {} - for frame in frames: - second = _parse_ts(frame["ts"]) // 1_000_000 - buckets[second] = buckets.get(second, 0) + 1 - response["buckets"] = [ - {"ts": _format_ts(second, 0), "count": count} - for second, count in sorted(buckets.items()) - ] - return response async def query_frames(project, tag, ts, window_ms=100, limit=1000, filter_expr=None, link_id=None): @@ -629,13 +753,19 @@ async def query_frames(project, tag, ts, window_ms=100, limit=1000, filter_expr= # Frame detail (lazy — one frame per call, via the resident session) # --------------------------------------------------------------------------- -async def decode_frame(project, tag, ts, node_id, link_id, marker): +async def decode_frame(project, tag, ts, node_id, link_id, marker, frame_number=None): """ Decode exactly one frame: locate its pcap by source identity, verify the 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). + + ``ts`` is not unique within one pcap (same-microsecond frames are kept + deliberately); an explicit ``frame_number`` from the frame list entry + disambiguates them and must still land on the exact ts. Without one the + first ts match decodes — fine unless two frames share a microsecond on + the same link. """ entries = gate_tag(project, tag) @@ -654,28 +784,38 @@ async def decode_frame(project, tag, ts, node_id, link_id, marker): raise ControllerNotFoundError(f"No capture file for marker '{marker}' (nothing ever matched)") frames = scan_pcap_frames(pcap) - # 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, + rebuilt_message = ( + f"No frame at ts {ts} in marker '{marker}' (the capture may have been rebuilt)" ) if frame_number is None: - raise ControllerNotFoundError( - f"No frame at ts {ts} in marker '{marker}' (the capture may have been rebuilt)" + # 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) - session = await _get_manager().session_for(pcap) - try: - result = await session.rpc("frame", {"frame": frame_number, "proto": True}) - except _SharkdRpcError as e: - if e.code == -8003: # frame number out of range — file changed under us - raise ControllerNotFoundError( - f"No frame at ts {ts} in marker '{marker}' (the capture may have been rebuilt)" - ) - raise SharkdError(f"sharkd frame failed: {e.message}") + 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}") tree = _rename_value(result.get("tree", [])) return { diff --git a/gns3server/core/tasks.py b/gns3server/core/tasks.py index f8e8dadc9..0b745a067 100644 --- a/gns3server/core/tasks.py +++ b/gns3server/core/tasks.py @@ -111,6 +111,11 @@ async def shutdown(app: FastAPI) -> None: auto_discover_images_task_handle.cancel() await HTTPClient.close_session() await MarkerManager.instance().stop() + # Kill resident sharkd sessions (marker replay) and drop their /tmp + # scratch copies before the process exits. + from gns3server.controller import marker_replay + + await marker_replay.close_sessions() await Controller.instance().stop() for module in MODULES: diff --git a/tests/api/routes/controller/test_marker_replay.py b/tests/api/routes/controller/test_marker_replay.py index 24fdf5c8e..b5db7398d 100644 --- a/tests/api/routes/controller/test_marker_replay.py +++ b/tests/api/routes/controller/test_marker_replay.py @@ -141,7 +141,6 @@ class TestReplayRoutes: assert body["frame_count"] == 4 assert body["start"] == "1693472000.500000" assert body["end"] == "1693472002.000000" - assert body["truncated"] is False assert [f["node_id"] for f in body["frames"]] == ["n1", "n2", "n1", "n2"] assert [f["ts"] for f in body["frames"]] == [ "1693472000.500000", "1693472001.000000", @@ -339,6 +338,36 @@ class TestReplayRoutes: assert ttl["filter_expr"] == "ip.ttl == 64" assert ttl["pos"] == 22 and ttl["size"] == 1 + @sharkd_present + async def test_detail_frame_number_disambiguates_same_ts( + self, app: FastAPI, client: AsyncClient, project: Project, no_residual_sessions + ) -> None: + + # Two frames in the same microsecond: only the explicit frame number + # (from the frame list entry) tells them apart. + link = _add_marker(project, tag=7, enabled=False, node_id="n1", frames=[ + (1693472000, 123456, _icmp_frame()), + (1693472000, 123456, _tcp_syn_frame()), + ]) + common = {"ts": "1693472000.123456", "node_id": "n1", + "link_id": link.id, "marker": "icmp"} + + response = await client.get( + app.url_path_for("replay_tag_frame_detail", project_id=project.id, tag=7), + params=common, + ) + assert response.status_code == status.HTTP_200_OK + assert response.json()["source"]["frame_number"] == 1 + assert response.json()["hex"] == _icmp_frame().hex() + + response = await client.get( + app.url_path_for("replay_tag_frame_detail", project_id=project.id, tag=7), + params={**common, "frame_number": 2}, + ) + assert response.status_code == status.HTTP_200_OK + assert response.json()["source"]["frame_number"] == 2 + assert response.json()["hex"] == _tcp_syn_frame().hex() + @sharkd_present async def test_detail_501_when_sharkd_disappears( self, app: FastAPI, client: AsyncClient, project: Project diff --git a/tests/controller/test_marker_replay.py b/tests/controller/test_marker_replay.py index a2f992440..3c3ed252b 100644 --- a/tests/controller/test_marker_replay.py +++ b/tests/controller/test_marker_replay.py @@ -22,18 +22,22 @@ Unit tests for the tag-keyed aggregate replay module (sharkd edition): nanosecond-magic normalization) and raw-bytes reads for the hex view * the tag gate (404 unknown tag, 409 while any marker captures) — engine-free * timeline assembly with injected columns: cross-source merge ordered by - (ts, source, frame number), same-microsecond tiebreak, frame-cap - degradation to buckets, display-filter application before count/slice + (ts, source, frame number), same-microsecond tiebreak, the uncapped full + frame list, display-filter application before count/slice * the tree key renaming (closed census key set, values untouched, unknown keys pass through verbatim, internal hf ids dropped) * the resident sharkd sessions — spawn/load/reuse, (mtime, size) - invalidation, LRU bound — against the real sharkd where installed, plus - the frame detail end-to-end (hex + renamed tree + filter expressions). + 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). """ +import asyncio import os import shutil import struct +import sys from types import SimpleNamespace from unittest.mock import patch @@ -47,6 +51,7 @@ from gns3server.controller.controller_error import ( from gns3server.controller import marker_replay from gns3server.controller.marker_replay import ( FilterError, + SharkdError, SharkdMissingError, build_timeline, decode_frame, @@ -141,6 +146,30 @@ def _cols(src="10.0.0.1", dst="10.0.0.3", proto="ICMP", info="Echo (ping) reques "bg": "ffffff", "fg": "000000"} +class _FakeSession: + """Duck-typed _SharkdSession for manager-level tests (no engine).""" + + def __init__(self, pcap): + self.pcap = pcap + self.last_used = 0 + self.closed = False + self._uses = 0 + self._detached = False + self.mtime_ns, self.size = 0, 0 + + def matches(self, stat): + return True + + def alive(self): + return not self.closed + + def touch(self): + pass + + async def close(self): + self.closed = True + + # --------------------------------------------------------------------------- # pcap scanning (engine-free backbone) # --------------------------------------------------------------------------- @@ -268,7 +297,9 @@ class TestTimeline: assert timeline["frames"] == [] assert timeline["sources"][0]["count"] == 0 - async def test_over_cap_degrades_to_buckets(self, tmp_path, monkeypatch): + async def test_frame_list_is_uncapped(self, tmp_path, monkeypatch): + # The list is the whole contract — no truncation flag, no buckets, + # however many frames the tag holds. _write_pcap(tmp_path / "n1_linkA_icmp.pcap", [ (1693472000, 0, b"a" * 60), (1693472000, 500000, b"a" * 60), (1693472001, 0, b"a" * 60), @@ -276,13 +307,10 @@ class TestTimeline: _fake_columns(monkeypatch, {"n1_linkA_icmp.pcap": {1: _cols(), 2: _cols(), 3: _cols()}}) project = _fake_project(tmp_path, {"linkA/icmp": _marker_entry(tag=7, enabled=False, node_id="n1")}) - timeline = await build_timeline(project, tag=7, frame_cap=2) - assert timeline["truncated"] is True - assert "frames" not in timeline - assert timeline["buckets"] == [ - {"ts": "1693472000.000000", "count": 2}, - {"ts": "1693472001.000000", "count": 1}, - ] + timeline = await build_timeline(project, tag=7) + assert timeline["frame_count"] == 3 + assert len(timeline["frames"]) == 3 + assert "truncated" not in timeline and "buckets" not in timeline async def test_filter_applies_before_count_and_slice(self, tmp_path, monkeypatch): # Three frames; the injected "matching set" (what a real engine would @@ -445,43 +473,155 @@ class TestSessions: with pytest.raises(SharkdMissingError): await build_timeline(project, tag=7) - async def test_lru_bound_evicts_least_recently_used(self, tmp_path): + async def test_cap_evicts_idle_lru_only(self, tmp_path): manager = marker_replay._SharkdManager() - - class FakeSession: - def __init__(self, pcap, used): - self.pcap, self.last_used = pcap, used - self.closed = False - self.mtime_ns, self.size = 0, 0 - - def matches(self, stat): - return False # always respawn → exercises the eviction path - - def alive(self): - return False - - def touch(self): - pass - - async def close(self): - self.closed = True - - fake_by_pcap = {} + fakes = {} async def fake_spawn(pcap, stat): - session = FakeSession(pcap, 0) - fake_by_pcap[pcap] = session + session = _FakeSession(pcap) + fakes[pcap] = session return session with patch.object(manager, "_spawn", side_effect=fake_spawn): + # cap + 2 distinct pcaps, each acquired and released (idle again). for i in range(marker_replay.SESSION_MAX + 2): pcap = tmp_path / f"pcap{i}" _write_pcap(pcap, [(1693472000, 0, b"a" * 60)]) - await manager.session_for(str(pcap)) - # Bounded to SESSION_MAX; the earliest (least recently used) got evicted. + async with manager.session(str(pcap)): + pass + # Bounded to SESSION_MAX; the earliest (least recently used) closed. assert len(manager._sessions) == marker_replay.SESSION_MAX - assert fake_by_pcap[str(tmp_path / "pcap0")].closed is True - assert fake_by_pcap[str(tmp_path / "pcap1")].closed is True + assert fakes[str(tmp_path / "pcap0")].closed is True + assert fakes[str(tmp_path / "pcap1")].closed is True + assert fakes[str(tmp_path / "pcap2")].closed is False + + async def test_in_use_session_survives_cap_pressure(self, tmp_path): + manager = marker_replay._SharkdManager() + + async def fake_spawn(pcap, stat): + return _FakeSession(pcap) + + with patch.object(manager, "_spawn", side_effect=fake_spawn): + pcap0 = tmp_path / "pcap0" + _write_pcap(pcap0, [(1693472000, 0, b"a" * 60)]) + async with manager.session(str(pcap0)) as held: + # More sources than the cap while pcap0 is held open. + for i in range(1, marker_replay.SESSION_MAX + 4): + pcap = tmp_path / f"pcap{i}" + _write_pcap(pcap, [(1693472000, 0, b"a" * 60)]) + async with manager.session(str(pcap)): + pass + # A held session is never the eviction victim (no mid-RPC kill). + assert held.closed is False + assert str(pcap0) in manager._sessions + # Released → the population is trimmed back under the cap. + assert len(manager._sessions) <= marker_replay.SESSION_MAX + await manager.close_all() + + async def test_concurrent_acquire_spawns_once(self, tmp_path): + manager = marker_replay._SharkdManager() + spawns = [] + + async def slow_spawn(pcap, stat): + spawns.append(pcap) + await asyncio.sleep(0.05) # widen the check-then-spawn window + return _FakeSession(pcap) + + pcap = tmp_path / "n1_linkA_icmp.pcap" + _write_pcap(pcap, [(1693472000, 0, b"a" * 60)]) + with patch.object(manager, "_spawn", side_effect=slow_spawn): + first, second = await asyncio.gather( + manager._acquire(str(pcap)), manager._acquire(str(pcap)) + ) + assert first is second # concurrent requests share one spawn + assert len(spawns) == 1 + await manager.close_all() + + async def test_rpc_timeout_kills_session(self, tmp_path, monkeypatch): + monkeypatch.setattr(marker_replay, "RPC_TIMEOUT_SECONDS", 0.2) + pcap = tmp_path / "n1_linkA_icmp.pcap" + _write_pcap(pcap, [(1693472000, 0, b"a" * 60)]) + # A process that never answers: the session must die (never serve the + # late reply to a later request) instead of staying resident. + proc = await asyncio.create_subprocess_exec( + "sleep", "60", + stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, + ) + session = marker_replay._SharkdSession( + str(pcap), str(tmp_path / "scratch"), "unused", proc, os.stat(str(pcap)) + ) + with pytest.raises(SharkdError, match="timed out"): + await session.rpc("frames", {}) + assert session.alive() is False + assert proc.returncode is not None + + async def test_stale_reply_id_aborts_session(self, tmp_path): + # A "sharkd" that always answers with a foreign id — every reply is a + # stale one as far as the caller is concerned; serving it would be + # shifted data, so the session must abort instead. + fake = tmp_path / "fake_sharkd.py" + fake.write_text( + "import json, sys\n" + "for line in sys.stdin:\n" + " json.loads(line)\n" + " sys.stdout.write(json.dumps({'jsonrpc': '2.0', 'id': 424242, 'result': []}) + '\\n')\n" + " sys.stdout.flush()\n" + ) + pcap = tmp_path / "n1_linkA_icmp.pcap" + _write_pcap(pcap, [(1693472000, 0, b"a" * 60)]) + proc = await asyncio.create_subprocess_exec( + sys.executable, str(fake), + stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, + ) + session = marker_replay._SharkdSession( + str(pcap), str(tmp_path / "scratch"), "unused", proc, os.stat(str(pcap)) + ) + with pytest.raises(SharkdError, match="id mismatch"): + await session.rpc("frames", {}) + assert session.alive() is False + assert proc.returncode is not None + + async def test_filter_error_only_for_the_filter_code(self, tmp_path, monkeypatch): + manager = marker_replay._SharkdManager() + monkeypatch.setattr(marker_replay, "_manager", manager) + pcap = tmp_path / "n1_linkA_icmp.pcap" + _write_pcap(pcap, [(1693472000, 0, b"a" * 60)]) + + class _RpcFail(_FakeSession): + def __init__(self, pcap, code): + super().__init__(pcap) + self.code = code + + async def rpc(self, method, params): + raise marker_replay._SharkdRpcError(self.code, "boom") + + manager._sessions[str(pcap)] = _RpcFail(str(pcap), marker_replay._ERR_INVALID_FILTER) + with pytest.raises(FilterError): + await marker_replay._columns_for(str(pcap), "icmp") + # Same failure WITHOUT a filter is an engine fault, not a 400. + manager._sessions[str(pcap)] = _RpcFail(str(pcap), marker_replay._ERR_INVALID_FILTER) + with pytest.raises(SharkdError): + await marker_replay._columns_for(str(pcap), None) + # A non-filter engine error with a filter set is NOT the client's fault. + manager._sessions[str(pcap)] = _RpcFail(str(pcap), -9999) + with pytest.raises(SharkdError): + await marker_replay._columns_for(str(pcap), "icmp") + + @sharkd_present + async def test_real_pagination_over_1000_frames(self, tmp_path): + # A full page measures well over the 64 KB default StreamReader limit + # (~190 KB with realistic columns) — pagination must not blow up the + # stream (nor desync the resident session). + pcap = tmp_path / "big.pcap" + _write_pcap(pcap, [(1693472000 + i, 0, _icmp_frame()) for i in range(1001)]) + manager = marker_replay._get_manager() + try: + columns = await marker_replay._columns_for(str(pcap), None) + assert len(columns) == 1001 + assert columns[1001]["src"] == "10.0.0.1" + assert columns[1001]["proto"] == "ICMP" + finally: + await manager.close_all() @sharkd_present async def test_real_session_columns_and_filter(self, tmp_path): @@ -518,12 +658,14 @@ class TestSessions: _write_pcap(pcap, [(1693472000, 123456, _icmp_frame())]) manager = marker_replay._get_manager() try: - first = await manager.session_for(str(pcap)) - assert await manager.session_for(str(pcap)) is first # warm reuse + async with manager.session(str(pcap)) as first: + pass + async with manager.session(str(pcap)) as warm: + assert warm is first # warm reuse # Rewrite the file (simulating a truncated/rebuilt capture). _write_pcap(pcap, [(1693473000, 0, _icmp_frame())]) - second = await manager.session_for(str(pcap)) - assert second is not first + async with manager.session(str(pcap)) as second: + assert second is not first columns = await marker_replay._columns_for(str(pcap), None) assert set(columns) == {1} finally: @@ -552,6 +694,54 @@ class TestDecodeFrame: await decode_frame(project, tag=7, ts="1693472000.123456", node_id="nobody", link_id="linkA", marker="icmp") + async def test_explicit_frame_number_out_of_range_404(self, tmp_path): + project = self._project(tmp_path) + with pytest.raises(ControllerNotFoundError, match="rebuilt"): + await decode_frame(project, tag=7, ts="1693472000.123456", + node_id="n1", link_id="linkA", marker="icmp", frame_number=5) + + async def test_explicit_frame_number_ts_mismatch_404(self, tmp_path): + # The frame number must still land on the exact round-tripped ts — + # a rebuilt capture cannot be decoded by stale coordinates. + project = self._project(tmp_path) + with pytest.raises(ControllerNotFoundError, match="rebuilt"): + await decode_frame(project, tag=7, ts="1693472001.000000", + node_id="n1", link_id="linkA", marker="icmp", frame_number=1) + + async def test_hex_read_failure_is_404_not_null_hex(self, tmp_path, monkeypatch): + project = self._project(tmp_path) + monkeypatch.setattr(marker_replay, "read_frame_bytes", lambda path, n: None) + with pytest.raises(ControllerNotFoundError, match="rebuilt"): + await decode_frame(project, tag=7, ts="1693472000.123456", + node_id="n1", link_id="linkA", marker="icmp") + + @sharkd_present + async def test_same_microsecond_frames_disambiguated_by_frame_number(self, tmp_path): + # ts is not unique within one pcap: two frames share a microsecond, + # and only the explicit frame number tells them apart. + pcap = tmp_path / "n1_linkA_icmp.pcap" + icmp, tcp = _icmp_frame(), _tcp_syn_frame() + _write_pcap(pcap, [(1693472000, 123456, icmp), (1693472000, 123456, tcp)]) + project = _fake_project(tmp_path, { + "linkA/icmp": _marker_entry(tag=7, enabled=False, node_id="n1"), + }) + manager = marker_replay._get_manager() + try: + default = await decode_frame(project, tag=7, ts="1693472000.123456", + node_id="n1", link_id="linkA", marker="icmp") + second = await decode_frame(project, tag=7, ts="1693472000.123456", + node_id="n1", link_id="linkA", marker="icmp", + frame_number=2) + finally: + await manager.close_all() + # Without a frame number: first ts match. + assert default["source"]["frame_number"] == 1 + assert default["hex"] == icmp.hex() + # With one: exactly that frame's bytes and tree. + assert second["source"]["frame_number"] == 2 + assert second["hex"] == tcp.hex() + assert second["field_count"] > 10 + @sharkd_present async def test_decode_end_to_end(self, tmp_path): manager = marker_replay._get_manager() @@ -595,7 +785,6 @@ class TestDecodeFrame: ]) manager = marker_replay._get_manager() try: - session = await manager.session_for(str(pcap)) known = set(marker_replay._KEY_RENAME) | {"h", "e"} seen = set() @@ -608,9 +797,10 @@ class TestDecodeFrame: for item in value: walk(item) - for frame_number in (1, 2): - result = await session.rpc("frame", {"frame": frame_number, "proto": True}) - walk(result.get("tree", [])) + async with manager.session(str(pcap)) as session: + for frame_number in (1, 2): + result = await session.rpc("frame", {"frame": frame_number, "proto": True}) + walk(result.get("tree", [])) assert seen <= known, f"unknown sharkd keys appeared: {seen - known}" finally: await manager.close_all()