diff --git a/docs/architecture-tour.md b/docs/architecture-tour.md index 9afac6f..9f0f01b 100644 --- a/docs/architecture-tour.md +++ b/docs/architecture-tour.md @@ -131,6 +131,9 @@ publish. Pacing uses absolute deadlines at 1.0x by default; it never skips frame snapshot at 30 Hz: a JSON header (status, rates, learning stats, game mode, milestone rank and total, sugar state, events, chat ring) followed by attachments: RGBA frame, f32 stereo 48 kHz audio (binjgb's unipolar u8 converted and DC-blocked), and a 17,407-byte spike bitset. + flysim serves it itself by default; with `FLY_FEED_VIA=bus` it publishes each snapshot on an + embedded flybus router and the `fly-edge` process serves the same bytes + (`docs/design/flybus.md`, "Feed over the bus"). - Control API (`docs/control-api.md`): loopback HTTP :7401. `POST /stimulate` (sugar: a timed PAM pulse, rate-limited server side), `POST /reward` (present, disabled by config), `POST /chat` (sanitized, deny-listed, ring of 12), `/status`, `/checkpoint`, `/pause`, `/resume`, @@ -159,8 +162,8 @@ sequenceDiagram S->>S: every 5 s hot copy, every 300 s durable checkpoint ``` -Where: `services/flysim/crates/flysim/src/{main,config,simloop,pacing,snapshot,feed,api,chat,store,eventlog,metrics}.rs`, -`docs/design/flysim.md`. +Where: `services/flysim/crates/flysim/src/{main,config,simloop,pacing,snapshot,feed,feedbus,api,chat,store,eventlog,metrics}.rs`, +`services/flysim/crates/fly-edge`, `docs/design/flysim.md`. ## 4. Stage page diff --git a/docs/design/flybus.md b/docs/design/flybus.md index 1dcff3b..1245167 100644 --- a/docs/design/flybus.md +++ b/docs/design/flybus.md @@ -1,6 +1,7 @@ # flybus: the communications bus -Status: **crate landed, nothing wired onto it**. Written 2026-09-22. Index only; the +Status: **crate landed; the feed rides it behind `FLY_FEED_VIA=bus`, off by default**. +Written 2026-09-22, amended 2026-09-23 (EDGE-01, below). Index only; the authority for the API and the wire format is the crate's own [README](../../services/flysim/crates/flybus/README.md), and the audit of the crate against the draft is the [conformance report](session-framework/bus-conformance.md). @@ -37,8 +38,8 @@ does not change any published contract by existing. ## Crate layout -`services/flysim/crates/flybus`, a workspace member of the flysim workspace; no other crate -depends on it yet. +`services/flysim/crates/flybus`, a workspace member of the flysim workspace. `flysim` depends +on it for the feed publisher (`src/feedbus.rs`) and `fly-edge` for the subscriber. | Module | Contents | | --- | --- | @@ -59,16 +60,131 @@ allocate/seal/read with quotas and router restarts, plus the conformance suites ## Wiring still pending -- **flysim publisher.** Router startup inside the sim service, a store root under its - runtime directory, and snapshot publication as artifact plus header envelope. +- ~~**flysim publisher.**~~ Done 2026-09-23 behind `FLY_FEED_VIA=bus`: see "Feed over the bus". - **flysim control services.** The control endpoints as RPC services with grants, so the "no button endpoint" structural guarantee is expressed as a grant table. - **Stage and bridge clients.** Both are TypeScript/Node; the crate is Rust only, so either a binding or a thin translating edge process is required before they leave the WebSocket - and HTTP surfaces. -- **Sizing.** `max_store_bytes`, `max_retained_bytes` and `max_latest_in_flight` need values - chosen for 1.2 MB frames at 30 to 60 Hz with a slow consumer, not the defaults. -- **Lifecycle.** Orphaned store directories are cleaned only when a new router starts on the - same root, so service restart order and the store root's location need a decision. + and HTTP surfaces. The operator chose the edge process (port decisions, 2026-09-23); for + the feed it exists (`fly-edge`), and they keep the WebSocket contract unchanged. The + control API (:7401) is the next slice and stays in flysim until then. +- ~~**Sizing.**~~ Decided 2026-09-23: amendment "Feed sizing" below. +- ~~**Lifecycle.**~~ Decided 2026-09-23: amendment "Feed store lifecycle" below. - **Migration order.** The feed is the cheaper first move; control should follow only once the bus carries the feed in production for a full session. + +## Feed over the bus (2026-09-23, EDGE-01) + +`feed.via` (`FLY_FEED_VIA`) picks who serves `ws://127.0.0.1:7400/feed`. `direct` is the +default and is the behaviour that predates the bus. With `bus`: + +```text + sim thread --watch--> publisher task --flybus (in memory)--> Router + (unchanged) (flysim-bus runtime) | /edge.sock + v (bound to "fly-edge") + fly-edge: Subscription -> watch -> flysim::feed :7400 +``` + +- flysim does not bind `feed.bind`. It starts a `Router` on a runtime of its own (two + threads, `flysim-bus`), store root `/store`, closed policy: `flysim` may declare + and publish `fly.feed.snapshots`, `fly-edge` may only subscribe to it, and the Unix socket + `/edge.sock` is launcher-bound to `fly-edge`. +- The topic is `retained: latest`. Each publication is one snapshot: the attachments the + header lists as sealed artifacts named `frame` (`image/x-rgba`), `audio` + (`audio/x-f32le`), `spikes` (`application/x-spike-bitset`), and the header as the payload + `{"header": {...}}`. A header over 48 KiB of JSON goes as a `header` artifact instead, so + the 65,536-byte envelope limit can never make a snapshot unpublishable. +- The sim thread is untouched. The publisher reads the same `watch` slot the direct server + reads, so a slow bus skips snapshots the way a slow WebSocket client does, and nothing on + the bus can hold the loop's publish. `fly_bus_published_total` and + `fly_bus_publish_failures_total` count it. +- `fly-edge` subscribes `latest`, one in flight, with replay, rebuilds each `Snapshot` with + `feedbus::receive` and serves it with flysim's own `feed::router`. `hello`, `wants`, + drop-oldest, the idle header and the framing are therefore the same code, and the bytes are + the same bytes: `crates/fly-edge/tests/parity.rs` replays the committed stage fixtures + through both paths at once and requires byte-equal messages per client flavour. +- `fly_frames_sent_total` and `fly_feed_clients` move to the edge with the clients; it exports + them under the same names on `FLY_EDGE_METRICS_ADDR` (`127.0.0.1:9102` in + `infra/units/flyedge.service`), and watchdog check 2 follows `FLY_FEED_VIA` in `fly.env` to + them. flysim's own copies read 0 in bus mode; `/status` is otherwise unchanged. The edge also + exports `fly_edge_bus_connected`, `fly_edge_bus_lost_total`, `fly_edge_bind_failures_total` + (the bus answered but :7400 was taken, most likely by a flysim still in direct mode) and + `fly_edge_decode_failures_total`. +- Nothing about the fly changes: the readout, the reward catalog, the adapter version and the + compatibility string are byte-identical in both modes (`--print-compatibility`). + +### Amendment 2026-09-23: feed sizing + +Measured on the live fly (release build, the real cartridge): a running snapshot is a +**92,160-byte** frame (160x144 RGBA; not the 640x480 "1.2 MB" the pending list assumed), a +**17,407-byte** spike bitset (139,255 neurons), about **12,800 bytes** of audio at realtime +(1,600 stereo f32 frames per 30 Hz snapshot at 48 kHz) and a 2 to 3 KB header: **122,367 +bytes** of artifacts, about 3.7 MB/s at 30 Hz. `flysim::feedbus::limits()`: + +| Limit | Value | Why | +| --- | --- | --- | +| `max_clients` | 8 | connections, pending handshakes included: the in-process publisher and the edge's one socket seat | +| `max_subscriptions_per_client` | 4 | the edge needs 1; this is what bounds the worst case | +| `max_latest_in_flight` | 2 | the default; the edge asks for 1 | +| `max_artifact_bytes` | 4 MiB | ten seconds of audio that piled up behind a late publish | +| `max_store_bytes` | 32 MiB | tmpfs, so RAM; ten times the worst case below | +| `max_retained_bytes` | 8 MiB | one retained snapshot, plus a large header artifact | +| `max_owners_per_client` / reserved | 64 / 8 | three artifacts per delivery, a few deliveries | +| others | small counts | one topic, no services | + +A `latest` subscriber that never consumes pins at most its queued slot plus its in-flight +credits (3 snapshots); the topic pins one retained value; the publisher holds one snapshot of +staging plus the sealed copy while sealing. Only one client can subscribe at all: the publisher +is in process, and `edge.sock` is launcher-bound to `fly-edge`, which the router admits once at +a time (a second connection is refused as already connected). The worst case is therefore that +one client holding all 4 subscriptions it may open, none consuming: 4 x 3 + 1 + 2 = **15 +snapshots, about 1.8 MB**, and publication never waits on any of them (a latest subscriber is +never a reason to refuse a publication, bus-v1 section 9). `crates/fly-edge/tests/stall.rs` +measures exactly that seat: four hoarding subscriptions, a fifth refused, a second connection +refused, pacer lag 0, no publication refused. The 15 is an upper bound; the measured store +was 472,061 bytes (under 4 snapshots), because fan-out adds roots and never copies, so four +subscriptions stuck on the same publications pin the same artifacts. + +### Amendment 2026-09-23: feed store lifecycle + +- **Location.** `feed.bus_dir` (`FLY_BUS_DIR`), `/run/fly/bus` on the containers: tmpfs, + 0700, owned by `fly`, created by tmpfiles and again by flysim. The store root is + `/store`, the socket `/edge.sock`. A reboot empties it. +- **Owner.** The router lives in flysim; its lifetime is flysim's. flysim removes a stale + socket file at start, and `Router::new` removes any store directory whose `flock` is free, + i.e. one a crashed flysim left behind. A clean stop removes its own directory. The edge owns + nothing on disk. +- **Order.** flysim first, the edge after it: `flyedge.service` is `After=` and + `Requires=flysim.service`, so an explicit stop or restart of flysim (the unstick rule's + restart included) takes the edge with it. A crash-restart of flysim needs nothing: the edge + sees the connection close, drops every WebSocket client, unbinds :7400 and reconnects every + 500 ms, binding :7400 again only when the first snapshot of the new router arrives. To the + stage that is exactly a flysim restart in direct mode: refused, then back. +- **Default.** `flyedge.service` is in no target and `07-enable.sh` does not enable it; + `05-deploy.sh` writes `FLY_FEED_VIA=direct` unless the env file says otherwise, and refuses + anything but `direct` or `bus` (any case, written lowercased). The switch and the way back + are in the unit's header. The edge gets a cpuset drop-in on the page's CPUs with the other + units, so once enabled it never runs on flysim's. +- **Paths.** `feed.bus_dir` must be absolute and non-empty (checked in both modes), since + flysim and the edge each resolve it. +- **Migration order** is unchanged: the feed first; control only after the bus has carried + the feed in production for a full session. + +### Known limits (review round 1, 2026-09-23) + +Accepted for now and written down rather than fixed: + +- **Feed counters off the container.** In bus mode flysim's `:9101` reports + `fly_feed_clients` and `fly_frames_sent_total` as 0, and the edge's copies are on loopback + `:9102` only. The watchdog follows `FLY_FEED_VIA`; anything that scrapes `:9101` from off + the container (the metrics dashboard) goes blind to the feed until it also scrapes the edge. +- **Store quota is per router, not per client.** Any client on `edge.sock` may allocate + artifacts up to the store cap; a hostile process running as the same user could fill the + store and make flysim's publications fail. The loop is unaffected (a refusal is counted, never + waited on), but the feed would stall. Same-user processes are inside the trust boundary + (crate README, "Limitations"). +- **Rollback while in bus mode.** Rolling back to a release without `fly-edge` while `fly.env` + still says `bus` leaves no one on :7400, and check 2 then reads the edge's absent `:9102` and + escalates. Switch back to `direct` first (the unit header's way back), then roll back. +- **Old fixtures.** `cold-open`, `steady` and `big-moment` predate `game.scene` and cannot be a + Rust `FeedHeader`, so fixture parity covers `macros`, `shop`, `center` and `bigpad`. diff --git a/infra/05-deploy.sh b/infra/05-deploy.sh index 705fb13..bb38302 100644 --- a/infra/05-deploy.sh +++ b/infra/05-deploy.sh @@ -413,6 +413,10 @@ log "05-deploy: non-secret env files" # never drift apart (see cpuset_partition's own header comment). They are # assigned in section 0b, which needs them earlier than this for the # deploy-time cpu pinning; nothing between here and there changes them. +# Who serves the feed (docs/design/flybus.md): refused here rather than at flysim's boot. +FLY_FEED_VIA_EFFECTIVE="$(feed_via_normalize "${FLY_FEED_VIA:-}")" \ + || die "05-deploy: FLY_FEED_VIA must be 'direct' or 'bus', got '${FLY_FEED_VIA}'" + tmp_fly_env="$(mktemp)" tmp_flypush_env="$(mktemp)" trap 'rm -f "$tmp_fly_env" "$tmp_flypush_env"' EXIT @@ -499,6 +503,16 @@ trap 'rm -f "$tmp_fly_env" "$tmp_flypush_env"' EXIT # "palette"/"plan" as "macros" with a warning, and refuses an unrecognised # value outright. echo "FLY_MACRO_MODE=${FLY_MACRO_MODE:-raw}" + # Who serves the feed WebSocket (docs/design/flybus.md, "Feed over the + # bus"). "direct" is the default and is flysim binding :7400 itself, as + # every release before this knob. "bus" makes flysim publish on its + # embedded feed bus and leave :7400 to flyedge.service, which this script + # never enables: see that unit's header for the switch. Written + # unconditionally, like FLY_MACRO_MODE, so one grep says which a box runs. + # Watchdog check 2 reads this line to know whose /metrics carries the + # feed counters (flysim's :9101, or flyedge's loopback :9102). + # Validated and lowercased above (feed_via_normalize). + echo "FLY_FEED_VIA=${FLY_FEED_VIA_EFFECTIVE}" # How long a macro leaves a target alone after a walk to it aborted # (macros.md section 12.1, the Viridian stall). Only written when it is set, # because the default lives in the crate and a box that has not tuned it @@ -633,9 +647,11 @@ if [ -n "${CPUSET:-}" ]; then "leaves cpuset.cpus.effective empty and the unit unstartable." else read -r sim_cpus page_cpus encoder_cpus <<< "$(cpuset_partition "$CPUSET" "$RAYON_THREADS_EFFECTIVE" "$ENCODER_CORES_EFFECTIVE")" - log "05-deploy: cpuset partition — flysim=$sim_cpus, xvfb/flystage/flystage-web/pulse/mediamtx=$page_cpus, flycast=$encoder_cpus" + log "05-deploy: cpuset partition — flysim=$sim_cpus, xvfb/flystage/flystage-web/pulse/mediamtx/flyedge=$page_cpus, flycast=$encoder_cpus" tmp_dropin="$(mktemp)" - for u in flysim xvfb flystage flystage-web flycast pulse mediamtx; do + # flyedge is off by default, but its drop-in is written with the rest so that the day + # it is enabled it serves the page from the page's CPUs, never from flysim's. + for u in flysim xvfb flystage flystage-web flycast pulse mediamtx flyedge; do case "$u" in flysim) cpus="$sim_cpus" ;; flycast) cpus="$encoder_cpus" ;; diff --git a/infra/bin/fly-watchdog b/infra/bin/fly-watchdog index 038ed88..c21a565 100755 --- a/infra/bin/fly-watchdog +++ b/infra/bin/fly-watchdog @@ -32,6 +32,9 @@ log_info() { : "${FLY_CONTROL_URL:=http://127.0.0.1:7401}" : "${FLY_METRICS_URL:=http://127.0.0.1:9101}" +# flyedge's loopback /metrics (units/flyedge.service), read by check 2 when +# fly.env says FLY_FEED_VIA=bus. +: "${FLY_EDGE_METRICS_URL:=http://127.0.0.1:9102}" : "${FLY_STATE_HOT:=/run/fly/state}" : "${FLY_MEDIA_DIR:=/srv/fly/media}" : "${MEDIAMTX_API:=http://127.0.0.1:9997}" @@ -335,10 +338,32 @@ check_flysim() { # read-only metrics listener (infra.md section 5; not superseded by the # feed/control contracts). Flat frames counter across two passes, or zero # clients, means the page is dead/frozen even though Chromium is alive. +# +# The two counters belong to whoever serves the feed: flysim itself, or with +# FLY_FEED_VIA=bus in fly.env, flyedge (docs/design/flybus.md, "Feed over the +# bus"), which exports them under the same names. FLY_FEED_METRICS_URL in the +# watchdog's own environment overrides both. # ============================================================================ +feed_metrics_url() { + if [ -n "${FLY_FEED_METRICS_URL:-}" ]; then + echo "$FLY_FEED_METRICS_URL" + return + fi + local via="" + if [ -f "$FLY_ENV_FILE" ]; then + # Lowercased: flysim reads the value case-insensitively, so `Bus` is bus mode. + via="$(awk -F= '/^FLY_FEED_VIA=/ { print $2; exit }' "$FLY_ENV_FILE" 2>/dev/null | tr -d ' \r"' | tr '[:upper:]' '[:lower:]' || true)" + fi + if [ "$via" = "bus" ]; then + echo "$FLY_EDGE_METRICS_URL" + else + echo "$FLY_METRICS_URL" + fi +} + check_flystage() { local metrics frames clients ok=1 - metrics="$(curl -fsS "${FLY_METRICS_URL}/metrics" 2>/dev/null || true)" + metrics="$(curl -fsS "$(feed_metrics_url)/metrics" 2>/dev/null || true)" if [ -z "$metrics" ]; then ok=0 else diff --git a/infra/build/build-flysim.sh b/infra/build/build-flysim.sh index ac6385a..3a56ca7 100755 --- a/infra/build/build-flysim.sh +++ b/infra/build/build-flysim.sh @@ -79,6 +79,12 @@ log "building in $crate_dir for target-cpu=haswell (the host is E5-2660 v3, Hasw ( cd "$crate_dir" RUSTFLAGS="-C target-cpu=haswell" cargo build --release --target "$CARGO_TARGET" --bin flysim "${features_args[@]}" + # fly-edge (FLY_FEED_VIA=bus, docs/design/flybus.md): the feed WebSocket + # served from flysim's feed bus. Small, and no cargo features of its own; + # built every time so a release can switch a container onto the bus + # without a rebuild. It lands next to OUT_PATH, where package-release.sh + # looks for it. + RUSTFLAGS="-C target-cpu=haswell" cargo build --release --target "$CARGO_TARGET" --bin fly-edge ) built="${crate_dir}/target/${CARGO_TARGET}/release/flysim" @@ -106,4 +112,10 @@ fi cp "$built" "$OUT_PATH" chmod 0755 "$OUT_PATH" log "built $OUT_PATH ($(du -h "$OUT_PATH" | cut -f1))" +edge_built="${crate_dir}/target/${CARGO_TARGET}/release/fly-edge" +[ -x "$edge_built" ] || die "expected binary not found after build: $edge_built" +edge_out="$(dirname "$OUT_PATH")/fly-edge" +cp "$edge_built" "$edge_out" +chmod 0755 "$edge_out" +log "built $edge_out ($(du -h "$edge_out" | cut -f1))" log "next: infra/build/package-release.sh VERSION $OUT_PATH " diff --git a/infra/build/package-release.sh b/infra/build/package-release.sh index 718d678..0610db2 100755 --- a/infra/build/package-release.sh +++ b/infra/build/package-release.sh @@ -11,6 +11,9 @@ # # Output: OUT_DIR/flybrain-.tar.gz, laid out as # flysim (the binary, mode 0755) +# fly-edge (the feed-bus edge, mode 0755, when build-flysim.sh +# left one beside FLYSIM_BIN; flyedge.service stays +# inactive on a release without it) # stage/... (apps/stage's build output) # bridge/... (services/bridge + node_modules) # data/fafb-v783/... (the connectome, from the repo; FLY_DATASET points here) @@ -72,6 +75,13 @@ mkdir -p "$release_dir" cp "$FLYSIM_BIN" "${release_dir}/flysim" chmod 0755 "${release_dir}/flysim" +EDGE_BIN="$(dirname "$FLYSIM_BIN")/fly-edge" +if [ -x "$EDGE_BIN" ]; then + cp "$EDGE_BIN" "${release_dir}/fly-edge" + chmod 0755 "${release_dir}/fly-edge" +else + log "no fly-edge beside $FLYSIM_BIN; packaging without it (FLY_FEED_VIA=bus unavailable in this release)" +fi cp -a "$STAGE_DIR" "${release_dir}/stage" cp -a "$BRIDGE_DIR" "${release_dir}/bridge" diff --git a/infra/config/fly-tmpfiles.conf b/infra/config/fly-tmpfiles.conf index e568cb5..4aafd6a 100644 --- a/infra/config/fly-tmpfiles.conf +++ b/infra/config/fly-tmpfiles.conf @@ -6,6 +6,9 @@ d /run/fly 0750 fly fly - d /run/fly/pulse 0750 fly fly - d /run/fly/state 0750 fly fly - d /run/fly/wd 0750 fly fly - +# ADDED: the feed bus (FLY_FEED_VIA=bus, docs/design/flybus.md): flysim's +# router socket and artifact store. flysim also creates it, 0700, on start. +d /run/fly/bus 0700 fly fly - d /var/lib/fly 0750 fly fly - d /var/lib/fly/chrome 0700 fly fly - d /srv/fly/state 0750 fly fly - diff --git a/infra/env/example.env b/infra/env/example.env index fc19970..21afecc 100644 --- a/infra/env/example.env +++ b/infra/env/example.env @@ -312,6 +312,13 @@ CHAT_DENY_LIST=/srv/fly/chat-deny.txt # "palette" and "plan" are the two modes section 12 replaced; flysim still reads # either as "macros", with a warning, for one release. FLY_MACRO_MODE=raw +# --- feed path ---------------------------------------------------------------- +# Who serves ws://127.0.0.1:7400/feed (docs/design/flybus.md, "Feed over the +# bus"). direct: flysim binds it, as always. bus: flysim publishes every +# snapshot on its embedded feed bus (/run/fly/bus) and flyedge.service serves +# the same bytes; enable that unit by hand (its header has the steps). +# Watchdog check 2 follows this setting to the edge's counters by itself. +FLY_FEED_VIA=direct # How long a macro leaves a target alone after a walk to it aborted "blocked" or # "timeout" (macros.md section 12.1). Session state, so a restart offers every # target once more. Unset means the default, 10. diff --git a/infra/lib/common.sh b/infra/lib/common.sh index 675fbb7..770d8e7 100755 --- a/infra/lib/common.sh +++ b/infra/lib/common.sh @@ -154,6 +154,20 @@ require_release_tag() { # (An earlier, eight-cpu version of this same live hotfix — CPUSET= # 1,3,5,7,9,11,13,15, ENCODER_CORES=2 (the default) — gave flysim=1,3,5,7, # page=9,11, flycast=13,15; infra/tests/lint.sh checks both shapes.) +# feed_via_normalize VALUE — print FLY_FEED_VIA lowercased (empty means "direct"), or +# return 1 for anything but direct|bus. flysim itself reads the value case-insensitively +# and refuses anything else at boot, which on a container is a restart loop; 05-deploy.sh +# refuses it at deploy instead and writes the lowercased word, so watchdog check 2 and +# flysim can never read the same line two ways (docs/design/flybus.md). +feed_via_normalize() { + local via + via="$(printf '%s' "${1:-direct}" | tr '[:upper:]' '[:lower:]')" + case "$via" in + direct|bus) printf '%s\n' "$via" ;; + *) return 1 ;; + esac +} + cpuset_partition() { local cpuset="$1" rayon_threads="$2" encoder_cores="${3:-2}" local sim_cpus remainder remainder_count page_count page_cpus encoder_cpus diff --git a/infra/tests/lint.sh b/infra/tests/lint.sh index 0a45f9a..f41e9bd 100755 --- a/infra/tests/lint.sh +++ b/infra/tests/lint.sh @@ -429,6 +429,140 @@ else fi rm -rf "$lint_tmp" +# --------------------------------------------------------------------------- +# 3b2. The feed bus edge (docs/design/flybus.md, "Feed over the bus"). +# +# flyedge.service is off unless the operator switches a container to +# FLY_FEED_VIA=bus by hand, and when it is on it must follow flysim, which +# owns the router. What would break that is statically visible: the unit +# ending up in fly.target or 07-enable's list, losing its ordering on +# flysim, or the deploy no longer writing the default. Watchdog check 2's +# choice of /metrics is driven for real against a fixture fly.env. +# --------------------------------------------------------------------------- +echo "--- flyedge.service: off by default, after and bound to flysim ---" +EDGE_UNIT="$INFRA_DIR/units/flyedge.service" +if [ ! -f "$EDGE_UNIT" ]; then + fail "units/flyedge.service is missing" +else + grep -qE '^After=.*\bflysim\.service\b' "$EDGE_UNIT" \ + && pass "flyedge.service orders itself After=flysim.service" \ + || fail "flyedge.service must be After=flysim.service: flysim owns the feed router" + grep -qE '^Requires=.*\bflysim\.service\b' "$EDGE_UNIT" \ + && pass "flyedge.service Requires=flysim.service" \ + || fail "flyedge.service must Require flysim.service, so a stop or restart of flysim takes the edge with it" + grep -qE '^ExecStart=/opt/fly/current/fly-edge$' "$EDGE_UNIT" \ + && pass "flyedge.service runs the release's fly-edge" \ + || fail "flyedge.service ExecStart must be /opt/fly/current/fly-edge" + grep -qE '^ConditionPathExists=/opt/fly/current/fly-edge$' "$EDGE_UNIT" \ + && pass "flyedge.service stays inactive on a release without fly-edge" \ + || fail "flyedge.service needs ConditionPathExists=/opt/fly/current/fly-edge (a release before it has none)" + grep -qE '^Environment=FLY_EDGE_METRICS_ADDR=127\.0\.0\.1:' "$EDGE_UNIT" \ + && pass "flyedge.service keeps its metrics on loopback" \ + || fail "flyedge.service FLY_EDGE_METRICS_ADDR must be a 127.0.0.1 address" +fi +# Every unit a target's Wants=/Requires= names, with backslash continuations joined and +# comments dropped: fly.target spreads both lists over several physical lines, and the +# continuation line is exactly where a new unit would be added. +target_pulls() { + awk ' + /^[[:space:]]*[#;]/ { next } + { + line = $0 + cont = sub(/\\[[:space:]]*$/, "", line) + buf = buf line + if (cont) next + if (buf ~ /^[[:space:]]*(Wants|Requires)=/) { sub(/^[^=]*=/, "", buf); print buf } + buf = "" + } + ' "$1" | tr -s ' \t' '\n' | grep -v '^$' || true +} +if target_pulls "$INFRA_DIR/units/fly.target" | grep -qx 'flyedge.service'; then + fail "fly.target pulls flyedge.service in; it must stay off until the operator enables it" +else + pass "fly.target does not pull flyedge.service in" +fi +# The parser itself: a unit named only on a continuation line must be found, a commented one +# must not, and the real fly.target must still yield flysim.service. +tp_fixture="$(mktemp "${TMPDIR:-/tmp}/fly-lint-target.XXXXXX")" +cat > "$tp_fixture" <<'TPTARGET' +[Unit] +Wants=network-online.target xvfb.service \ + flysim.service flyedge.service +# Requires=commented.service +Requires=xvfb.service \ + pulse.service +TPTARGET +tp_units="$(target_pulls "$tp_fixture")" +if printf '%s\n' "$tp_units" | grep -qx 'flyedge.service' \ + && printf '%s\n' "$tp_units" | grep -qx 'pulse.service' \ + && ! printf '%s\n' "$tp_units" | grep -qx 'commented.service' \ + && target_pulls "$INFRA_DIR/units/fly.target" | grep -qx 'flysim.service'; then + pass "target_pulls reads continuation lines and skips comments (fixture + fly.target)" +else + fail "target_pulls missed a continuation line or read a comment: $(echo "$tp_units" | tr '\n' ' ')" +fi +rm -f "$tp_fixture" +if grep -E '^(ALWAYS_ON_UNITS|APP_UNITS)=' "$INFRA_DIR/07-enable.sh" "$INFRA_DIR/verify.sh" | grep -q 'flyedge'; then + fail "07-enable.sh or verify.sh lists flyedge.service as always-on" +else + pass "07-enable.sh and verify.sh leave flyedge.service alone" +fi +if grep -qF 'FLY_FEED_VIA_EFFECTIVE="$(feed_via_normalize "${FLY_FEED_VIA:-}")"' "$INFRA_DIR/05-deploy.sh" \ + && grep -qF 'echo "FLY_FEED_VIA=${FLY_FEED_VIA_EFFECTIVE}"' "$INFRA_DIR/05-deploy.sh"; then + pass "05-deploy.sh validates FLY_FEED_VIA and writes the normalized value" +else + fail "05-deploy.sh must run FLY_FEED_VIA through feed_via_normalize and write FLY_FEED_VIA_EFFECTIVE" +fi +# shellcheck source=../lib/common.sh +fv_out="$(bash -c '. "$1/lib/common.sh" + for v in "" direct DIRECT bus Bus BUS; do printf "%s=%s " "${v:-empty}" "$(feed_via_normalize "$v")"; done + for v in buss "bus " direct,bus; do feed_via_normalize "$v" >/dev/null && printf "ACCEPTED:%s " "$v"; done; true' _ "$INFRA_DIR" 2>&1)" +if [ "$fv_out" = "empty=direct direct=direct DIRECT=direct bus=bus Bus=bus BUS=bus " ]; then + pass "feed_via_normalize: direct|bus in any case, empty is direct, anything else refused" +else + fail "feed_via_normalize: got '$fv_out'" +fi +if grep -qE '^[[:space:]]*for u in flysim .*\bflyedge\b.*; do$' "$INFRA_DIR/05-deploy.sh"; then + pass "05-deploy.sh writes a cpuset drop-in for flyedge.service" +else + fail "05-deploy.sh cpuset loop must include flyedge (the page's CPUs, never flysim's)" +fi +if grep -qE '^Environment=FLY_FEED_VIA' "$INFRA_DIR/units/flysim.service"; then + fail "flysim.service pins FLY_FEED_VIA; it belongs to fly.env so a box can be switched by deploy" +else + pass "flysim.service leaves FLY_FEED_VIA to fly.env" +fi + +echo "--- fly-watchdog check 2: the feed counters follow FLY_FEED_VIA ---" +if ! tail -n1 "$INFRA_DIR/bin/fly-watchdog" | grep -qE '^main "\$@"$'; then + fail "fly-watchdog: expected the last line to be 'main \"\$@\"' — the check-2 fixture strips it" +else + fe_fixture="$(mktemp -d "${TMPDIR:-/tmp}/fly-lint-edge.XXXXXX")" + sed '$d' "$INFRA_DIR/bin/fly-watchdog" > "$fe_fixture/wd.sh" + feed_url_case() { + local label="$1" env_line="$2" override="$3" want="$4" got + printf '%s\n' "$env_line" > "$fe_fixture/fly.env" + got="$(FLY_ENV_FILE="$fe_fixture/fly.env" FLY_FEED_METRICS_URL="$override" \ + FLY_METRICS_URL=http://sim FLY_EDGE_METRICS_URL=http://edge \ + WD_RUN_DIR="$fe_fixture/run" WD_STATE_DIR="$fe_fixture/state" \ + TEXTFILE_DIR="$fe_fixture/textfile" \ + bash -c "source '$fe_fixture/wd.sh'; feed_metrics_url" 2>&1 || true)" + if [ "$got" = "$want" ]; then + pass "check 2 feed metrics: $label -> $got" + else + fail "check 2 feed metrics: $label: got '$got', want '$want'" + fi + } + feed_url_case "direct" "FLY_FEED_VIA=direct" "" "http://sim" + feed_url_case "no FLY_FEED_VIA line (a fly.env before it)" "FLY_GAME=pokemon-red" "" "http://sim" + feed_url_case "bus" "FLY_FEED_VIA=bus" "" "http://edge" + feed_url_case "Bus (flysim lowercases)" "FLY_FEED_VIA=Bus" "" "http://edge" + feed_url_case "BUS" "FLY_FEED_VIA=BUS" "" "http://edge" + feed_url_case "quoted bus" 'FLY_FEED_VIA="bus"' "" "http://edge" + feed_url_case "explicit override wins" "FLY_FEED_VIA=bus" "http://other" "http://other" + rm -rf "$fe_fixture" +fi + # --------------------------------------------------------------------------- # 3c. lib/common.sh cpuset_partition — the three-way cpuset split used by # 05-deploy.sh section 3b (flysim / page-capture / flycast). Run as its own diff --git a/infra/units/flyedge.service b/infra/units/flyedge.service new file mode 100644 index 0000000..a0d0f17 --- /dev/null +++ b/infra/units/flyedge.service @@ -0,0 +1,56 @@ +# infra/units/flyedge.service — pushed to /etc/systemd/system/flyedge.service. +# +# The feed WebSocket served from flysim's feed bus (docs/design/flybus.md, +# "Feed over the bus"; services/flysim/crates/fly-edge). DISABLED BY DEFAULT: +# it is in no target's Wants=/Requires= and 07-enable.sh does not enable it. +# With FLY_FEED_VIA=direct (the default, written into /etc/fly/fly.env by +# 05-deploy.sh) flysim binds 127.0.0.1:7400 itself and this unit has nothing +# to do. To move the feed onto the bus on one container: +# +# 1. FLY_FEED_VIA=bus in the env file, then 05-deploy.sh (rewrites fly.env); +# 2. systemctl enable --now flyedge.service; systemctl restart flysim.service +# (flysim stops binding :7400, the edge binds it once the first snapshot +# is on the bus); +# 3. nothing for the watchdog: check 2 reads FLY_FEED_VIA from fly.env and +# follows the feed counters to this unit's loopback /metrics. +# +# Back: FLY_FEED_VIA=direct, deploy, systemctl disable --now flyedge.service, +# restart flysim. +# +# Ordering (docs/design/flybus.md, amendment "Feed store lifecycle"): flysim +# owns the router and its store under /run/fly/bus, so it starts first and +# the edge follows it. Requires= makes an explicit stop or restart of flysim +# (the unstick rule's `systemctl restart flysim.service` included) stop or +# restart the edge with it. A crash-restart of flysim is covered by the edge +# itself: it drops its clients, unbinds :7400 and reconnects every 500 ms, +# so nothing here has to be restarted by hand. The edge holds no state; the +# store is flysim's and a new router removes the previous one's directory. +[Unit] +Description=flyedge: the feed WebSocket served from flysim's feed bus +After=flysim.service +Requires=flysim.service +# A release that predates fly-edge has no binary; stay cleanly inactive +# rather than restart-looping (the flybridge.service header explains why a +# Condition, not a start limit). +ConditionPathExists=/opt/fly/current/fly-edge + +[Service] +Type=simple +User=fly +# FLY_FEED_VIA, FLY_BUS_DIR and the rest of flysim's configuration: the edge +# reads the same file so the two cannot disagree about the port or the bus. +EnvironmentFile=/etc/fly/fly.env +Environment=FLY_FEED_BIND=127.0.0.1:7400 +Environment=FLY_BUS_DIR=/run/fly/bus +# Its own read-only /metrics and /healthz, for watchdog check 2 in bus mode. +# Loopback only: nothing off the container needs the edge's counters. +Environment=FLY_EDGE_METRICS_ADDR=127.0.0.1:9102 +ExecStart=/opt/fly/current/fly-edge +Restart=always +RestartSec=2 +# A few snapshots in flight and a WebSocket per client; the store itself is +# flysim's (tmpfs, bounded by feedbus::limits at 32 MiB). +MemoryMax=256M + +[Install] +WantedBy=fly.target diff --git a/infra/units/flysim.service b/infra/units/flysim.service index 8363bed..df2dc48 100644 --- a/infra/units/flysim.service +++ b/infra/units/flysim.service @@ -30,6 +30,10 @@ WatchdogSec=30 User=fly EnvironmentFile=/etc/fly/fly.env Environment=FLY_FEED_BIND=127.0.0.1:7400 +# Used only with FLY_FEED_VIA=bus (fly.env; default direct): the embedded +# feed router's socket and artifact store, on tmpfs. flyedge.service names +# the same directory. docs/design/flybus.md, "Feed over the bus". +Environment=FLY_BUS_DIR=/run/fly/bus Environment=FLY_CONTROL_BIND=127.0.0.1:7401 Environment=FLY_METRICS_ADDR=0.0.0.0:9101 Environment=FLY_STATE_HOT=/run/fly/state diff --git a/services/flysim/Cargo.lock b/services/flysim/Cargo.lock index 7399620..322d886 100644 --- a/services/flysim/Cargo.lock +++ b/services/flysim/Cargo.lock @@ -416,6 +416,25 @@ dependencies = [ "serde", ] +[[package]] +name = "fly-edge" +version = "0.1.1" +dependencies = [ + "anyhow", + "axum", + "clap", + "flate2", + "flybus", + "flysim", + "futures-util", + "serde_json", + "tempfile", + "tokio", + "tokio-tungstenite", + "tracing", + "tracing-subscriber", +] + [[package]] name = "fly-session" version = "0.1.1" @@ -484,6 +503,7 @@ dependencies = [ "fly-session-types", "flybrain-core", "flybrain-gb", + "flybus", "futures-util", "jsonschema", "serde", diff --git a/services/flysim/Cargo.toml b/services/flysim/Cargo.toml index dec6a38..2def2d3 100644 --- a/services/flysim/Cargo.toml +++ b/services/flysim/Cargo.toml @@ -1,6 +1,7 @@ [workspace] resolver = "3" members = [ + "crates/fly-edge", "crates/fly-session", "crates/fly-session-types", "crates/flybrain-core", diff --git a/services/flysim/crates/fly-edge/Cargo.toml b/services/flysim/crates/fly-edge/Cargo.toml new file mode 100644 index 0000000..90ba894 --- /dev/null +++ b/services/flysim/crates/fly-edge/Cargo.toml @@ -0,0 +1,36 @@ +[package] +name = "fly-edge" +version.workspace = true +edition = "2024" +rust-version.workspace = true +license.workspace = true +publish = false +description = "The feed WebSocket (:7400) served from flysim's feed bus (FLY_FEED_VIA=bus)." + +[lib] +name = "fly_edge" +path = "src/lib.rs" + +[[bin]] +name = "fly-edge" +path = "src/main.rs" + +[dependencies] +# The feed server and the bus encoding are flysim's own modules (`feed`, `feedbus`, +# `snapshot`), so the edge writes the WebSocket bytes with the code flysim uses in direct mode. +flysim = { path = "../flysim" } +flybus = { path = "../flybus" } + +anyhow = "1.0" +axum = { version = "0.8", features = ["ws"] } +clap = { version = "4.5", features = ["derive"] } +tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "time", "signal", "macros"] } +tracing = "0.1" +tracing-subscriber = { version = "0.3", features = ["env-filter"] } + +[dev-dependencies] +flate2 = { workspace = true } +futures-util = "0.3" +serde_json = { workspace = true } +tempfile = "3" +tokio-tungstenite = "0.29" diff --git a/services/flysim/crates/fly-edge/src/lib.rs b/services/flysim/crates/fly-edge/src/lib.rs new file mode 100644 index 0000000..cdf9375 --- /dev/null +++ b/services/flysim/crates/fly-edge/src/lib.rs @@ -0,0 +1,344 @@ +//! `fly-edge`: the feed WebSocket, served from flysim's feed bus. +//! +//! With `FLY_FEED_VIA=bus` flysim does not bind the feed port. It publishes every snapshot on an +//! embedded flybus router (`flysim::feedbus`), and this process subscribes and serves +//! `ws:///feed` to the stage, the bridge and tests. The contract is still +//! `docs/feed-protocol.md`, byte for byte: the snapshots come off the bus as the same +//! `flysim::snapshot::Snapshot` values and are written by the same `flysim::feed` server, so the +//! per-client `hello`, `wants`, drop-oldest and idle cadence are flysim's own code. +//! +//! Lifecycle (`docs/design/flybus.md`, amendment "Feed store lifecycle"): +//! +//! - the feed port is bound only once the first snapshot has arrived, so before that a client +//! is refused exactly as it would be by a flysim that has not started; +//! - when the bus goes away (flysim stopped or restarted) the edge drops every client and unbinds +//! the port, again exactly what a stopped flysim looks like to the stage, then reconnects every +//! `retry` until a router answers. It never serves a stale snapshot as if it were live. + +use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +use anyhow::{Context, Result, anyhow}; +use axum::extract::State; +use axum::http::StatusCode; +use axum::response::IntoResponse; +use axum::routing::get; +use flybus::{Client, ClientConfig, SubscriptionConfig}; +use flysim::feed::{self, FeedState}; +use flysim::feedbus; +use flysim::metrics::{Metrics, metric}; +use tokio::sync::{oneshot, watch}; + +/// What the edge needs to know. Built from flysim's own configuration, so both processes read +/// one environment file and cannot disagree about the port, the bus directory or the cadence. +#[derive(Debug, Clone)] +pub struct EdgeConfig { + /// `feed.bus_dir`: the router's socket and store root. + pub bus_dir: PathBuf, + /// `feed.bind`, the port flysim leaves alone in bus mode. + pub feed_bind: SocketAddr, + /// `1 / loop.idle_snapshot_hz`, the protocol's header-only cadence. + pub idle_period: Duration, + /// `FLY_EDGE_METRICS_ADDR`: `/metrics` and `/healthz` for the watchdog, when set. + pub metrics_bind: Option, + /// Delay between attempts to reach the bus. + pub retry: Duration, +} + +impl EdgeConfig { + pub fn from_flysim(config: &flysim::config::Config, metrics_bind: Option) -> Self { + Self { + bus_dir: config.feed.bus_dir.clone(), + feed_bind: config.feed.bind, + idle_period: config.publish_periods().1, + metrics_bind, + retry: Duration::from_millis(500), + } + } +} + +/// The edge's counters. `feed` is the same `Metrics` type flysim uses, so +/// `fly_frames_sent_total` and `fly_feed_clients` mean exactly what they mean there. +#[derive(Debug, Default)] +pub struct EdgeMetrics { + pub feed: Arc, + /// Snapshots taken off the bus and handed to the feed server. + pub snapshots: AtomicU64, + /// 1 while subscribed and serving. + pub connected: AtomicU64, + /// Times a serving session ended because the bus went away. + pub bus_lost: AtomicU64, + /// Publications that could not be turned back into a snapshot. + pub decode_failures: AtomicU64, + /// Sessions that reached the bus but could not bind the feed port. + pub bind_failures: AtomicU64, +} + +impl EdgeMetrics { + pub fn render(&self) -> String { + let mut out = String::with_capacity(1_024); + let feed = &self.feed; + metric( + &mut out, + "fly_frames_sent_total", + "counter", + "Feed snapshots written to a client socket.", + Metrics::get(&feed.frames_sent), + ); + metric( + &mut out, + "fly_feed_clients", + "gauge", + "Feed clients currently subscribed.", + feed.clients(), + ); + metric( + &mut out, + "fly_feed_dropped_total", + "counter", + "Snapshots superseded before a slow client could be sent them.", + Metrics::get(&feed.feed_dropped), + ); + metric( + &mut out, + "fly_edge_snapshots_total", + "counter", + "Snapshots taken off the feed bus.", + self.snapshots.load(Ordering::Relaxed), + ); + metric( + &mut out, + "fly_edge_bus_connected", + "gauge", + "1 while the edge is subscribed to the feed bus and serving.", + self.connected.load(Ordering::Relaxed), + ); + metric( + &mut out, + "fly_edge_bus_lost_total", + "counter", + "Serving sessions ended by the feed bus going away.", + self.bus_lost.load(Ordering::Relaxed), + ); + metric( + &mut out, + "fly_edge_decode_failures_total", + "counter", + "Feed bus publications that did not decode to a snapshot.", + self.decode_failures.load(Ordering::Relaxed), + ); + metric( + &mut out, + "fly_edge_bind_failures_total", + "counter", + "Times the bus was reachable but the feed port could not be bound.", + self.bind_failures.load(Ordering::Relaxed), + ); + out + } +} + +/// Serve until the process is stopped. Only a metrics listener that cannot bind is fatal; +/// everything about the bus is retried. +pub async fn run(config: EdgeConfig, metrics: Arc) -> Result<()> { + if let Some(addr) = config.metrics_bind { + let listener = tokio::net::TcpListener::bind(addr) + .await + .with_context(|| format!("binding the edge metrics listener on {addr}"))?; + let app = axum::Router::new() + .route("/metrics", get(prometheus)) + .route("/healthz", get(healthz)) + .with_state(Arc::clone(&metrics)); + tokio::spawn(async move { + if let Err(error) = axum::serve(listener, app).await { + tracing::error!(%error, "the edge metrics listener stopped"); + } + }); + } + // One line per outage of each kind, not one per retry. + let mut last: Option<&'static str> = None; + loop { + let end = session(&config, &metrics).await; + match &end { + SessionEnd::BusLost => { + tracing::warn!("the feed bus went away; clients dropped, reconnecting"); + } + SessionEnd::Unreachable(error) if last != Some(end.kind()) => { + tracing::info!(error = format!("{error:#}"), "waiting for the feed bus"); + } + SessionEnd::BindFailed(error) if last != Some(end.kind()) => { + // The bus is fine; the port is not ours. Most likely flysim is still in direct + // mode and holds it (FLY_FEED_VIA is not bus), or another process does. + tracing::warn!( + error = format!("{error:#}"), + "the bus is up but the feed port cannot be bound; retrying" + ); + } + _ => {} + } + last = match end { + SessionEnd::BusLost => None, + other => Some(other.kind()), + }; + tokio::time::sleep(config.retry).await; + } +} + +/// Why a [`session`] ended. +enum SessionEnd { + /// No router answered, or it closed before the first snapshot. Nothing was served. + Unreachable(anyhow::Error), + /// Subscribed and holding a snapshot, but the feed port could not be bound. + BindFailed(anyhow::Error), + /// A session that served has ended because the bus went away. + BusLost, +} + +impl SessionEnd { + fn kind(&self) -> &'static str { + match self { + Self::Unreachable(_) => "unreachable", + Self::BindFailed(_) => "bind", + Self::BusLost => "lost", + } + } +} + +async fn prometheus(State(metrics): State>) -> impl IntoResponse { + ( + [( + axum::http::header::CONTENT_TYPE, + "text/plain; version=0.0.4", + )], + metrics.render(), + ) +} + +async fn healthz(State(metrics): State>) -> impl IntoResponse { + if metrics.connected.load(Ordering::Relaxed) == 1 { + (StatusCode::OK, "ok") + } else { + (StatusCode::SERVICE_UNAVAILABLE, "waiting for the feed bus") + } +} + +/// One subscription's lifetime. +async fn session(config: &EdgeConfig, metrics: &EdgeMetrics) -> SessionEnd { + let (subscription, client, first) = match subscribe(config, metrics).await { + Ok(subscribed) => subscribed, + Err(error) => return SessionEnd::Unreachable(error), + }; + let listener = match tokio::net::TcpListener::bind(config.feed_bind).await { + Ok(listener) => listener, + Err(error) => { + metrics.bind_failures.fetch_add(1, Ordering::Relaxed); + return SessionEnd::BindFailed( + anyhow::Error::new(error) + .context(format!("binding the feed listener on {}", config.feed_bind)), + ); + } + }; + serve(config, metrics, client, subscription, first, listener).await; + SessionEnd::BusLost +} + +/// Connect, subscribe and wait for the first snapshot that decodes. +async fn subscribe( + config: &EdgeConfig, + metrics: &EdgeMetrics, +) -> Result<(flybus::Subscription, Client, flysim::snapshot::Snapshot)> { + let client = Client::connect_unix( + feedbus::socket_path(&config.bus_dir), + ClientConfig::new(feedbus::EDGE, feedbus::store_root(&config.bus_dir)), + ) + .await + .map_err(|error| anyhow!("connecting to the feed bus: {error}"))?; + // One in flight: while a snapshot is being copied out, the next one waits in the single + // latest slot and anything newer replaces it. The edge is never more than one behind. + let mut subscription = client + .subscribe( + feedbus::TOPIC, + SubscriptionConfig::latest().in_flight(1).replay(true), + ) + .await + .map_err(|error| anyhow!("subscribing to {}: {error}", feedbus::TOPIC))?; + let first = loop { + let message = subscription + .next() + .await + .ok_or_else(|| anyhow!("the feed bus closed before the first snapshot"))?; + match feedbus::receive(&message).await { + Ok(snapshot) => break snapshot, + Err(error) => { + metrics.decode_failures.fetch_add(1, Ordering::Relaxed); + tracing::warn!(%error, "a feed bus publication did not decode"); + } + } + }; + Ok((subscription, client, first)) +} + +/// Serve `listener` from `subscription` until the bus goes away. +async fn serve( + config: &EdgeConfig, + metrics: &EdgeMetrics, + client: Client, + mut subscription: flybus::Subscription, + first: flysim::snapshot::Snapshot, + listener: tokio::net::TcpListener, +) { + let (snapshots, receiver) = watch::channel(Arc::new(first)); + metrics.snapshots.fetch_add(1, Ordering::Relaxed); + tracing::info!(feed = %config.feed_bind, bus = %config.bus_dir.display(), "serving the feed from the bus"); + metrics.connected.store(1, Ordering::Relaxed); + + let state = FeedState { + snapshots: receiver, + metrics: Arc::clone(&metrics.feed), + idle_period: config.idle_period, + }; + let (stop, stopped) = oneshot::channel::<()>(); + let server = tokio::spawn(async move { + let result = axum::serve(listener, feed::router(state)) + .with_graceful_shutdown(async move { + let _ = stopped.await; + }) + .await; + if let Err(error) = result { + tracing::error!(%error, "the feed listener stopped"); + } + }); + + while let Some(message) = subscription.next().await { + match feedbus::receive(&message).await { + Ok(snapshot) => { + drop(message); + snapshots.send_replace(Arc::new(snapshot)); + metrics.snapshots.fetch_add(1, Ordering::Relaxed); + } + Err(error) => { + metrics.decode_failures.fetch_add(1, Ordering::Relaxed); + tracing::warn!(%error, "a feed bus publication did not decode"); + if client.closed().is_some() { + break; + } + } + } + } + + // The bus is gone. Dropping the sender ends every client's pump (a closed stream, as when + // flysim itself stops), and the graceful shutdown unbinds the port. + metrics.connected.store(0, Ordering::Relaxed); + metrics.bus_lost.fetch_add(1, Ordering::Relaxed); + drop(snapshots); + let _ = stop.send(()); + if tokio::time::timeout(Duration::from_secs(5), server) + .await + .is_err() + { + tracing::warn!("the feed listener took more than 5 s to stop"); + } +} diff --git a/services/flysim/crates/fly-edge/src/main.rs b/services/flysim/crates/fly-edge/src/main.rs new file mode 100644 index 0000000..067637b --- /dev/null +++ b/services/flysim/crates/fly-edge/src/main.rs @@ -0,0 +1,81 @@ +//! `fly-edge`: serve the feed WebSocket from flysim's feed bus. +//! +//! ```sh +//! FLY_FEED_VIA=bus flysim & +//! fly-edge +//! ``` +//! +//! Configured through the same environment as flysim (`FLY_FEED_BIND`, `FLY_BUS_DIR`, +//! `FLYSIM_LOOP_IDLE_SNAPSHOT_HZ`, or `--config flysim.toml`), plus `FLY_EDGE_METRICS_ADDR` for +//! its own `/metrics` and `/healthz`. `infra/units/flyedge.service` runs it with no arguments. + +use std::sync::Arc; + +use anyhow::{Context, Result}; +use clap::Parser; +use fly_edge::{EdgeConfig, EdgeMetrics}; + +#[derive(Debug, Parser)] +#[command( + name = "fly-edge", + about = "The feed WebSocket, served from flysim's feed bus.", + version +)] +struct Args { + /// Path to `flysim.toml`, read for `[feed]` and `[loop]`. Environment overrides apply as + /// they do for flysim. + #[arg(long, value_name = "PATH")] + config: Option, +} + +fn main() -> Result<()> { + let args = Args::parse(); + init_tracing(); + let config = flysim::config::Config::load(args.config.as_deref())?; + let metrics_bind = match std::env::var("FLY_EDGE_METRICS_ADDR") { + Ok(value) if !value.is_empty() => Some(value.parse().with_context(|| { + format!("FLY_EDGE_METRICS_ADDR: {value:?} is not a host:port address") + })?), + _ => None, + }; + let edge = EdgeConfig::from_flysim(&config, metrics_bind); + if config.feed.via != flysim::config::FeedVia::Bus { + tracing::warn!( + "FLY_FEED_VIA is not \"bus\": flysim serves the feed itself and binds {}; \ + this edge will wait for a bus that is not there", + edge.feed_bind + ); + } + tracing::info!(config = ?edge, "fly-edge starting"); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_name("fly-edge") + .enable_all() + .build() + .context("building the tokio runtime")?; + runtime.block_on(async move { + let metrics = Arc::new(EdgeMetrics::default()); + let mut terminate = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .context("installing the SIGTERM handler")?; + tokio::select! { + result = fly_edge::run(edge, metrics) => result, + _ = tokio::signal::ctrl_c() => { tracing::info!("SIGINT: shutting down"); Ok(()) } + _ = terminate.recv() => { tracing::info!("SIGTERM: shutting down"); Ok(()) } + } + }) +} + +/// Logs to stderr, like flysim, under `FLY_EDGE_LOG` (or `RUST_LOG`). +fn init_tracing() { + use tracing_subscriber::EnvFilter; + let filter = EnvFilter::try_from_env("FLY_EDGE_LOG") + .or_else(|_| EnvFilter::try_from_default_env()) + .unwrap_or_else(|_| EnvFilter::new("info")); + tracing_subscriber::fmt() + .with_env_filter(filter) + .with_writer(std::io::stderr) + .with_target(false) + .init(); +} diff --git a/services/flysim/crates/fly-edge/tests/common/mod.rs b/services/flysim/crates/fly-edge/tests/common/mod.rs new file mode 100644 index 0000000..e35c3ed --- /dev/null +++ b/services/flysim/crates/fly-edge/tests/common/mod.rs @@ -0,0 +1,246 @@ +#![allow(dead_code)] + +//! Shared pieces of the edge tests: the committed `.flyfeed` fixtures as snapshots, a feed +//! client, the two serving paths side by side, and a `.flyfeed` writer. + +use std::io::Read as _; +use std::net::SocketAddr; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use fly_edge::{EdgeConfig, EdgeMetrics}; +use flysim::feed::FeedState; +use flysim::feedbus; +use flysim::metrics::Metrics; +use flysim::snapshot::{AttachmentKind, FeedHeader, Snapshot}; +use futures_util::{SinkExt as _, StreamExt as _}; +use tokio::sync::watch; +use tokio_tungstenite::tungstenite::Message as WsMessage; + +pub fn repo_root() -> PathBuf { + PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../../../..") + .canonicalize() + .expect("the repository root is above services/flysim/crates/fly-edge") +} + +/// `u32 LE headerLength | header JSON | (u32 LE length | bytes)*`, split. +pub fn split(message: &[u8]) -> (&[u8], Vec<&[u8]>) { + let read = |at: usize| u32::from_le_bytes(message[at..at + 4].try_into().unwrap()) as usize; + let header_len = read(0); + let header = &message[4..4 + header_len]; + let mut at = 4 + header_len; + let mut attachments = Vec::new(); + while at < message.len() { + let len = read(at); + attachments.push(&message[at + 4..at + 4 + len]); + at += 4 + len; + } + (header, attachments) +} + +/// A wire message back into the snapshot that produced it, or `None` when its header predates +/// fields the Rust producer always writes (the three oldest fixtures lack `game.scene`). +pub fn snapshot_of(message: &[u8]) -> Option { + let (header, attachments) = split(message); + let header: FeedHeader = serde_json::from_slice(header).ok()?; + let mut snapshot = Snapshot { + header, + frame: Arc::new(Vec::new()), + audio: Arc::new(Vec::new()), + spikes: Arc::new(Vec::new()), + }; + for (kind, bytes) in snapshot + .header + .attachments + .clone() + .into_iter() + .zip(attachments) + { + let bytes = Arc::new(bytes.to_vec()); + match kind { + AttachmentKind::Frame => snapshot.frame = bytes, + AttachmentKind::Audio => snapshot.audio = bytes, + AttachmentKind::Spikes => snapshot.spikes = bytes, + } + } + Some(snapshot) +} + +/// Every record of `apps/stage/public/fixtures/.flyfeed.gz`, as wire messages. +pub fn fixture_messages(name: &str) -> Vec> { + let path = repo_root().join(format!("apps/stage/public/fixtures/{name}.flyfeed.gz")); + let gz = std::fs::read(&path).unwrap_or_else(|error| panic!("{}: {error}", path.display())); + let mut bytes = Vec::new(); + flate2::read::GzDecoder::new(gz.as_slice()) + .read_to_end(&mut bytes) + .unwrap(); + assert_eq!(&bytes[..8], b"FLYFEED\0", "{name}"); + let read = |at: usize| u32::from_le_bytes(bytes[at..at + 4].try_into().unwrap()) as usize; + assert_eq!(read(8), 1, "{name}: container version"); + let mut at = 16 + read(12); + let mut out = Vec::new(); + while at < bytes.len() { + let len = read(at); + out.push(bytes[at + 4..at + 4 + len].to_vec()); + at += 4 + len; + } + out +} + +/// A `.flyfeed` file of `messages` (`packages/feed/src/fixture.ts`). +pub fn encode_flyfeed(name: &str, source: &str, messages: &[Vec]) -> Vec { + let wall = |message: &Vec| -> u64 { + let header: serde_json::Value = serde_json::from_slice(split(message).0).unwrap(); + header["wallMs"].as_u64().unwrap_or(0) + }; + let duration = match (messages.first(), messages.last()) { + (Some(first), Some(last)) => wall(last).saturating_sub(wall(first)), + _ => 0, + }; + let manifest = serde_json::json!({ + "name": name, + "protocol": 1, + "recordedAt": "1970-01-01T00:00:00.000Z", + "source": source, + "snapshotCount": messages.len(), + "durationMs": duration, + "hz": 30, + "attachmentPolicy": { + "frame": { "stride": 1 }, + "audio": { "stride": 1 }, + "spikes": { "stride": 1 } + }, + }); + let manifest = serde_json::to_vec(&manifest).unwrap(); + let mut out = b"FLYFEED\0".to_vec(); + out.extend_from_slice(&1u32.to_le_bytes()); + out.extend_from_slice(&(manifest.len() as u32).to_le_bytes()); + out.extend_from_slice(&manifest); + for message in messages { + out.extend_from_slice(&(message.len() as u32).to_le_bytes()); + out.extend_from_slice(message); + } + out +} + +pub fn free_port() -> SocketAddr { + std::net::TcpListener::bind("127.0.0.1:0") + .unwrap() + .local_addr() + .unwrap() +} + +pub type Ws = + tokio_tungstenite::WebSocketStream>; + +/// Connect and say `hello`, retrying while the port is not bound yet (the edge binds only once +/// its first snapshot has arrived). +pub async fn connect(addr: SocketAddr, wants: &[&str]) -> Ws { + let url = format!("ws://{addr}/feed"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(20); + loop { + match tokio_tungstenite::connect_async(&url).await { + Ok((mut ws, _)) => { + let hello = serde_json::json!({ "protocol": 1, "client": "test", "wants": wants }); + ws.send(WsMessage::Text(hello.to_string().into())) + .await + .unwrap(); + return ws; + } + Err(error) => { + assert!(tokio::time::Instant::now() < deadline, "{url}: {error}"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + } + } +} + +/// The next binary message, within `within`. +pub async fn next_binary(ws: &mut Ws, within: Duration) -> Vec { + let deadline = tokio::time::Instant::now() + within; + loop { + let message = tokio::time::timeout_at(deadline, ws.next()) + .await + .expect("a snapshot in time") + .expect("the feed stays open") + .expect("a well-formed frame"); + if let WsMessage::Binary(bytes) = message { + return bytes.to_vec(); + } + } +} + +pub fn seq_of(message: &[u8]) -> u64 { + let header: serde_json::Value = serde_json::from_slice(split(message).0).unwrap(); + header["seq"].as_u64().unwrap() +} + +/// The same watch slot served both ways at once: flysim's direct server on `direct`, and the +/// bus (router, publisher, edge) on `edge`. Owns its runtime-side tasks through the handles. +pub struct Paths { + pub snapshots: watch::Sender>, + pub direct: SocketAddr, + pub edge: SocketAddr, + pub publisher_metrics: Arc, + pub edge_metrics: Arc, + pub bus_dir: tempfile::TempDir, + pub bus: feedbus::BusFeed, +} + +/// Idle cadence long enough that no test sees a header repeated for idleness. +pub const NO_IDLE: Duration = Duration::from_secs(3_600); + +/// Start both paths over `first`. `edge` false leaves the edge out (a test then plays its part). +pub async fn start(first: Snapshot, with_edge: bool) -> Paths { + let bus_dir = tempfile::tempdir().unwrap(); + let (snapshots, receiver) = watch::channel(Arc::new(first)); + let bus = feedbus::start_router(bus_dir.path()).await.unwrap(); + let publisher_metrics = Arc::new(Metrics::default()); + tokio::spawn(feedbus::run_publisher( + bus.router.clone(), + receiver.clone(), + Arc::clone(&publisher_metrics), + )); + + let direct = free_port(); + let listener = tokio::net::TcpListener::bind(direct).await.unwrap(); + let state = FeedState { + snapshots: receiver, + metrics: Arc::new(Metrics::default()), + idle_period: NO_IDLE, + }; + tokio::spawn(async move { axum::serve(listener, flysim::feed::router(state)).await }); + + let edge = free_port(); + let edge_metrics = Arc::new(EdgeMetrics::default()); + if with_edge { + let config = EdgeConfig { + bus_dir: bus_dir.path().to_path_buf(), + feed_bind: edge, + idle_period: NO_IDLE, + metrics_bind: None, + retry: Duration::from_millis(50), + }; + tokio::spawn(fly_edge::run(config, Arc::clone(&edge_metrics))); + } + Paths { + snapshots, + direct, + edge, + publisher_metrics, + edge_metrics, + bus_dir, + bus, + } +} + +pub fn out_dir() -> Option { + std::env::var_os("FLY_EDGE_PARITY_OUT").map(PathBuf::from) +} + +pub fn write(path: &Path, bytes: &[u8]) { + std::fs::create_dir_all(path.parent().unwrap()).unwrap(); + std::fs::write(path, bytes).unwrap(); +} diff --git a/services/flysim/crates/fly-edge/tests/parity.rs b/services/flysim/crates/fly-edge/tests/parity.rs new file mode 100644 index 0000000..a041500 --- /dev/null +++ b/services/flysim/crates/fly-edge/tests/parity.rs @@ -0,0 +1,301 @@ +//! Parity: a feed served through the bus and `fly-edge` is the feed flysim serves directly. +//! +//! The committed stage fixtures (`apps/stage/public/fixtures/*.flyfeed.gz`, the recordings the +//! stage's e2e suite plays) are fed snapshot by snapshot into one watch slot, served both ways at +//! once, and recorded by one client per path and per `wants` flavour: the stage's (everything), +//! the bridge's (nothing) and a frame-only one. The two recordings must match: headers equal +//! apart from wall-time fields and attachments byte-equal -- and in fact the whole messages are +//! byte-equal, because both are written by `Snapshot::encode` from equal snapshots. The edge's +//! attachments must also equal the fixture's own. +//! +//! `FLY_EDGE_PARITY_OUT=` also writes each pair of recordings as `.flyfeed` files, which +//! `packages/feed`'s `decodeFlyfeed` reads. `FLY_EDGE_PARITY_ALL=1` replays whole fixtures +//! instead of their first 400 snapshots. + +mod common; + +use std::time::Duration; + +use common::*; +use serde_json::Value; + +/// The fixtures whose headers carry every field the Rust producer writes. The three older ones +/// (`cold-open`, `steady`, `big-moment`) predate `game.scene` and cannot be a Rust `Snapshot`. +const FIXTURES: [&str; 4] = ["macros", "shop", "center", "bigpad"]; +const WANTS: [(&str, &[&str]); 3] = [ + ("all", &["frame", "audio", "spikes"]), + ("none", &[]), + ("frame", &["frame"]), +]; + +/// The header with every `wallMs` removed, at any depth. +fn without_wall_time(header: &[u8]) -> Value { + fn strip(value: &mut Value) { + match value { + Value::Object(map) => { + map.remove("wallMs"); + map.values_mut().for_each(strip); + } + Value::Array(items) => items.iter_mut().for_each(strip), + _ => {} + } + } + let mut value: Value = serde_json::from_slice(header).unwrap(); + strip(&mut value); + value +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_edge_writes_the_bytes_the_direct_feed_writes_for_every_committed_fixture() { + let limit = if std::env::var_os("FLY_EDGE_PARITY_ALL").is_some() { + usize::MAX + } else { + 400 + }; + for name in FIXTURES { + let snapshots: Vec<_> = fixture_messages(name) + .iter() + .take(limit) + .map(|message| { + ( + message.clone(), + snapshot_of(message).unwrap_or_else(|| panic!("{name}")), + ) + }) + .collect(); + assert!( + snapshots.len() >= 100, + "{name}: {} snapshots", + snapshots.len() + ); + + let paths = start(snapshots[0].1.clone(), true).await; + let mut clients = Vec::new(); + for (flavour, wants) in WANTS { + let direct = connect(paths.direct, wants).await; + let edge = connect(paths.edge, wants).await; + clients.push((flavour, direct, edge, Vec::new(), Vec::new())); + } + + // Lockstep: publish one snapshot, wait until every client has it. Nothing is superseded, + // so both recordings are complete and comparable message by message. + for (index, (_, snapshot)) in snapshots.iter().enumerate() { + if index > 0 { + paths + .snapshots + .send_replace(std::sync::Arc::new(snapshot.clone())); + } + for (_, direct, edge, direct_log, edge_log) in &mut clients { + for (ws, log) in [ + (&mut *direct, &mut *direct_log), + (&mut *edge, &mut *edge_log), + ] { + let message = next_binary(ws, Duration::from_secs(20)).await; + assert_eq!(seq_of(&message), snapshot.header.seq, "{name} #{index}"); + log.push(message); + } + } + } + + for (flavour, _, _, direct_log, edge_log) in &clients { + assert_eq!(direct_log.len(), snapshots.len()); + assert_eq!(edge_log.len(), direct_log.len()); + for (index, (direct, edge)) in direct_log.iter().zip(edge_log).enumerate() { + let (direct_header, direct_attachments) = split(direct); + let (edge_header, edge_attachments) = split(edge); + assert_eq!( + without_wall_time(direct_header), + without_wall_time(edge_header), + "{name}/{flavour} #{index}: headers" + ); + assert_eq!( + direct_attachments, edge_attachments, + "{name}/{flavour} #{index}: attachments" + ); + // The stronger fact: the whole message, wall times included, is the same bytes. + assert!(direct == edge, "{name}/{flavour} #{index}: messages differ"); + if *flavour == "all" { + let (_, fixture_attachments) = split(&snapshots[index].0); + assert_eq!( + edge_attachments, fixture_attachments, + "{name} #{index}: vs the fixture" + ); + } + } + if let Some(dir) = out_dir() { + write( + &dir.join(format!("{name}-{flavour}-direct.flyfeed")), + &encode_flyfeed(name, "flysim direct", direct_log), + ); + write( + &dir.join(format!("{name}-{flavour}-edge.flyfeed")), + &encode_flyfeed(name, "flysim bus + fly-edge", edge_log), + ); + } + } + let published = flysim::metrics::Metrics::get(&paths.publisher_metrics.bus_published); + assert!( + published >= snapshots.len() as u64, + "{name}: {published} published" + ); + assert_eq!( + flysim::metrics::Metrics::get(&paths.publisher_metrics.bus_publish_failures), + 0 + ); + eprintln!( + "{name}: {} snapshots x {} flavours identical on both paths", + snapshots.len(), + WANTS.len() + ); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_header_too_large_for_an_envelope_travels_as_an_artifact_and_arrives_intact() { + let mut snapshot = snapshot_of(&fixture_messages("macros")[10]).unwrap(); + // Far past flybus's 65,536-byte envelope: 400 events of 200 characters. + for id in 0..400u64 { + snapshot.header.events.push(flysim::snapshot::FeedEvent { + id: 10_000 + id, + wall_ms: 1_757_000_000_000 + id, + brain_ms: 5.0, + kind: flysim::snapshot::FeedEventKind::System, + label: "x".repeat(200), + value: None, + reward_kind: None, + by: None, + }); + } + assert!(serde_json::to_vec(&snapshot.header).unwrap().len() > 65_536); + let paths = start(snapshot.clone(), true).await; + let mut direct = connect(paths.direct, &["frame", "audio", "spikes"]).await; + let mut edge = connect(paths.edge, &["frame", "audio", "spikes"]).await; + let direct = next_binary(&mut direct, Duration::from_secs(20)).await; + let edge = next_binary(&mut edge, Duration::from_secs(20)).await; + assert!(direct == edge); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn the_edge_drops_its_clients_and_unbinds_when_the_bus_goes_away_then_comes_back() { + let snapshot = snapshot_of(&fixture_messages("shop")[5]).unwrap(); + let paths = start(snapshot.clone(), true).await; + let mut edge = connect(paths.edge, &[]).await; + next_binary(&mut edge, Duration::from_secs(20)).await; + + // flysim stopping is its router stopping. + let Paths { + snapshots, + edge: edge_addr, + edge_metrics, + bus_dir, + bus, + .. + } = paths; + bus.router.shutdown(); + drop(bus); + drop(snapshots); + let closed = tokio::time::timeout(Duration::from_secs(10), async { + use futures_util::StreamExt as _; + loop { + match edge.next().await { + None | Some(Err(_)) => break, + Some(Ok(tokio_tungstenite::tungstenite::Message::Close(_))) => break, + Some(Ok(_)) => continue, + } + } + }) + .await; + assert!(closed.is_ok(), "the client was not dropped"); + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while tokio::net::TcpStream::connect(edge_addr).await.is_ok() { + assert!( + std::time::Instant::now() < deadline, + "the feed port stayed bound" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert_eq!( + edge_metrics + .connected + .load(std::sync::atomic::Ordering::Relaxed), + 0 + ); + + // A new flysim on the same directory: the edge finds it and serves again. + let (snapshots, receiver) = tokio::sync::watch::channel(std::sync::Arc::new(snapshot.clone())); + let bus = flysim::feedbus::start_router(bus_dir.path()).await.unwrap(); + tokio::spawn(flysim::feedbus::run_publisher( + bus.router.clone(), + receiver, + std::sync::Arc::new(flysim::metrics::Metrics::default()), + )); + let mut edge = connect(edge_addr, &[]).await; + let message = next_binary(&mut edge, Duration::from_secs(20)).await; + assert_eq!(seq_of(&message), snapshot.header.seq); + assert_eq!( + edge_metrics + .bus_lost + .load(std::sync::atomic::Ordering::Relaxed), + 1 + ); + drop(snapshots); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_edge_whose_port_is_taken_keeps_retrying_and_serves_once_it_is_free() { + let snapshot = snapshot_of(&fixture_messages("center")[7]).unwrap(); + // Someone else (flysim still in direct mode, say) holds the feed port before the edge starts. + let squatter = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let port = squatter.local_addr().unwrap(); + let bus_dir = tempfile::tempdir().unwrap(); + let (snapshots, receiver) = tokio::sync::watch::channel(std::sync::Arc::new(snapshot.clone())); + let bus = flysim::feedbus::start_router(bus_dir.path()).await.unwrap(); + tokio::spawn(flysim::feedbus::run_publisher( + bus.router.clone(), + receiver, + std::sync::Arc::new(flysim::metrics::Metrics::default()), + )); + let metrics = std::sync::Arc::new(fly_edge::EdgeMetrics::default()); + tokio::spawn(fly_edge::run( + fly_edge::EdgeConfig { + bus_dir: bus_dir.path().to_path_buf(), + feed_bind: port, + idle_period: NO_IDLE, + metrics_bind: None, + retry: Duration::from_millis(50), + }, + std::sync::Arc::clone(&metrics), + )); + // It reaches the bus, fails to bind, and says so rather than claiming to wait for the bus. + let deadline = std::time::Instant::now() + Duration::from_secs(10); + while metrics + .bind_failures + .load(std::sync::atomic::Ordering::Relaxed) + < 3 + { + assert!( + std::time::Instant::now() < deadline, + "the edge never reached the bus" + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert_eq!( + metrics.connected.load(std::sync::atomic::Ordering::Relaxed), + 0 + ); + assert_eq!( + metrics.bus_lost.load(std::sync::atomic::Ordering::Relaxed), + 0 + ); + + drop(squatter); + let mut edge = connect(port, &[]).await; + let message = next_binary(&mut edge, Duration::from_secs(20)).await; + assert_eq!(seq_of(&message), snapshot.header.seq); + assert_eq!( + metrics.connected.load(std::sync::atomic::Ordering::Relaxed), + 1 + ); + drop(snapshots); + drop(bus); +} diff --git a/services/flysim/crates/fly-edge/tests/stall.rs b/services/flysim/crates/fly-edge/tests/stall.rs new file mode 100644 index 0000000..50176f0 --- /dev/null +++ b/services/flysim/crates/fly-edge/tests/stall.rs @@ -0,0 +1,354 @@ +//! A slow or absent edge never slows the loop. +//! +//! A thread stands in for the sim loop: flysim's own `Pacer` at realtime speed and 60 Hz Game +//! Boy frames, publishing full-size snapshots (a real 92,160-byte frame, a 17,407-byte spike +//! bitset for 139,255 neurons, 12,800 bytes of audio) into the watch slot every second frame, +//! with `watch::Sender::send`, exactly as `Sim::publish` does. Around it, three kinds of bad +//! consumer: +//! +//! - the edge is up but three of its WebSocket clients never read, so their sockets fill; +//! - the edge's place on the bus is held by a client that opens every subscription it may +//! (4) and never releases a delivery on any of them (the "slow edge", at its worst); +//! - nobody is subscribed at all (the "absent edge"). +//! +//! The gated tests assert the claim, and only the claim: the pacer reports no lag, no watch send +//! waits on a consumer, no publication is refused, the store stays bounded, and a healthy client +//! still reaches the newest snapshot. Those hold on a box at any load, because none of them is a +//! rate. +//! +//! How fast the loop's sleeps come back and how many snapshots a debug-build publisher gets +//! through measure the OS scheduler and the CPU left over, not the bus: a starved publisher +//! coalesces by design. Those bounds are in `the_three_scenarios_keep_their_rates`, which is +//! `#[ignore]`d; run it on a quiet box with +//! `cargo test --release -p fly-edge --test stall -- --ignored --nocapture`. + +mod common; + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +use common::*; +use flybus::{Client, ClientConfig, SubscriptionConfig}; +use flysim::feedbus; +use flysim::metrics::Metrics; +use flysim::pacing::Pacer; +use flysim::snapshot::{AttachmentKind, FeedStatus, Snapshot}; + +/// A full-size running snapshot from a real fixture frame. +fn full_snapshot() -> Snapshot { + let mut snapshot = snapshot_of(&fixture_messages("macros")[3]).unwrap(); + assert_eq!(snapshot.frame.len(), flysim::snapshot::FRAME_BYTES); + snapshot.header.status = FeedStatus::Running; + snapshot.header.attachments = AttachmentKind::ALL.to_vec(); + snapshot.spikes = Arc::new(vec![0b1010_0101; 139_255usize.div_ceil(8)]); + snapshot.audio = Arc::new(vec![7; 12_800]); + snapshot +} + +struct LoopReport { + frames: u64, + published: u64, + lag_seconds: f64, + worst_send: Duration, + worst_shortfall: f64, + p99_shortfall: f64, +} + +/// Run the stand-in loop for `seconds` on its own thread. +fn run_loop( + snapshots: tokio::sync::watch::Sender>, + template: Snapshot, + seconds: f64, +) -> LoopReport { + std::thread::spawn(move || { + let frame_ms = flysim::config::GAMEBOY_MS_PER_FRAME; + let mut pacer = Pacer::new(frame_ms, 1.0, Instant::now()); + let frames = (seconds * 1000.0 / frame_ms) as u64; + let mut worst_send = Duration::ZERO; + let mut shortfalls = Vec::with_capacity(frames as usize); + let mut seq = template.header.seq; + let mut published = 0; + for frame in 0..frames { + if frame % 2 == 0 { + let mut snapshot = template.clone(); + seq += 1; + snapshot.header.seq = seq; + snapshot.header.frame = frame; + let started = Instant::now(); + let _ = snapshots.send(Arc::new(snapshot)); + worst_send = worst_send.max(started.elapsed()); + published += 1; + } + let now = Instant::now(); + shortfalls.push(pacer.shortfall_seconds(now)); + let sleep = pacer.next_sleep(now); + if !sleep.is_zero() { + std::thread::sleep(sleep); + } + } + shortfalls.sort_by(f64::total_cmp); + LoopReport { + frames, + published, + lag_seconds: pacer.lag_seconds(), + worst_send, + worst_shortfall: *shortfalls.last().unwrap(), + p99_shortfall: shortfalls[shortfalls.len() * 99 / 100], + } + }) + .join() + .unwrap() +} + +fn print_report(report: &LoopReport, publisher: &Metrics, what: &str) { + eprintln!( + "{what}: {} frames, {} snapshots, lag {:.3} s, worst send {:?}, shortfall p99 {:.2} ms worst {:.2} ms, bus published {} failed {}", + report.frames, + report.published, + report.lag_seconds, + report.worst_send, + report.p99_shortfall * 1e3, + report.worst_shortfall * 1e3, + Metrics::get(&publisher.bus_published), + Metrics::get(&publisher.bus_publish_failures), + ); +} + +/// The claim: the loop is never held by the bus, whatever the load. +fn assert_unharmed(report: &LoopReport, publisher: &Metrics, what: &str) { + assert_eq!(report.lag_seconds, 0.0, "{what}: the pacer fell behind"); + // A watch send is a lock and a swap. A send that waited on a consumer would be a whole + // stall, seconds; 50 ms leaves room for a preempted thread on a loaded box. + assert!( + report.worst_send < Duration::from_millis(50), + "{what}: a send took {:?}", + report.worst_send + ); + // A slow or absent consumer is never a reason to refuse a latest publication. + assert_eq!( + Metrics::get(&publisher.bus_publish_failures), + 0, + "{what}: a publication was refused" + ); + // Coalescing is allowed, stopping is not. + assert!( + Metrics::get(&publisher.bus_published) >= 1, + "{what}: nothing reached the bus" + ); +} + +/// Rates: meaningful only on a quiet box (see the module comment). +fn assert_rates(report: &LoopReport, publisher: &Metrics, what: &str) { + // Sleep overshoot is absorbed by the next frame; under one frame at p99 means the loop kept + // its absolute deadlines. + assert!( + report.p99_shortfall < 0.016, + "{what}: p99 shortfall {:.2} ms", + report.p99_shortfall * 1e3 + ); + let published = Metrics::get(&publisher.bus_published); + assert!( + published * 2 >= report.published, + "{what}: only {published} of {} reached the bus", + report.published + ); +} + +const SECONDS: f64 = 6.0; + +/// What a scenario leaves for the rate checks. +struct Outcome { + report: LoopReport, + publisher: Arc, + /// Snapshots the healthy client received, where there is one. + healthy_received: Option, +} + +async fn stalled_clients() -> Outcome { + let template = full_snapshot(); + let paths = start(template.clone(), true).await; + // Three stages that said hello and then stopped reading: their sockets fill and stay full. + let mut stalled = Vec::new(); + for _ in 0..3 { + stalled.push(connect(paths.edge, &["frame", "audio", "spikes"]).await); + } + // One healthy stage, read continuously. + let mut healthy = connect(paths.edge, &["frame", "audio", "spikes"]).await; + let newest = Arc::new(AtomicU64::new(0)); + let received = Arc::new(AtomicU64::new(0)); + let reader = { + let (newest, received) = (Arc::clone(&newest), Arc::clone(&received)); + tokio::spawn(async move { + loop { + let message = next_binary(&mut healthy, Duration::from_secs(120)).await; + newest.store(seq_of(&message), Ordering::Relaxed); + received.fetch_add(1, Ordering::Relaxed); + } + }) + }; + + let snapshots = paths.snapshots.clone(); + let report = tokio::task::spawn_blocking(move || run_loop(snapshots, template, SECONDS)) + .await + .unwrap(); + print_report(&report, &paths.publisher_metrics, "stalled clients"); + assert_unharmed(&report, &paths.publisher_metrics, "stalled clients"); + + // The healthy client reaches the last snapshot published: the newest one always gets + // through, however many in between were coalesced. + let last = paths.snapshots.borrow().header.seq; + let deadline = Instant::now() + Duration::from_secs(60); + while newest.load(Ordering::Relaxed) < last { + assert!( + Instant::now() < deadline, + "healthy client stuck at {} of {last}", + newest.load(Ordering::Relaxed) + ); + tokio::time::sleep(Duration::from_millis(20)).await; + } + // The stalled ones are still connected, not dropped for being slow. + assert_eq!(paths.edge_metrics.feed.clients(), 4); + reader.abort(); + drop(stalled); + Outcome { + report, + publisher: Arc::clone(&paths.publisher_metrics), + healthy_received: Some(received.load(Ordering::Relaxed)), + } +} + +async fn hoarding_subscriber() -> Outcome { + let template = full_snapshot(); + let paths = start(template.clone(), false).await; + // The edge's seat, taken by a subscriber that keeps every delivery it gets. + let client = Client::connect_unix( + feedbus::socket_path(paths.bus_dir.path()), + ClientConfig::new(feedbus::EDGE, feedbus::store_root(paths.bus_dir.path())), + ) + .await + .unwrap(); + // Every subscription the seat may open, each keeping every delivery at the in-flight cap: + // the worst case the store has to hold (`feedbus::limits`, flybus.md "Feed sizing"). + let seats = feedbus::limits().max_subscriptions_per_client; + let mut hoards = Vec::new(); + for _ in 0..seats { + let mut subscription = client + .subscribe( + feedbus::TOPIC, + SubscriptionConfig::latest().in_flight(2).replay(true), + ) + .await + .unwrap(); + hoards.push(tokio::spawn(async move { + let mut kept = Vec::new(); + while let Some(message) = subscription.next().await { + kept.push(message); + } + kept.len() + })); + } + assert!( + client + .subscribe(feedbus::TOPIC, SubscriptionConfig::latest()) + .await + .is_err(), + "a subscription past max_subscriptions_per_client was admitted" + ); + // And the seat is the only one: a second connection as the edge is refused. + assert!( + Client::connect_unix( + feedbus::socket_path(paths.bus_dir.path()), + ClientConfig::new(feedbus::EDGE, feedbus::store_root(paths.bus_dir.path())), + ) + .await + .is_err(), + "a second client was admitted on edge.sock" + ); + + let snapshots = paths.snapshots.clone(); + let report = tokio::task::spawn_blocking(move || run_loop(snapshots, template, SECONDS)) + .await + .unwrap(); + print_report(&report, &paths.publisher_metrics, "hoarding subscriber"); + assert_unharmed(&report, &paths.publisher_metrics, "hoarding subscriber"); + let stats = paths.bus.router.stats(); + eprintln!( + "hoarding subscriber: store {} bytes, retained {} bytes", + stats.store_bytes, stats.retained_bytes + ); + // Held: per subscription two in flight and one queued, plus one retained and whatever is + // mid-seal: 4 * 3 + 1 + 2 = 15 snapshots at most. Bounded, not growing with the number + // published. + assert!( + stats.store_bytes <= 15 * 122_367, + "store holds {} bytes", + stats.store_bytes + ); + for hoard in hoards { + hoard.abort(); + } + Outcome { + report, + publisher: Arc::clone(&paths.publisher_metrics), + healthy_received: None, + } +} + +async fn absent_edge() -> Outcome { + let template = full_snapshot(); + let paths = start(template.clone(), false).await; + let snapshots = paths.snapshots.clone(); + let report = tokio::task::spawn_blocking(move || run_loop(snapshots, template, SECONDS)) + .await + .unwrap(); + print_report(&report, &paths.publisher_metrics, "absent edge"); + assert_unharmed(&report, &paths.publisher_metrics, "absent edge"); + let stats = paths.bus.router.stats(); + assert!( + stats.store_bytes <= 3 * 122_367, + "store holds {} bytes", + stats.store_bytes + ); + Outcome { + report, + publisher: Arc::clone(&paths.publisher_metrics), + healthy_received: None, + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn clients_of_the_edge_that_never_read_do_not_lag_the_loop_or_the_healthy_client() { + stalled_clients().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn a_bus_subscriber_that_never_releases_does_not_lag_the_loop_or_fill_the_store() { + hoarding_subscriber().await; +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn an_absent_edge_costs_the_loop_nothing() { + absent_edge().await; +} + +/// The same three scenarios, plus the rates. A measurement of the box as much as of the bus, +/// so not part of the workspace gate. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "timing: needs a quiet box; run with --release -- --ignored"] +async fn the_three_scenarios_keep_their_rates() { + for (what, outcome) in [ + ("stalled clients", stalled_clients().await), + ("hoarding subscriber", hoarding_subscriber().await), + ("absent edge", absent_edge().await), + ] { + assert_rates(&outcome.report, &outcome.publisher, what); + if let Some(got) = outcome.healthy_received { + assert!( + got * 2 >= outcome.report.published, + "{what}: the healthy client got only {got} of {}", + outcome.report.published + ); + } + } +} diff --git a/services/flysim/crates/flybus/README.md b/services/flysim/crates/flybus/README.md index 125391a..003fbe4 100644 --- a/services/flysim/crates/flybus/README.md +++ b/services/flysim/crates/flybus/README.md @@ -14,8 +14,9 @@ Where this crate narrows or extends the draft, the difference is listed under sections 2 to 11 is audited against this code, with the test that proves it, in `docs/design/session-framework/bus-conformance.md`. -Nothing in the crate is specific to a game, a brain or a stream. It is a workspace member and -no other crate depends on it yet. +Nothing in the crate is specific to a game, a brain or a stream. It is a workspace member; +`flysim` embeds a router for the feed (`FLY_FEED_VIA=bus`, `flysim::feedbus`) and `fly-edge` +subscribes to it (`docs/design/flybus.md`, "Feed over the bus"). ## Layout diff --git a/services/flysim/crates/flysim/Cargo.toml b/services/flysim/crates/flysim/Cargo.toml index 4d3109e..9f6c3db 100644 --- a/services/flysim/crates/flysim/Cargo.toml +++ b/services/flysim/crates/flysim/Cargo.toml @@ -30,6 +30,7 @@ cuda = ["flybrain-core/cuda"] [dependencies] flybrain-core = { path = "../flybrain-core" } flybrain-gb = { path = "../flybrain-gb" } +flybus = { path = "../flybus" } anyhow = "1.0" axum = { version = "0.8", features = ["ws"] } diff --git a/services/flysim/crates/flysim/src/config.rs b/services/flysim/crates/flysim/src/config.rs index 08d4674..3580217 100644 --- a/services/flysim/crates/flysim/src/config.rs +++ b/services/flysim/crates/flysim/src/config.rs @@ -9,7 +9,7 @@ //! - the `FLY_*` names the systemd units already set (`FLY_GAME`, `FLY_ROM`, `FLY_DATASET`, //! `FLY_STATE`, `FLY_STATE_HOT`, `FLY_FEED_BIND`, `FLY_CONTROL_BIND`, `FLY_METRICS_ADDR`, //! `FLY_ROM_SHA256`, `FLY_ROM_PLATFORMER_SHA256`, `FLY_CHAT_ENABLED`, `FLY_CHAT_DENY_LIST`, -//! `FLY_MACRO_MODE`, `RAYON_NUM_THREADS`); +//! `FLY_MACRO_MODE`, `FLY_FEED_VIA`, `FLY_BUS_DIR`, `RAYON_NUM_THREADS`); //! - `FLYSIM_
_` for everything, e.g. `FLYSIM_LOOP_SPEED=0`. //! //! Nothing here is secret (`docs/control-api.md`: "No secrets live in this service or its @@ -104,6 +104,35 @@ pub struct Feed { pub bind: SocketAddr, /// Audio attachment sample rate. The page wants Web Audio's native 48 kHz. pub audio_hz: u32, + /// Who serves `:7400/feed` (`docs/design/flybus.md`, "Feed over the bus"). + pub via: FeedVia, + /// The bus runtime directory in `bus` mode: the router's socket and its artifact store. + /// Belongs on tmpfs; a store here holds a few snapshots, never history. + pub bus_dir: PathBuf, +} + +/// Where the feed WebSocket is served from. +/// +/// `direct` is the default and is the behaviour that predates the bus, byte for byte: flysim +/// binds `feed.bind` itself. `bus` starts an embedded flybus router, publishes every snapshot +/// on it, and leaves `feed.bind` to the `fly-edge` process. The control API stays in flysim +/// either way. Nothing about the fly changes with this knob: it is outside the simulation loop +/// and outside the compatibility string. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum FeedVia { + #[default] + Direct, + Bus, +} + +impl FeedVia { + pub const fn as_str(self) -> &'static str { + match self { + Self::Direct => "direct", + Self::Bus => "bus", + } + } } #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -197,6 +226,8 @@ impl Default for Feed { Self { bind: "127.0.0.1:7400".parse().expect("literal address"), audio_hz: 48_000, + via: FeedVia::Direct, + bus_dir: PathBuf::from("/run/fly/bus"), } } } @@ -255,6 +286,12 @@ impl Config { if let Some(value) = get("FLY_FEED_BIND") { self.feed.bind = parse_addr("FLY_FEED_BIND", value)?; } + if let Some(value) = get("FLY_FEED_VIA") { + self.feed.via = parse_feed_via("FLY_FEED_VIA", value)?; + } + if let Some(value) = get("FLY_BUS_DIR") { + self.feed.bus_dir = PathBuf::from(value); + } if let Some(value) = get("FLY_CONTROL_BIND") { self.control.bind = parse_addr("FLY_CONTROL_BIND", value)?; } @@ -330,6 +367,12 @@ impl Config { if let Some(value) = get("FLYSIM_FEED_AUDIO_HZ") { self.feed.audio_hz = parse("FLYSIM_FEED_AUDIO_HZ", value)?; } + if let Some(value) = get("FLYSIM_FEED_VIA") { + self.feed.via = parse_feed_via("FLYSIM_FEED_VIA", value)?; + } + if let Some(value) = get("FLYSIM_FEED_BUS_DIR") { + self.feed.bus_dir = PathBuf::from(value); + } if let Some(value) = get("FLYSIM_CONTROL_BIND") { self.control.bind = parse_addr("FLYSIM_CONTROL_BIND", value)?; } @@ -416,6 +459,16 @@ impl Config { if self.control.sugar_per_minute == 0 { bail!("control.sugar_per_minute must be at least 1"); } + // The router's socket and store, and the edge's way to them. A relative path would + // resolve against whichever working directory each process happens to have, so the two + // could silently disagree; an empty one is a typo. Checked in either mode, so a bad + // value is found before the day a box is switched to the bus. + if self.feed.bus_dir.as_os_str().is_empty() || !self.feed.bus_dir.is_absolute() { + bail!( + "feed.bus_dir (FLY_BUS_DIR) must be an absolute path, got {:?}", + self.feed.bus_dir + ); + } if self.feed.bind == self.control.bind { bail!("feed.bind and control.bind must differ (7400 and 7401)"); } @@ -464,6 +517,14 @@ impl Config { } } +fn parse_feed_via(name: &str, value: &str) -> Result { + match value.to_ascii_lowercase().as_str() { + "direct" => Ok(FeedVia::Direct), + "bus" => Ok(FeedVia::Bus), + _ => bail!("{name}: {value:?} is not a feed path; expected \"direct\" or \"bus\""), + } +} + fn parse_addr(name: &str, value: &str) -> Result { value .parse() @@ -687,6 +748,44 @@ mod tests { assert_eq!(config, Config::default()); } + #[test] + fn the_feed_path_is_direct_unless_the_environment_says_bus() { + let config = Config::default(); + assert_eq!(config.feed.via, FeedVia::Direct); + assert_eq!(config.feed.bus_dir, PathBuf::from("/run/fly/bus")); + + let mut config = Config::default(); + config + .apply_env(&env(&[("FLY_FEED_VIA", "bus"), ("FLY_BUS_DIR", "/tmp/fly-bus")])) + .unwrap(); + assert_eq!(config.feed.via, FeedVia::Bus); + assert_eq!(config.feed.bus_dir, PathBuf::from("/tmp/fly-bus")); + + let mut config = Config::default(); + config.apply_env(&env(&[("FLYSIM_FEED_VIA", "DIRECT")])).unwrap(); + assert_eq!(config.feed.via, FeedVia::Direct); + + // A typo is a refusal, not a silent fallback to one of the two. + let error = Config::default().apply_env(&env(&[("FLY_FEED_VIA", "buss")])).unwrap_err(); + assert!(error.to_string().contains("FLY_FEED_VIA"), "{error}"); + assert_eq!(toml::from_str::("[feed]\nvia = \"bus\"\n").unwrap().feed.via, FeedVia::Bus); + } + + #[test] + fn the_bus_dir_must_be_absolute_and_not_empty() { + Config::default().validate().unwrap(); + for bad in ["", "run/fly/bus", "./bus"] { + let mut config = Config::default(); + config.feed.bus_dir = PathBuf::from(bad); + let error = config.validate().unwrap_err(); + assert!(error.to_string().contains("FLY_BUS_DIR"), "{bad:?}: {error}"); + } + // Through the environment too. + let mut config = Config::default(); + config.apply_env(&env(&[("FLY_BUS_DIR", "relative/bus")])).unwrap(); + assert!(config.validate().is_err()); + } + #[test] fn the_example_file_parses_and_validates() { let path = concat!(env!("CARGO_MANIFEST_DIR"), "/../../flysim.toml.example"); diff --git a/services/flysim/crates/flysim/src/feed.rs b/services/flysim/crates/flysim/src/feed.rs index dad5879..2e3d1c0 100644 --- a/services/flysim/crates/flysim/src/feed.rs +++ b/services/flysim/crates/flysim/src/feed.rs @@ -9,8 +9,14 @@ //! - 30 snapshots a second while running, 2 while paused or booting (header only); //! - drop-oldest, never queue: the sim publishes into a `watch` slot, so a slow client misses //! snapshots instead of slowing the loop down. Those misses are counted. +//! +//! The server only needs a [`FeedState`]: a watch slot of snapshots, the counters and the idle +//! cadence. flysim builds one from its own state when it serves the feed itself +//! (`FLY_FEED_VIA=direct`), and `fly-edge` builds one from the snapshots it takes off the bus +//! (`FLY_FEED_VIA=bus`), so both paths run this same code and write the same bytes. use std::sync::Arc; +use std::time::Duration; use axum::Router; use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; @@ -19,9 +25,22 @@ use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::routing::any; use serde::Deserialize; +use tokio::sync::watch; +use crate::metrics::Metrics; use crate::snapshot::{AttachmentKind, FeedStatus, PROTOCOL, Snapshot, Wants}; -use crate::{AppState, metrics::Metrics}; + +/// Everything the feed server reads. +#[derive(Clone)] +pub struct FeedState { + /// The newest snapshot. Dropping its sender ends every client's stream. + pub snapshots: watch::Receiver>, + /// `frames_sent`, `feed_clients` and `feed_dropped` are the ones this module moves. + pub metrics: Arc, + /// The protocol's idle cadence: how long a paused or booting stream waits before it + /// repeats the current header (`config.publish_periods().1`). + pub idle_period: Duration, +} /// The one JSON text message a client sends on connect. #[derive(Debug, Clone, Deserialize)] @@ -37,7 +56,7 @@ pub struct ClientHello { /// Close code for a protocol violation, as the reference server uses. const CLOSE_PROTOCOL_ERROR: u16 = 1002; -pub fn router(state: AppState) -> Router { +pub fn router(state: FeedState) -> Router { Router::new() .route("/feed", any(upgrade)) .fallback(not_found) @@ -48,11 +67,11 @@ async fn not_found() -> Response { (StatusCode::NOT_FOUND, "not found").into_response() } -async fn upgrade(upgrade: WebSocketUpgrade, State(state): State) -> Response { +async fn upgrade(upgrade: WebSocketUpgrade, State(state): State) -> Response { upgrade.on_upgrade(move |socket| serve_client(socket, state)) } -async fn serve_client(mut socket: WebSocket, state: AppState) { +async fn serve_client(mut socket: WebSocket, state: FeedState) { let Some(hello) = read_hello(&mut socket).await else { return; }; @@ -64,9 +83,9 @@ async fn serve_client(mut socket: WebSocket, state: AppState) { spikes = wants.spikes, "feed client connected" ); - state.shared.metrics.client_joined(); + state.metrics.client_joined(); let result = pump(&mut socket, &state, wants).await; - state.shared.metrics.client_left(); + state.metrics.client_left(); match result { Ok(()) => tracing::info!("feed client disconnected"), Err(error) => tracing::info!(%error, "feed client dropped"), @@ -117,9 +136,9 @@ async fn read_hello(socket: &mut WebSocket) -> Option { None } -async fn pump(socket: &mut WebSocket, state: &AppState, wants: Wants) -> Result<(), axum::Error> { +async fn pump(socket: &mut WebSocket, state: &FeedState, wants: Wants) -> Result<(), axum::Error> { let mut receiver = state.snapshots.clone(); - let (_, idle_period) = state.shared.config.publish_periods(); + let idle_period = state.idle_period; let mut last_seq = 0u64; // The current snapshot first, so a client that connects while paused or booting sees the @@ -161,18 +180,18 @@ async fn pump(socket: &mut WebSocket, state: &AppState, wants: Wants) -> Result< async fn send( socket: &mut WebSocket, - state: &AppState, + state: &FeedState, snapshot: &Arc, wants: Wants, last_seq: &mut u64, ) -> Result<(), axum::Error> { let seq = snapshot.header.seq; if seq > *last_seq + 1 && *last_seq != 0 { - Metrics::add(&state.shared.metrics.feed_dropped, seq - *last_seq - 1); + Metrics::add(&state.metrics.feed_dropped, seq - *last_seq - 1); } *last_seq = seq; socket.send(Message::Binary(snapshot.encode(wants).into())).await?; - Metrics::incr(&state.shared.metrics.frames_sent); + Metrics::incr(&state.metrics.frames_sent); Ok(()) } diff --git a/services/flysim/crates/flysim/src/feedbus.rs b/services/flysim/crates/flysim/src/feedbus.rs new file mode 100644 index 0000000..bf0cbf0 --- /dev/null +++ b/services/flysim/crates/flysim/src/feedbus.rs @@ -0,0 +1,354 @@ +//! The feed over flybus (`FLY_FEED_VIA=bus`, `docs/design/flybus.md` "Feed over the bus"). +//! +//! Both halves of the bus encoding live here, so the publisher in flysim and the subscriber in +//! `fly-edge` cannot drift apart: +//! +//! - [`publish`] turns one [`Snapshot`] into one publication on [`TOPIC`]: every attachment the +//! header lists as a sealed artifact named after its kind (`frame`, `audio`, `spikes`), and the +//! header itself as the envelope payload `{"header": {...}}`. A header too large for an +//! envelope travels as a `header` artifact instead, so no snapshot is ever unpublishable. +//! - [`receive`] turns that publication back into the same [`Snapshot`], which `fly-edge` hands to +//! [`crate::feed`] exactly as flysim does. The WebSocket bytes are therefore produced by the same +//! `Snapshot::encode` on both paths. +//! +//! The simulation thread never sees any of this. It publishes into its `watch` slot as it +//! always has; [`run_publisher`] is a task on the bus's own runtime that reads that slot and +//! skips whatever it was too slow to see, the same drop-oldest rule every feed client gets. +//! A stalled router, a full store or an absent edge can cost snapshots on the bus, never a +//! frame of the loop. + +use std::io::Write as _; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use flybus::{ + Artifact, BusError, Client, ClientConfig, ErrorCode, Grants, Limits, Message, Pattern, Policy, + PublishReceipt, Retained, Router, RouterConfig, UnixListenerHandle, +}; +use serde_json::{Map, Value}; +use tokio::sync::watch; + +use crate::metrics::Metrics; +use crate::snapshot::{AttachmentKind, FeedHeader, Snapshot}; + +/// The one topic: `latest` retention, so a subscriber that joins late starts from the newest +/// snapshot and one that falls behind is coalesced rather than queued. +pub const TOPIC: &str = "fly.feed.snapshots"; +/// The publisher's participant id (in-process, launcher-bound). +pub const PUBLISHER: &str = "flysim"; +/// The edge's participant id; the Unix socket is bound to it. +pub const EDGE: &str = "fly-edge"; +/// Socket file under `feed.bus_dir`, bound to [`EDGE`] only. +pub const SOCKET: &str = "edge.sock"; +/// Store root under `feed.bus_dir`; the router makes its per-incarnation directory inside it. +pub const STORE: &str = "store"; +/// Headers up to this many JSON bytes ride in the envelope; anything larger becomes an artifact. +/// Well under flybus's 65,536-byte envelope limit, leaving room for the attachment references +/// and the router's ids. A real header is 2 to 8 KB. +pub const HEADER_INLINE_MAX: usize = 48 * 1024; +/// The attachment name of an out-of-line header. +pub const HEADER_ARTIFACT: &str = "header"; + +/// `/edge.sock`. +pub fn socket_path(bus_dir: &Path) -> PathBuf { + bus_dir.join(SOCKET) +} + +/// `/store`. +pub fn store_root(bus_dir: &Path) -> PathBuf { + bus_dir.join(STORE) +} + +/// The router limits for the feed (`docs/design/flybus.md`, amendment "Feed sizing"). +/// +/// One snapshot with attachments is 122,367 bytes on the live fly: a 92,160-byte 160x144 RGBA +/// frame, a 17,407-byte spike bitset (139,255 neurons) and about 12,800 bytes of audio (1,600 +/// stereo f32 frames at 48 kHz per 30 Hz snapshot). A `latest` subscriber pins at most its one +/// queued slot plus its in-flight credits, the topic pins one retained value, and the publisher +/// holds one snapshot of staging plus the sealed copy while it seals. +/// +/// Only one client can subscribe at all: the publisher is in process, and the one socket is +/// launcher-bound to [`EDGE`], which the router admits once at a time. So the worst case is +/// that client holding every subscription it may open ([`Limits::max_subscriptions_per_client`], +/// 4), each never consuming with in-flight credits at the cap of 2: `4 * 3 + 1 + 2 = 15` +/// snapshots, about 1.8 MB. `max_clients` bounds connections, pending handshakes included, not +/// subscribers. The store cap is well over ten times that so a burst of catch-up audio after a +/// stall still fits, and it is RAM (tmpfs), so it is kept small on purpose. +pub fn limits() -> Limits { + Limits { + max_clients: 8, + max_services: 8, + max_topics: 8, + max_subscriptions_per_client: 4, + max_subscriptions: 16, + max_latest_in_flight: 2, + max_owners_per_client: 64, + reserved_owners_per_client: 8, + // Audio accumulates while the loop is behind its publish deadline; 4 MiB is ten seconds + // of it, far past anything the pacer allows before it logs lag. + max_artifact_bytes: 4 << 20, + max_store_bytes: 32 << 20, + max_retained_bytes: 8 << 20, + ..Limits::default() + } +} + +/// flysim may declare and publish the feed topic; the edge may only subscribe to it. +pub fn policy() -> Policy { + Policy::closed() + .client( + PUBLISHER, + Grants { + publish: vec![Pattern::exact(TOPIC)], + manage_topics: vec![Pattern::exact(TOPIC)], + ..Grants::default() + }, + ) + .client( + EDGE, + Grants { + subscribe: vec![Pattern::exact(TOPIC)], + ..Grants::default() + }, + ) +} + +/// A running router and the edge's socket. Dropping it stops listening; the router stops with +/// the runtime it was started on. +pub struct BusFeed { + pub router: Router, + _listener: UnixListenerHandle, +} + +/// Start the embedded router under `bus_dir` and listen for the edge on `/edge.sock`. +/// +/// Must run inside a Tokio runtime. `Router::new` removes store directories a previous flysim +/// left behind (their `flock` is free once that process is gone); a stale socket file is removed +/// here, because a socket outlives its listener on disk. +pub async fn start_router(bus_dir: &Path) -> anyhow::Result { + use anyhow::Context as _; + use std::os::unix::fs::DirBuilderExt as _; + std::fs::DirBuilder::new() + .recursive(true) + .mode(0o700) + .create(bus_dir) + .with_context(|| format!("creating the bus directory {}", bus_dir.display()))?; + let root = store_root(bus_dir); + std::fs::DirBuilder::new() + .recursive(true) + .mode(0o700) + .create(&root) + .with_context(|| format!("creating the bus store root {}", root.display()))?; + let socket = socket_path(bus_dir); + match std::fs::remove_file(&socket) { + Ok(()) => tracing::info!(socket = %socket.display(), "removed a stale bus socket"), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(error).with_context(|| format!("removing {}", socket.display())); + } + } + let mut config = RouterConfig::new(root); + config.limits = limits(); + config.policy = policy(); + let router = Router::new(config).context("starting the flybus router")?; + let listener = router + .listen_unix_as(&socket, EDGE) + .await + .with_context(|| format!("listening on {}", socket.display()))?; + tracing::info!( + socket = %socket.display(), + store = %router.store_dir().display(), + router = router.router_id(), + "feed bus listening" + ); + Ok(BusFeed { + router, + _listener: listener, + }) +} + +fn attachment_name(kind: AttachmentKind) -> &'static str { + match kind { + AttachmentKind::Frame => "frame", + AttachmentKind::Audio => "audio", + AttachmentKind::Spikes => "spikes", + } +} + +fn content_type(kind: AttachmentKind) -> &'static str { + match kind { + AttachmentKind::Frame => "image/x-rgba", + AttachmentKind::Audio => "audio/x-f32le", + AttachmentKind::Spikes => "application/x-spike-bitset", + } +} + +fn bytes_of(snapshot: &Snapshot, kind: AttachmentKind) -> &[u8] { + match kind { + AttachmentKind::Frame => &snapshot.frame, + AttachmentKind::Audio => &snapshot.audio, + AttachmentKind::Spikes => &snapshot.spikes, + } +} + +async fn seal(client: &Client, bytes: &[u8], content_type: &str) -> Result { + let mut writer = client + .artifacts() + .allocate(bytes.len() as u64, content_type) + .await?; + writer + .write_all(bytes) + .map_err(|error| BusError::new(ErrorCode::StoreFailure, format!("staging: {error}")))?; + writer.seal().await +} + +/// Publish one snapshot: its attachments as artifacts, its header as the payload. +pub async fn publish(client: &Client, snapshot: &Snapshot) -> Result { + let header = &snapshot.header; + let json = serde_json::to_vec(header).expect("a FeedHeader always serializes"); + + let mut kinds: Vec = Vec::with_capacity(3); + for kind in header.attachments.iter().copied() { + if !kinds.contains(&kind) { + kinds.push(kind); + } + } + let mut artifacts: Vec<(&'static str, Artifact)> = Vec::with_capacity(4); + for kind in kinds { + let artifact = seal(client, bytes_of(snapshot, kind), content_type(kind)).await?; + artifacts.push((attachment_name(kind), artifact)); + } + + let mut payload = Map::new(); + if json.len() <= HEADER_INLINE_MAX { + let value: Value = serde_json::from_slice(&json).expect("a serialized header re-parses"); + payload.insert("header".into(), value); + } else { + artifacts.push(( + HEADER_ARTIFACT, + seal(client, &json, "application/json").await?, + )); + } + let attachments: Vec<(&str, &Artifact)> = artifacts + .iter() + .map(|(name, artifact)| (*name, artifact)) + .collect(); + client.publish(TOPIC, payload, &attachments).await +} + +/// Rebuild the snapshot one publication carries. The message's delivery is released when the +/// caller drops it; every byte has been copied out by then. +pub async fn receive(message: &Message) -> Result { + let invalid = |what: String| BusError::new(ErrorCode::InvalidEnvelope, what); + let header: FeedHeader = match message.payload().get("header") { + Some(value) => serde_json::from_value(value.clone()) + .map_err(|error| invalid(format!("feed header: {error}")))?, + None => { + let bytes = message.artifact(HEADER_ARTIFACT)?.read_all().await?; + serde_json::from_slice(&bytes) + .map_err(|error| invalid(format!("feed header artifact: {error}")))? + } + }; + let mut snapshot = Snapshot { + header, + frame: Arc::new(Vec::new()), + audio: Arc::new(Vec::new()), + spikes: Arc::new(Vec::new()), + }; + let kinds = snapshot.header.attachments.clone(); + for kind in kinds { + let bytes = Arc::new(message.artifact(attachment_name(kind))?.read_all().await?); + match kind { + AttachmentKind::Frame => snapshot.frame = bytes, + AttachmentKind::Audio => snapshot.audio = bytes, + AttachmentKind::Spikes => snapshot.spikes = bytes, + } + } + Ok(snapshot) +} + +/// How often a failing publisher repeats its warning. +const WARN_EVERY: Duration = Duration::from_secs(10); + +/// Publish every snapshot the sim puts in its watch slot until the sim is gone. +/// +/// Connects in process as [`PUBLISHER`], declares [`TOPIC`] with `latest` retention and +/// publishes the current snapshot first, so an edge that connects at once still sees the boot +/// state. A refused publication (a full store, say) is counted and skipped; a lost connection +/// is re-made after a second. Borrows of the watch slot end before any await, exactly as in +/// [`crate::feed`]: a held borrow is a lock the sim thread's next publish would wait on. +pub async fn run_publisher( + router: Router, + mut snapshots: watch::Receiver>, + metrics: Arc, +) { + let store_root = router.store_root().to_path_buf(); + let mut last_warning: Option = None; + let mut warn = |error: &BusError, what: &str| { + if last_warning.is_none_or(|at| at.elapsed() >= WARN_EVERY) { + tracing::warn!(%error, "feed bus: {what}"); + last_warning = Some(Instant::now()); + } + }; + loop { + let transport = router.connect_in_memory_as(PUBLISHER); + let client = + match Client::connect(transport, ClientConfig::new(PUBLISHER, &store_root)).await { + Ok(client) => client, + Err(error) => { + warn(&error, "the publisher could not connect"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + }; + if let Err(error) = client.declare_topic(TOPIC, Retained::Latest).await { + warn(&error, "the feed topic could not be declared"); + tokio::time::sleep(Duration::from_secs(1)).await; + continue; + } + let mut current = snapshots.borrow_and_update().clone(); + loop { + match publish(&client, ¤t).await { + Ok(_) => Metrics::incr(&metrics.bus_published), + Err(error) => { + Metrics::incr(&metrics.bus_publish_failures); + warn(&error, "a snapshot was not published"); + if client.closed().is_some() { + break; + } + } + } + if snapshots.changed().await.is_err() { + // The sim thread is gone; so is the service. + client.close().await; + return; + } + current = snapshots.borrow_and_update().clone(); + } + tokio::time::sleep(Duration::from_secs(1)).await; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_limits_validate_and_hold_the_worst_case_with_room() { + let limits = limits(); + limits.validate().unwrap(); + // A full snapshot on the live fly (see `limits`). + let snapshot_bytes = crate::snapshot::FRAME_BYTES + 139_255usize.div_ceil(8) + 12_800; + assert_eq!(snapshot_bytes, 122_367); + // One subscribing client (the socket's), every subscription it may open, none consuming. + let subscriptions = limits.max_subscriptions_per_client as u64; + let pinned = subscriptions * (1 + limits.max_latest_in_flight) + 1 + 2; + assert_eq!(pinned, 15); + assert!( + pinned * snapshot_bytes as u64 * 10 <= limits.max_store_bytes, + "{pinned}" + ); + assert!(limits.max_artifact_bytes >= crate::snapshot::FRAME_BYTES as u64 * 40); + } +} diff --git a/services/flysim/crates/flysim/src/lib.rs b/services/flysim/crates/flysim/src/lib.rs index be481b0..fe47337 100644 --- a/services/flysim/crates/flysim/src/lib.rs +++ b/services/flysim/crates/flysim/src/lib.rs @@ -8,7 +8,8 @@ //! //! ```text //! +-- watch --> feed :7400/feed (axum + ws) -//! sim thread ---------+ +//! sim thread ---------+ \-> feedbus -> flybus -> fly-edge :7400/feed +//! | (FLY_FEED_VIA=bus instead of the line above) //! agent +-- Shared ------------> api :7401 (axum) //! emulator | /status /stimulate /reward /checkpoint //! adapter | /pause /resume /events /healthz /metrics @@ -25,6 +26,7 @@ pub mod chat; pub mod config; pub mod eventlog; pub mod feed; +pub mod feedbus; pub mod macros; pub mod metrics; pub mod pacing; @@ -41,7 +43,7 @@ use std::sync::Arc; use anyhow::{Context, Result}; use tokio::sync::{mpsc, watch}; -use crate::config::Config; +use crate::config::{Config, FeedVia}; use crate::eventlog::{EventRing, now_wall_ms}; use crate::simloop::{COMMAND_QUEUE, Command, Shared, Sim, booting_snapshot}; use crate::snapshot::Snapshot; @@ -59,6 +61,15 @@ impl AppState { pub fn snapshot(&self) -> Arc { Arc::clone(&self.snapshots.borrow()) } + + /// What the feed server needs, when flysim serves the feed itself. + pub fn feed(&self) -> feed::FeedState { + feed::FeedState { + snapshots: self.snapshots.clone(), + metrics: Arc::clone(&self.shared.metrics), + idle_period: self.shared.config.publish_periods().1, + } + } } /// Run the service until a signal or a fatal simulation error. @@ -86,10 +97,17 @@ pub fn run(config: Config) -> Result<()> { let feed_addr = config.feed.bind; let control_addr = config.control.bind; let metrics_addr = config.control.metrics_bind; + let via = config.feed.via; let listeners = runtime.block_on(async { - let feed = tokio::net::TcpListener::bind(feed_addr) - .await - .with_context(|| format!("binding the feed listener on {feed_addr}"))?; + // In bus mode the feed port belongs to `fly-edge`; binding it here would take it away. + let feed = match via { + FeedVia::Direct => Some( + tokio::net::TcpListener::bind(feed_addr) + .await + .with_context(|| format!("binding the feed listener on {feed_addr}"))?, + ), + FeedVia::Bus => None, + }; let control = tokio::net::TcpListener::bind(control_addr) .await .with_context(|| format!("binding the control listener on {control_addr}"))?; @@ -104,16 +122,42 @@ pub fn run(config: Config) -> Result<()> { Ok::<_, anyhow::Error>((feed, control, metrics)) })?; let (feed_listener, control_listener, metrics_listener) = listeners; - tracing::info!(feed = %feed_addr, control = %control_addr, metrics = ?metrics_addr, "listening"); + tracing::info!( + feed = %feed_addr, + feed_via = via.as_str(), + control = %control_addr, + metrics = ?metrics_addr, + "listening" + ); - { - let state = state.clone(); + if let Some(feed_listener) = feed_listener { + let state = state.feed(); runtime.spawn(async move { if let Err(error) = axum::serve(feed_listener, feed::router(state)).await { tracing::error!(%error, "the feed listener stopped"); } }); } + // The bus gets a runtime of its own, so neither its router nor the artifact copies can take + // a worker from the control API; and it is fed from the watch slot, never from the sim thread. + let bus_runtime = match via { + FeedVia::Direct => None, + FeedVia::Bus => { + let bus_runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_name("flysim-bus") + .enable_all() + .build() + .context("building the bus runtime")?; + let bus = bus_runtime.block_on(feedbus::start_router(&config.feed.bus_dir))?; + bus_runtime.spawn(feedbus::run_publisher( + bus.router.clone(), + state.snapshots.clone(), + Arc::clone(&state.shared.metrics), + )); + Some((bus_runtime, bus)) + } + }; { let state = state.clone(); runtime.spawn(async move { @@ -139,6 +183,13 @@ pub fn run(config: Config) -> Result<()> { let result = sim.run(¬ifier); notifier.notify("STOPPING=1\n"); drop(sim); + if let Some((bus_runtime, bus)) = bus_runtime { + // The publisher ends by itself once the watch sender is gone; stopping the runtime under + // it, rather than the router first, keeps a last in-flight publish from being logged as + // a refusal. The edge sees the socket close either way. + drop(bus); + bus_runtime.shutdown_timeout(std::time::Duration::from_secs(1)); + } runtime.shutdown_timeout(std::time::Duration::from_secs(2)); result } diff --git a/services/flysim/crates/flysim/src/metrics.rs b/services/flysim/crates/flysim/src/metrics.rs index 994e864..49d04a2 100644 --- a/services/flysim/crates/flysim/src/metrics.rs +++ b/services/flysim/crates/flysim/src/metrics.rs @@ -43,6 +43,10 @@ pub struct Metrics { pub lag_ms: AtomicU64, /// 1 when the restore fell back past the newest candidate. pub restore_fallback: AtomicU64, + /// Snapshots published on the feed bus (`FLY_FEED_VIA=bus`); 0 in direct mode. + pub bus_published: AtomicU64, + /// Snapshots the feed bus refused or could not take; each one is skipped, never retried. + pub bus_publish_failures: AtomicU64, } impl Metrics { @@ -81,7 +85,7 @@ impl Metrics { } /// One metric line plus its help and type headers. -fn metric(out: &mut String, name: &str, kind: &str, help: &str, value: impl std::fmt::Display) { +pub fn metric(out: &mut String, name: &str, kind: &str, help: &str, value: impl std::fmt::Display) { use std::fmt::Write as _; let _ = writeln!(out, "# HELP {name} {help}"); let _ = writeln!(out, "# TYPE {name} {kind}"); @@ -114,6 +118,20 @@ pub fn render(metrics: &Metrics, snapshot: &Snapshot, now_wall_ms: u64) -> Strin "Snapshots superseded before a slow client could be sent them.", Metrics::get(&metrics.feed_dropped), ); + metric( + &mut out, + "fly_bus_published_total", + "counter", + "Snapshots published on the feed bus (FLY_FEED_VIA=bus).", + Metrics::get(&metrics.bus_published), + ); + metric( + &mut out, + "fly_bus_publish_failures_total", + "counter", + "Snapshots the feed bus did not take; skipped, like any superseded snapshot.", + Metrics::get(&metrics.bus_publish_failures), + ); metric( &mut out, "fly_snapshots_published_total", diff --git a/services/flysim/crates/flysim/src/simloop.rs b/services/flysim/crates/flysim/src/simloop.rs index f5ecd55..0b24bcb 100644 --- a/services/flysim/crates/flysim/src/simloop.rs +++ b/services/flysim/crates/flysim/src/simloop.rs @@ -149,7 +149,7 @@ pub struct DecoderChannelStatus { #[derive(Debug)] pub struct Shared { pub config: Config, - pub metrics: Metrics, + pub metrics: Arc, pub events: EventRing, /// `Date.now()` at the top of the most recent loop iteration. `GET /healthz` is 200 while /// this is less than two seconds old, which is true while paused as well: a paused loop is @@ -173,7 +173,7 @@ impl Shared { pub fn new(config: Config, events: EventRing) -> Self { Self { config, - metrics: Metrics::default(), + metrics: Arc::default(), events, heartbeat_ms: AtomicU64::new(0), versions: OnceLock::new(), diff --git a/services/flysim/flysim.toml.example b/services/flysim/flysim.toml.example index ed3ca17..eb98f50 100644 --- a/services/flysim/flysim.toml.example +++ b/services/flysim/flysim.toml.example @@ -62,6 +62,13 @@ bind = "127.0.0.1:7400" # Audio attachment rate. 48 kHz is Web Audio's native rate on Linux, so the page never resamples. # env: FLYSIM_FEED_AUDIO_HZ audio_hz = 48000 +# Who serves `bind`: "direct" (flysim, the default) or "bus" (flysim publishes on an embedded +# flybus router and the `fly-edge` process serves the same bytes; docs/design/flybus.md). +# env: FLY_FEED_VIA, FLYSIM_FEED_VIA +via = "direct" +# The bus router's socket and artifact store in "bus" mode. tmpfs. +# env: FLY_BUS_DIR, FLYSIM_FEED_BUS_DIR +bus_dir = "/run/fly/bus" [control] # http://127.0.0.1:7401 — docs/control-api.md. Loopback only; there is no auth because nothing