alex/flybrain
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge feat/fly-edge-feed: EDGE-01, the feed over flybus behind FLY_FEED_VIA, served by fly-edge

Browse source
This commit is contained in:
acamilo 2026-09-23 13:13:18 +00:00
commit 0ca5f000d6
29 changed files with 2373 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
docs/architecture-tour.md
View file

@ -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

136
docs/design/flybus.md
View file

@ -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<Snapshot>--> publisher task --flybus (in memory)--> Router
(unchanged) (flysim-bus runtime) | <bus_dir>/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 `<bus_dir>/store`, closed policy: `flysim` may declare
and publish `fly.feed.snapshots`, `fly-edge` may only subscribe to it, and the Unix socket
`<bus_dir>/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
`<bus_dir>/store`, the socket `<bus_dir>/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`.

20
infra/05-deploy.sh
View file

@ -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" ;;

27
infra/bin/fly-watchdog
View file

@ -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

12
infra/build/build-flysim.sh
View file

@ -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 <stage-dir> <bridge-dir> <out-dir>"

10
infra/build/package-release.sh
View file

@ -11,6 +11,9 @@
#
# Output: OUT_DIR/flybrain-<version>.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"

3
infra/config/fly-tmpfiles.conf
View file

@ -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 -

7
infra/env/example.env vendored
View file

@ -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.

14
infra/lib/common.sh
View file

@ -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

134
infra/tests/lint.sh
View file

@ -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

56
infra/units/flyedge.service Normal file
View file

@ -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

4
infra/units/flysim.service
View file

@ -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

20
services/flysim/Cargo.lock generated
View file

@ -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",

1
services/flysim/Cargo.toml
View file

@ -1,6 +1,7 @@
[workspace]
resolver = "3"
members = [
"crates/fly-edge",
"crates/fly-session",
"crates/fly-session-types",
"crates/flybrain-core",

36
services/flysim/crates/fly-edge/Cargo.toml Normal file
View file

@ -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"

344
services/flysim/crates/fly-edge/src/lib.rs Normal file
View file

@ -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.bind>/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<SocketAddr>,
/// Delay between attempts to reach the bus.
pub retry: Duration,
}
impl EdgeConfig {
pub fn from_flysim(config: &flysim::config::Config, metrics_bind: Option<SocketAddr>) -> 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<Metrics>,
/// 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<EdgeMetrics>) -> 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<Arc<EdgeMetrics>>) -> impl IntoResponse {
(
[(
axum::http::header::CONTENT_TYPE,
"text/plain; version=0.0.4",
)],
metrics.render(),
)
}
async fn healthz(State(metrics): State<Arc<EdgeMetrics>>) -> 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");
}
}

81
services/flysim/crates/fly-edge/src/main.rs Normal file
View file

@ -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<std::path::PathBuf>,
}
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();
}

246
services/flysim/crates/fly-edge/tests/common/mod.rs Normal file
View file

@ -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<Snapshot> {
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/<name>.flyfeed.gz`, as wire messages.
pub fn fixture_messages(name: &str) -> Vec<Vec<u8>> {
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<u8>]) -> Vec<u8> {
let wall = |message: &Vec<u8>| -> 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<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;
/// 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<u8> {
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<Arc<Snapshot>>,
pub direct: SocketAddr,
pub edge: SocketAddr,
pub publisher_metrics: Arc<Metrics>,
pub edge_metrics: Arc<EdgeMetrics>,
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<PathBuf> {
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();
}

301
services/flysim/crates/fly-edge/tests/parity.rs Normal file
View file

@ -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=<dir>` 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);
}

354
services/flysim/crates/fly-edge/tests/stall.rs Normal file
View file

@ -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<Arc<Snapshot>>,
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<Metrics>,
/// Snapshots the healthy client received, where there is one.
healthy_received: Option<u64>,
}
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
);
}
}
}

5
services/flysim/crates/flybus/README.md
View file

@ -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

1
services/flysim/crates/flysim/Cargo.toml
View file

@ -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"] }

101
services/flysim/crates/flysim/src/config.rs
View file

@ -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_<SECTION>_<KEY>` 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<FeedVia> {
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<SocketAddr> {
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::<Config>("[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");

41
services/flysim/crates/flysim/src/feed.rs
View file

@ -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<Arc<Snapshot>>,
/// `frames_sent`, `feed_clients` and `feed_dropped` are the ones this module moves.
pub metrics: Arc<Metrics>,
/// 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<AppState>) -> Response {
async fn upgrade(upgrade: WebSocketUpgrade, State(state): State<FeedState>) -> 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<ClientHello> {
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<Snapshot>,
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(())
}

354
services/flysim/crates/flysim/src/feedbus.rs Normal file
View file

@ -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";
/// `<bus_dir>/edge.sock`.
pub fn socket_path(bus_dir: &Path) -> PathBuf {
bus_dir.join(SOCKET)
}
/// `<bus_dir>/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 `<bus_dir>/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<BusFeed> {
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<Artifact, BusError> {
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<PublishReceipt, BusError> {
let header = &snapshot.header;
let json = serde_json::to_vec(header).expect("a FeedHeader always serializes");
let mut kinds: Vec<AttachmentKind> = 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<Snapshot, BusError> {
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<Arc<Snapshot>>,
metrics: Arc<Metrics>,
) {
let store_root = router.store_root().to_path_buf();
let mut last_warning: Option<Instant> = 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, &current).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);
}
}

67
services/flysim/crates/flysim/src/lib.rs
View file

@ -8,7 +8,8 @@
//!
//! ```text
//! +-- watch<Snapshot> --> 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<Snapshot> {
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(&notifier);
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
}

20
services/flysim/crates/flysim/src/metrics.rs
View file

@ -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",

4
services/flysim/crates/flysim/src/simloop.rs
View file

@ -149,7 +149,7 @@ pub struct DecoderChannelStatus {
#[derive(Debug)]
pub struct Shared {
pub config: Config,
pub metrics: Metrics,
pub metrics: Arc<Metrics>,
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(),

7
services/flysim/flysim.toml.example
View file

@ -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