Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e043810eb0 | ||
|
|
5b80a1a010 | ||
|
|
b9803f93fb | ||
|
|
3df2378831 | ||
|
|
81c230a09c | ||
|
|
be3740f5f9 | ||
|
|
0d836d14c2 | ||
|
|
76c1a13e11 | ||
|
|
aa0515af1c |
@@ -891,6 +891,206 @@ session that owns the AEC's lifetime. **Trigger to revisit: a fourth tracked chi
|
||||
routine, or a measured teardown exceeds 5 s.** The fix, when triggered, is to drain viewers
|
||||
concurrently while still owned by `shutdown_children` — not to detach them.
|
||||
|
||||
**Round 17 (2026-07-26 night) — two reviews: the repair planner (changes-requested, all applied)
|
||||
and the 0c actor design (four blocking issues, all accepted).** 0c is now sliced, because the
|
||||
fault-handling surface — not the design — is what grew.
|
||||
|
||||
*The repair planner: no P1s, four reachable P2s and a P3, all applied in `9145b2a`.*
|
||||
|
||||
- **Only the canonical forms are ours.** `classify` recognised any loopback with one
|
||||
pixelpass-looking endpoint, so a third party's `module-loopback source=some_mic
|
||||
sink=pixelpass_capture_4242` was ours to unload once that pid died; and a `sink=` token nested
|
||||
inside a quoted `sink_input_properties` value could be read as a top-level argument. The whole
|
||||
recorded argument string must now equal what pixelpass itself writes.
|
||||
- **The matcher's templates are generated from the loader's own renderers.** Hard-coding
|
||||
`latency_msec=20` beside a matcher means a loader change silently blinds repair to every module
|
||||
the new build loads — the fail-closed-and-silent class this project has now been bitten by
|
||||
three times (F2, F13-1, the sticky-uncertainty inversion). `host/audio.rs` loads through the
|
||||
same renderers, so drift is a compile-time question. Blindness is also *reported*:
|
||||
`unrecognised_pixelpass_modules` names anything matching `pixelpass_capture_*` that no
|
||||
canonical form recognises, so a newer pixelpass's shapes cannot make an older `--repair`
|
||||
quietly clean up nothing.
|
||||
- **Ordering is not a licence either.** Planning loopbacks before the sink is necessary and
|
||||
insufficient: an unload can fail or be skipped, and a loopback can appear after planning. The
|
||||
sink unload is now gated on `sink_still_referenced` against the fresh snapshot — any other
|
||||
module naming that sink blocks it, ours or not, because the question is what would break rather
|
||||
than who owns it.
|
||||
- **Undecidable is not dead.** `Path::exists()` maps a permission error, a missing `/proc` and a
|
||||
foreign pid namespace all to `false`, which read here as "dead, unload it". Liveness is now
|
||||
`Alive | Dead | Unknown` via `try_exists()` behind a `/proc/self/stat` preflight, `Unknown`
|
||||
behaves exactly like `Alive`, and it is reported separately so holding back is visible.
|
||||
- **⚠️ One prescribed fix was not implementable as written, and measuring first is what caught
|
||||
it.** The reviewer's fix for fingerprint fidelity was "use `pactl -f json list modules` and
|
||||
deserialize the complete `argument`". **On pactl 17.0 those records carry no module index at
|
||||
all** (`"index": null`), and `unload-module` accepts only an index — JSON alone cannot drive
|
||||
repair. Replacement: **two listings, correlated positionally and checked** (ids and names from
|
||||
the short listing, exact arguments from JSON; equal counts and equal names at every position or
|
||||
the run refuses, with retries for a concurrent load). Verified on this host: both listings
|
||||
return the same 17 modules in an identical name sequence, from 41 physical lines. The check
|
||||
also turns the reviewer's fabricated-row attack from exploitable into harmless — a crafted
|
||||
short-listing line has no JSON counterpart, so the sequences misalign and repair stops instead
|
||||
of unloading an index inferred from text. *This is the reachability rule applied to a
|
||||
prescription rather than a finding: the chain was valid, the API it assumed did not exist.*
|
||||
- **Measured before relying on it (pactl 17.0, live server):** recorded arguments come back
|
||||
byte-for-byte as passed, joined with single spaces, in order, and **`@DEFAULT_SINK@` is not
|
||||
resolved** to the concrete device. Both facts are load-bearing for exact matching — had either
|
||||
been false, the P2 fix would itself have been a silent blinding — so both carry a test.
|
||||
- Normalisation deleted (P3): within one invocation every snapshot comes from one server, so
|
||||
re-rendering does not happen and normalising only made different arguments compare equal. The
|
||||
residual ABA window (planned module vanishes, a byte-identical one takes its index) cannot be
|
||||
closed through an index-only unload API, and is now stated as a limitation in `Fingerprint`'s
|
||||
own doc comment instead of implied away.
|
||||
- Five vacuity gaps closed: a raw-pactl-text-to-plan test (the whole planner suite survived a
|
||||
parser that dropped every argument), per-pid liveness counters over two pids, non-canonical and
|
||||
nested-quote cases, and a reference-gate test. **One gap deliberately left open and declared:**
|
||||
a comparator using only `id + args` cannot be killed by a non-vacuous test, because the module
|
||||
*name* determines which argument grammar can match at all — that field is enforced structurally
|
||||
by `classify`, and a test appearing to cover it would be the self-satisfying kind.
|
||||
- **Field-verified twice on the live graph:** the A/B orphan test still removes exactly the two
|
||||
orphans with the module table otherwise byte-identical, and a new fixture — a dead pid's legacy
|
||||
sink plus a *non-canonical* loopback naming it — unloads nothing, reports the unrecognised
|
||||
module, and reports the sink as still referenced.
|
||||
|
||||
*The 0c actor design: four blocking issues, all accepted; the epoch requirement conceded.*
|
||||
|
||||
- **A bounded join must not move the OS handle into `spawn_blocking`.** My ladder would have
|
||||
taken the thread handle out of the guard to poll it; if the close future is then cancelled or
|
||||
unwinds, `Drop` finds no handle and can neither poison nor fail-stop, while the blocking task
|
||||
stays wedged forever and can pin runtime shutdown. This is the **same defect shape as round
|
||||
15's** — a defence disarmed exactly when needed. The handle stays owned across every await;
|
||||
`is_finished()` is polled and `join()` called only once it reports finished. Same rule for the
|
||||
event task's handle (`await` through `&mut JoinHandle`).
|
||||
- **`Commit::UnloadNow(id)` cannot forget the id.** An immediate unload can time out or be
|
||||
cancelled, and a ledger that never recorded the module cannot retry or reconcile it. Slots
|
||||
become a state machine — `Vacant | Loading { token, expected } | Loaded { fp } | Unloading
|
||||
{ fp }` — with **affine** permits carrying a unique token, so two permitted loads for one slot
|
||||
cannot both commit.
|
||||
- **`kill_on_drop` does not roll back a server-side mutation.** A bounded `pactl load-module`
|
||||
killed after the server created the module but before its id was read leaves a module with no
|
||||
id anywhere. So an ambiguous load requires **bounded reconciliation by fingerprint** — reusing
|
||||
repair's classification idea inside the live session, never its dead-pid policy — before any
|
||||
further capture may start. Related: cancellation must never be `select!`ed against
|
||||
`Command::output()`, or a completed load's id is dropped on the floor.
|
||||
- **`_exit` is right, but the pre-exit sequence must not be able to block.** Event emission,
|
||||
stdio flushing and tracing all take locks a wedged thread may hold, so the watchdog able to
|
||||
`_exit` past a stalled diagnostic has to be **armed before** the wedge is detected, not created
|
||||
in response to it. And `_exit` skips `CaptureHandle::Drop`, so `gst-launch-1.0` and any
|
||||
in-flight `pactl` need parent-death/process-group containment or they outlive the host that
|
||||
reported its own death — with gst still holding screen-capture resources.
|
||||
- **Epoch conceded, and my vacuity instinct was right.** `object.serial` is unique and never
|
||||
reused while global ids are, so "the object at this id still has the serial I recorded" is
|
||||
complete proof of identity; there is no same-core interleaving that serial equality misses.
|
||||
Epoch is carried for diagnostics and explicitly **not** a gate. It would only become
|
||||
load-bearing across a daemon incarnation or an actor reconnect, and the design makes core
|
||||
failure terminal with no reconnect — if that changes, the right answer is a core-incarnation
|
||||
nonce, not a "something churned" counter that invalidates observations on unrelated traffic.
|
||||
- **"Unjoinability, not slowness" is not literally implementable** and the wording is corrected:
|
||||
no bounded observation distinguishes "returns one millisecond later" from "never returns", so
|
||||
the death condition is *failure to terminate within the post-cancellation policy deadline*.
|
||||
Two budgets, not one — a running MainLoop quitting is a different question from an
|
||||
initialisation call returning after cancellation, and the second is normally longer.
|
||||
- **`GraphCmd::Route(Vec<u32>)` is deleted rather than fixed.** Matching and routing stay inside
|
||||
the actor's registry callback, where removals are already ordered against routes in-thread, so
|
||||
the privacy race is not introduced at all. For phase 6 the rule is structural: the only
|
||||
addressable type is an `ObservedNode { global_id, serial, epoch }` constructible solely from
|
||||
the actor's own observation, kept private and non-`Copy`, revalidated on serial immediately
|
||||
before any mutation. A bare id is not addressable.
|
||||
- **An unacked `ClearRoutes` is not a wedge** (agreed), with one qualification taken: a stream
|
||||
setting `node.dont-reconnect`/`node.dont-fallback` may be left silent rather than moved back to
|
||||
the default, so the outcome is surfaced as `ClearRoutesUnconfirmed` rather than treated as
|
||||
benign. Separately, blindly clearing `target.object` can erase a target the user set manually —
|
||||
the prior value must be recorded and restored only while it is still pixelpass-owned.
|
||||
- **One terminal fault needs a coordinator, not an emitter.** If the actor emits `CoreError`
|
||||
immediately and the subsequent teardown then fails to join, peerspeak never learns the process
|
||||
is fail-stopping. Actor faults are internal *candidates*; the tokio-side coordinator emits
|
||||
exactly one final fault, and `Wedged` overrides any earlier candidate. Because a callback panic
|
||||
can cross `extern "C"` and abort before any event is produced, **peerspeak must treat
|
||||
unexpected stdout EOF as a synthetic terminal fault** rather than trusting that a JSON line
|
||||
arrives.
|
||||
|
||||
*Measured for the actor argument (3 of 3 trials, live graph):* pipewire-pulse accepts **two sinks
|
||||
with an identical `node.name`** — no rename, no suffix, no refusal, both visible as `<name>` and
|
||||
`<name>.monitor` — and `pulsesrc device=<name>.monitor` attached to the **older** one every time.
|
||||
So a surviving wedged owner does not merely risk a collision: it **silently steals the next
|
||||
session's capture** while the loopbacks feed the new sink. That retires "detach and carry on" as
|
||||
an option, and it is the evidence behind rejecting session-unique sink names (which would trade a
|
||||
fail-stop ownership fault for silent accumulation, and re-open the discovery grammar 0c step 1
|
||||
just closed and field-proved).
|
||||
|
||||
**0c step 2 is therefore sliced, and the slices land and are reviewed independently.** Nothing
|
||||
here reopens D6 — the connection-owned-sink design is unchanged; what grew is the process-
|
||||
lifecycle and fault surface, and a material part of it is pre-existing debt 0c forced into the
|
||||
light (the `abort()` orphan race, the unbounded join, peerspeak advertising a dead share):
|
||||
|
||||
| slice | scope | why it can land alone |
|
||||
|-------|-------|-----------------------|
|
||||
| S1 | repair planner (`919d5bd` + `9145b2a`) | done; awaiting re-review, then merge |
|
||||
| S2 | peerspeak host-fault path: always-on notice channel, EOF synthesis, session-scoped fault, clear `is_sharing` + presence ticket, `ScreenShareStopped` then error | fixes a defect **today** — a dead share stays advertised — and is independent of the actor |
|
||||
| S3 | pixelpass ledger transactions + ambiguous-load reconciliation + child containment + pre-armed watchdog + poison state machine + supervisor health arm | fixes the `abort()` orphan race **today**; no libpipewire work |
|
||||
| S4 | the `AudioGraphOwner` actor itself, the readiness handshake, and both measured budgets | the only slice that needs new PipeWire mechanism |
|
||||
| S5 | the two live exit gates: two-host ownership, and the never-yet-run Stop Share SIGINT gate | needs S4 on the graph |
|
||||
|
||||
**Round 18 (2026-07-26 night) — two more repair review rounds. `--repair` now reads and unloads
|
||||
through libpulse, and one of the review's own prescriptions had to be replaced after measuring.**
|
||||
|
||||
*Round 17c — the re-review of my round-17a fixes found two more blocking P2s. Two of the four
|
||||
fixes I had applied were themselves defective; this is the third time the "audit your own fixes"
|
||||
rule has paid.*
|
||||
|
||||
- **My two-listing correlation was unsound.** Pairing short-listing indices with JSON arguments by
|
||||
position breaks whenever module names repeat: another client loading one module and unloading
|
||||
another *between the two calls* leaves counts and names aligned while every argument has shifted
|
||||
by one, so a foreign module inherits a canonical fingerprint. The name check cannot see it and
|
||||
the retry never fires, because correlation "succeeded".
|
||||
- **My liveness fix still converted invisible-but-alive into dead.** A `/proc/self` preflight
|
||||
proves nothing: inside a pid namespace — a container, a distrobox — `self` stays visible while
|
||||
every process in the parent namespace is invisible, and `hidepid` has the same shape.
|
||||
|
||||
*Round 18 (round 4) — the fix for both, and a third defect neither of us had reached.*
|
||||
|
||||
- **Record boundaries in `pactl list short modules` are unprovable, and this needs no adversary.**
|
||||
A genuine module whose argument contains a newline renders a first line that is byte-exactly one
|
||||
of our canonical forms, with the rest dropped as an unparseable continuation — no forged index,
|
||||
so no duplicate-index check can see it. **Field-confirmed on the live server** with
|
||||
`…latency_msec=20\nremix=false`, `remix` being a real loopback option. A tab in the same position
|
||||
is worse: it hides a sink reference from the gate that protects a still-referenced sink.
|
||||
- **Locality was a guess.** `PULSE_SERVER` is a fallback *list*, so `unix:/missing tcp:remote:4713`
|
||||
passes any "starts with unix:" test and then connects to another machine, where local pids mean
|
||||
nothing and a live remote host's modules look dead.
|
||||
- **Resolution: `src/repair/introspect.rs`, one verified-local connection.** `pa_module_info`
|
||||
carries index, name and exact argument in a single record; `pa_context_is_local()` answers
|
||||
locality about the connection actually established; and unloading goes back through that same
|
||||
connection, so listing and destruction cannot disagree about which server they mean. Bounded
|
||||
throughout (3 s connect, 3 s per request, non-blocking iteration plus a 2 ms sleep). The layer
|
||||
holds no policy but "refuse the wrong server" — every decision stays in the pure planner.
|
||||
- **The dependency was the user's call, taken with sign-off after vetting.** libpulse-binding
|
||||
2.30.1: MIT/Apache-2.0, 5.5M downloads, 3 new crates total, a build script that only probes
|
||||
pkg-config, no network or subprocess use in any source, and all three historical RustSec
|
||||
advisories (2018-0020/0021, 2019-0038) fixed by 2.6.0. Reasoning recorded beside the dep.
|
||||
- ⚠️ **REUSABLE — a prescription can fail reachability, not just a finding.** The reviewer's
|
||||
fidelity fix was "use `pactl -f json list modules`". On pactl 17 those records carry **no module
|
||||
index at all** (`"index": null`) while `unload-module` accepts only an index, so it can never
|
||||
stand alone. Measuring first is what caught it.
|
||||
- ⚠️ **REUSABLE — the field test found a bug no unit test could reach, and it was 0b's bug again.**
|
||||
The first introspection version did its work correctly and then aborted on the way out:
|
||||
`Assertion '!e->dead' failed at mainloop.c:207, function mainloop_io_free()` — SIGABRT, core
|
||||
dumped, **exit 134, so a fully successful repair reported failure to its caller**. Rust drops
|
||||
fields in declaration order and the context's teardown frees IO events living in the mainloop,
|
||||
which I had declared first. Fixed, then hardened past the fix: `Drop` explicitly takes and
|
||||
destroys the context before the mainloop, so the ordering no longer depends on where the fields
|
||||
are written. **Field-order drop hazards are not a peerspeak-specific lesson; they recur wherever
|
||||
one object's teardown reaches into another's.**
|
||||
- **Still open, deliberately, and recorded rather than guessed:** closing the namespace hole needs
|
||||
modules to carry an **owner token** (machine/boot identity plus pid-namespace identity), with
|
||||
token-less modules treated as `Unknown`. That changes what pixelpass writes into the graph *and
|
||||
how far back `--repair` can clean up* — orphans from any older build would become uncleanable,
|
||||
which is a regression in the tool's entire purpose. `NSpid > 1` remains a sound negative signal;
|
||||
`NSpid == 1` is explicitly **not** proof, since its leftmost value is relative to the procfs that
|
||||
was mounted.
|
||||
- **Deferred, now cheap to reconsider:** `host/audio.rs` still loads modules via `pactl` and parses
|
||||
the index off stdout, which is part of why S3's ambiguous-load problem exists. With libpulse in
|
||||
the tree, `pa_context_load_module` returns the index through an observable operation.
|
||||
|
||||
**Round 1 — 13 items, 12 accepted.** Phase reorder (AEC machine before dry-run); typed capture
|
||||
plan (accepted, moved *earlier* than proposed); Phase 3 five-part gate; tag-consumption gating;
|
||||
Phase 6 matrix mandatory; 0b unwind backstop restored **and my mutation test corrected — it
|
||||
|
||||
@@ -137,6 +137,16 @@ pub enum CoreCommand {
|
||||
/// Stop sharing our screen: kill the pixelpass host and clear the presence
|
||||
/// ticket. No-op when not sharing.
|
||||
StopScreenShare,
|
||||
/// **Core-internal.** The running pixelpass host's stdout ended — the
|
||||
/// process died (or its event stream broke), so the share identified by
|
||||
/// `generation` is over: reap the child, pull the ticket off presence, and
|
||||
/// tell the user. Synthesized by the core's own notice-forwarder task; the
|
||||
/// UI never sends it. `generation` scopes the fault to one specific host
|
||||
/// spawn, so a stale fault (the user already stopped, or started a new
|
||||
/// share) is ignored rather than tearing down the wrong share.
|
||||
ScreenShareHostFault {
|
||||
generation: u64,
|
||||
},
|
||||
/// Watch a peer's screen share: spawn a pixelpass viewer for `ticket` and
|
||||
/// open it in a local player.
|
||||
ViewShare {
|
||||
@@ -271,6 +281,7 @@ pub fn delivery_class(cmd: &CoreCommand) -> DeliveryClass {
|
||||
quality: _,
|
||||
}
|
||||
| CoreCommand::StopScreenShare
|
||||
| CoreCommand::ScreenShareHostFault { generation: _ }
|
||||
| CoreCommand::ViewShare {
|
||||
ticket: _,
|
||||
settings: _,
|
||||
@@ -363,6 +374,7 @@ pub fn coalesce_key(cmd: &CoreCommand) -> Option<CoalesceKey> {
|
||||
quality: _,
|
||||
}
|
||||
| CoreCommand::StopScreenShare
|
||||
| CoreCommand::ScreenShareHostFault { generation: _ }
|
||||
| CoreCommand::ViewShare {
|
||||
ticket: _,
|
||||
settings: _,
|
||||
|
||||
+144
-36
@@ -1397,10 +1397,25 @@ async fn run_core_loop(
|
||||
// later opt-in can immediately publish whatever is currently running.
|
||||
let mut current_game: Option<crate::game::DetectedGame> = None;
|
||||
let mut network_mode = NetworkMode::default();
|
||||
// Pixelpass binary override (config), and the ticket of our own active screen
|
||||
// share (rides our presence so the room — incl. late joiners — can watch).
|
||||
// Pixelpass binary override (config), and our own active screen share: the
|
||||
// ticket rides our presence so the room — incl. late joiners — can watch,
|
||||
// and the generation ties host-fault notices to this specific host spawn
|
||||
// (see `ScreenShareHostFault`). One variable on purpose: the ticket and the
|
||||
// generation must appear and vanish together, or a stale fault could tear
|
||||
// down a share it doesn't belong to.
|
||||
let mut pixelpass_override: Option<String> = None;
|
||||
let mut current_sharing: Option<String> = None;
|
||||
struct ActiveShare {
|
||||
generation: u64,
|
||||
ticket: String,
|
||||
}
|
||||
let mut current_sharing: Option<ActiveShare> = None;
|
||||
// Monotonic per-spawn counter feeding `ActiveShare::generation`.
|
||||
let mut share_generations: u64 = 0;
|
||||
// Host faults re-enter the loop here (the notice-forwarder task can't touch
|
||||
// loop state). The loop keeps `host_fault_tx` to clone into each share's
|
||||
// forwarder, so this channel never closes — the select arm's `Some` pattern
|
||||
// is total in practice and a closed-channel branch would be unreachable.
|
||||
let (host_fault_tx, mut host_fault_rx) = mpsc::unbounded_channel::<u64>();
|
||||
|
||||
let mut active_session: Option<ActiveSession> = None;
|
||||
// Standalone capture-only mic meter, live only when no session exists.
|
||||
@@ -1549,6 +1564,13 @@ async fn run_core_loop(
|
||||
// reachable it is already covered — nothing to add here.
|
||||
None => break,
|
||||
},
|
||||
// A share's notice-forwarder task reported the host's stdout ended.
|
||||
// The `Some` pattern is total: this loop owns `host_fault_tx` (see
|
||||
// its declaration), so the channel cannot close — no `None` arm is
|
||||
// written because one would be unreachable by construction.
|
||||
Some(generation) = host_fault_rx.recv() => {
|
||||
CoreCommand::ScreenShareHostFault { generation }
|
||||
}
|
||||
game_change = next_game_change(&mut game_rx) => {
|
||||
// The detector worker published a new debounced game (or `None`).
|
||||
let Some(detected) = game_change else {
|
||||
@@ -1567,7 +1589,7 @@ async fn run_core_loop(
|
||||
let self_state = presence.to_state(
|
||||
is_muted.load(Ordering::Relaxed),
|
||||
net.endpoint.addr(),
|
||||
current_sharing.clone(),
|
||||
current_sharing.as_ref().map(|s| s.ticket.clone()),
|
||||
);
|
||||
let _ = session.room_state.update_self_state(self_state).await;
|
||||
}
|
||||
@@ -1711,6 +1733,13 @@ async fn run_core_loop(
|
||||
net.file_router.clear();
|
||||
*current_room.lock().unwrap() = None;
|
||||
}
|
||||
// Any advertised share died with that session — deliberately —
|
||||
// so retire it HERE, before the invalid-ticket early exit below
|
||||
// can skip it. Left populated, the killed host's stdout EOF
|
||||
// would pass the ScreenShareHostFault staleness gate and
|
||||
// surface as a spurious "ended unexpectedly" error on top of
|
||||
// the ticket error (Gemini review of S2, P2-1).
|
||||
current_sharing = None;
|
||||
|
||||
// If a network-mode / identity change was deferred while a call was
|
||||
// active, rebuild the persistent stack now — after the old session is
|
||||
@@ -1806,8 +1835,8 @@ async fn run_core_loop(
|
||||
secret_key.clone(),
|
||||
));
|
||||
|
||||
// Fresh join starts not sharing; clear any stale share ticket.
|
||||
current_sharing = None;
|
||||
// (The share was already retired beside the session teardown
|
||||
// above; a fresh join starts not sharing.)
|
||||
let self_state =
|
||||
presence.to_state(is_muted.load(Ordering::Relaxed), endpoint.addr(), None);
|
||||
|
||||
@@ -2828,8 +2857,11 @@ async fn run_core_loop(
|
||||
is_muted.store(new_state, Ordering::Relaxed);
|
||||
|
||||
if let Some(session) = &active_session {
|
||||
let self_state =
|
||||
presence.to_state(new_state, net.endpoint.addr(), current_sharing.clone());
|
||||
let self_state = presence.to_state(
|
||||
new_state,
|
||||
net.endpoint.addr(),
|
||||
current_sharing.as_ref().map(|s| s.ticket.clone()),
|
||||
);
|
||||
let _ = session.room_state.update_self_state(self_state).await;
|
||||
}
|
||||
}
|
||||
@@ -2842,7 +2874,7 @@ async fn run_core_loop(
|
||||
let self_state = presence.to_state(
|
||||
is_muted.load(Ordering::Relaxed),
|
||||
net.endpoint.addr(),
|
||||
current_sharing.clone(),
|
||||
current_sharing.as_ref().map(|s| s.ticket.clone()),
|
||||
);
|
||||
let _ = session.room_state.update_self_state(self_state).await;
|
||||
}
|
||||
@@ -3155,7 +3187,7 @@ async fn run_core_loop(
|
||||
let self_state = presence.to_state(
|
||||
is_muted.load(Ordering::Relaxed),
|
||||
net.endpoint.addr(),
|
||||
current_sharing.clone(),
|
||||
current_sharing.as_ref().map(|s| s.ticket.clone()),
|
||||
);
|
||||
let _ = session.room_state.update_self_state(self_state).await;
|
||||
}
|
||||
@@ -3361,7 +3393,7 @@ async fn run_core_loop(
|
||||
let self_state = presence.to_state(
|
||||
is_muted.load(Ordering::Relaxed),
|
||||
net.endpoint.addr(),
|
||||
current_sharing.clone(),
|
||||
current_sharing.as_ref().map(|s| s.ticket.clone()),
|
||||
);
|
||||
let _ = session.room_state.update_self_state(self_state).await;
|
||||
}
|
||||
@@ -3435,46 +3467,62 @@ async fn run_core_loop(
|
||||
continue;
|
||||
}
|
||||
};
|
||||
// Forward pixelpass `app_audio` events (only emitted when an app
|
||||
// is selected) to the UI so it can warn when the chosen app's
|
||||
// audio drops. The channel closes when the host dies (drain hits
|
||||
// EOF), ending the forwarder task on its own.
|
||||
let notices = audio_app.as_deref().map(|_| {
|
||||
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<
|
||||
crate::screenshare::PixelpassEvent,
|
||||
>();
|
||||
let ui_tx_notices = ui_tx.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(ev) = rx.recv().await {
|
||||
let active = match ev {
|
||||
crate::screenshare::PixelpassEvent::AppAudioRouted => true,
|
||||
crate::screenshare::PixelpassEvent::AppAudioLost => false,
|
||||
_ => continue,
|
||||
};
|
||||
if ui_tx_notices
|
||||
.send(UiEvent::ShareAudioActive(active))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
// Every share gets a notice forwarder — not just app-audio ones.
|
||||
// pixelpass `app_audio` events (only emitted when an app is
|
||||
// selected) become UI warnings, and the drain's terminal `Eof`
|
||||
// becomes a host fault scoped to this spawn's generation, so a
|
||||
// host that dies is torn down instead of staying advertised in
|
||||
// presence forever. On a failed spawn the sender is dropped
|
||||
// before the drain ever runs, so the forwarder just ends and no
|
||||
// fault is sent (the spawn error carries the news instead).
|
||||
share_generations += 1;
|
||||
let generation = share_generations;
|
||||
let (notices_tx, mut notices_rx) =
|
||||
tokio::sync::mpsc::unbounded_channel::<crate::screenshare::HostNotice>();
|
||||
let ui_tx_notices = ui_tx.clone();
|
||||
let fault_tx = host_fault_tx.clone();
|
||||
tokio::spawn(async move {
|
||||
while let Some(notice) = notices_rx.recv().await {
|
||||
match notice {
|
||||
crate::screenshare::HostNotice::Event(ev) => {
|
||||
let active = match ev {
|
||||
crate::screenshare::PixelpassEvent::AppAudioRouted => true,
|
||||
crate::screenshare::PixelpassEvent::AppAudioLost => false,
|
||||
_ => continue,
|
||||
};
|
||||
if ui_tx_notices
|
||||
.send(UiEvent::ShareAudioActive(active))
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
// Terminal by contract: nothing follows on the
|
||||
// channel, so the task ends here.
|
||||
crate::screenshare::HostNotice::Eof => {
|
||||
let _ = fault_tx.send(generation);
|
||||
break;
|
||||
}
|
||||
}
|
||||
});
|
||||
tx
|
||||
}
|
||||
});
|
||||
match crate::screenshare::spawn_host(
|
||||
&bin,
|
||||
audio_app.as_deref(),
|
||||
&settings,
|
||||
quality,
|
||||
notices,
|
||||
notices_tx,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok((child, ticket)) => {
|
||||
crate::log_msg("Screen share host started");
|
||||
session.teardown.set_host(child);
|
||||
current_sharing = Some(ticket.clone());
|
||||
current_sharing = Some(ActiveShare {
|
||||
generation,
|
||||
ticket: ticket.clone(),
|
||||
});
|
||||
let self_state = presence.to_state(
|
||||
is_muted.load(Ordering::Relaxed),
|
||||
net.endpoint.addr(),
|
||||
@@ -3523,6 +3571,66 @@ async fn run_core_loop(
|
||||
let _ = ui_tx.send(UiEvent::ScreenShareStopped).await;
|
||||
}
|
||||
|
||||
CoreCommand::ScreenShareHostFault { generation } => {
|
||||
// Stale unless it names the share we are advertising RIGHT NOW.
|
||||
// Every deliberate end of a share (StopScreenShare, Leave, a
|
||||
// fresh Join) clears `current_sharing` before or while reaping
|
||||
// the child, and the reaped child's stdout EOF then arrives
|
||||
// here late — dropping it is the correct handling, not an edge
|
||||
// case. A mismatched generation likewise: that fault belongs to
|
||||
// an older spawn than the share now running.
|
||||
let stale = current_sharing.as_ref().map(|s| s.generation) != Some(generation);
|
||||
if stale {
|
||||
continue;
|
||||
}
|
||||
crate::log_msg(
|
||||
"Screen share host died (stdout EOF with the share still advertised)",
|
||||
);
|
||||
current_sharing = None;
|
||||
// Pull the ticket off presence FIRST, before the reap: if the
|
||||
// child only closed stdout and lives on, `stop_host` burns the
|
||||
// full stop grace before the SIGKILL fallback, and for that
|
||||
// whole window peers would still see (and click Watch on) a
|
||||
// share whose host is already gone (Gemini S2-merge review,
|
||||
// P2-1).
|
||||
if let Some(session) = &mut active_session {
|
||||
let self_state = presence.to_state(
|
||||
is_muted.load(Ordering::Relaxed),
|
||||
net.endpoint.addr(),
|
||||
None,
|
||||
);
|
||||
let _ = session.room_state.update_self_state(self_state).await;
|
||||
}
|
||||
// Stopped next — it clears the UI's sharing state — so the
|
||||
// local UI also stops saying "sharing" before the reap wait,
|
||||
// and the error explaining why comes only after, so the user
|
||||
// is never left looking at a "sharing" UI with an error
|
||||
// beside it.
|
||||
let _ = ui_tx.send(UiEvent::ScreenShareStopped).await;
|
||||
let mut unconfirmed = false;
|
||||
if let Some(session) = &mut active_session {
|
||||
// The child is usually already dead, so this confirms the
|
||||
// reap immediately; if it merely closed stdout and lives
|
||||
// on, this is the SIGINT → grace → SIGKILL path. Either
|
||||
// way the dead-or-dying child leaves the teardown slot, so
|
||||
// `is_sharing` stops lying.
|
||||
unconfirmed = matches!(
|
||||
session.teardown.stop_host().await,
|
||||
Some(teardown::StopOutcome::Unconfirmed)
|
||||
);
|
||||
}
|
||||
let detail = if unconfirmed {
|
||||
" Its process also couldn't be confirmed dead — check for a stray pixelpass."
|
||||
} else {
|
||||
""
|
||||
};
|
||||
let _ = ui_tx
|
||||
.send(UiEvent::Error(format!(
|
||||
"Screen share ended unexpectedly — pixelpass exited.{detail}"
|
||||
)))
|
||||
.await;
|
||||
}
|
||||
|
||||
CoreCommand::ViewShare { ticket, settings } => {
|
||||
let bin = match crate::screenshare::pixelpass_path(pixelpass_override.as_deref()) {
|
||||
Some(b) => b,
|
||||
|
||||
+91
-8
@@ -90,6 +90,21 @@ pub enum PixelpassEvent {
|
||||
Other,
|
||||
}
|
||||
|
||||
/// What the host's stdout drain forwards to the core over the notice channel.
|
||||
///
|
||||
/// `Eof` is **synthesized here**, not parsed: pixelpass has no "I died" event,
|
||||
/// and a crash can abort across `extern "C"` before any JSON line is written,
|
||||
/// so the stream ending is the only reliable death signal. A read *error*
|
||||
/// counts too — either way the event stream is gone and the host must be
|
||||
/// treated as over.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub enum HostNotice {
|
||||
/// A parsed pixelpass event line.
|
||||
Event(PixelpassEvent),
|
||||
/// The host's stdout ended (EOF or read error). Terminal: nothing follows.
|
||||
Eof,
|
||||
}
|
||||
|
||||
/// Parse a single stdout line from pixelpass `--output json`. Pure: no I/O.
|
||||
pub fn parse_pixelpass_event(line: &str) -> Option<PixelpassEvent> {
|
||||
let line = line.trim();
|
||||
@@ -370,7 +385,10 @@ pub fn is_available(config_override: Option<&str>) -> bool {
|
||||
/// `audio_app` is `Some`, pixelpass captures only that app's audio instead of the
|
||||
/// whole desktop sink, which avoids the call-loopback echo (A23). The child keeps
|
||||
/// running (streaming to viewers) until killed or dropped; remaining stdout is
|
||||
/// drained in a background task so a full pipe can't stall the host. We do
|
||||
/// drained in a background task so a full pipe can't stall the host. The drain
|
||||
/// forwards every parsed event over `notices` and — the part no share may opt
|
||||
/// out of — a terminal [`HostNotice::Eof`] when the stream ends, which is the
|
||||
/// caller's only reliable signal that the host died. We do
|
||||
/// not pass encode/viewer overrides unless the local settings explicitly ask for
|
||||
/// them, so pixelpass keeps its own defaults in the common case.
|
||||
pub async fn spawn_host(
|
||||
@@ -378,7 +396,7 @@ pub async fn spawn_host(
|
||||
audio_app: Option<&str>,
|
||||
settings: &ScreenShareSettings,
|
||||
quality: ShareQuality,
|
||||
notices: Option<tokio::sync::mpsc::UnboundedSender<PixelpassEvent>>,
|
||||
notices: tokio::sync::mpsc::UnboundedSender<HostNotice>,
|
||||
) -> std::io::Result<(Child, String)> {
|
||||
let args = host_args(audio_app, settings, quality);
|
||||
// Log the exact argv we hand pixelpass so a field log can confirm which
|
||||
@@ -433,7 +451,7 @@ pub async fn spawn_host(
|
||||
if let Some(stderr) = stderr {
|
||||
drain_stderr_in_background(stderr);
|
||||
}
|
||||
drain_in_background(lines, "host", notices);
|
||||
drain_in_background(lines, "host", Some(notices));
|
||||
Ok((child, ticket))
|
||||
}
|
||||
|
||||
@@ -572,13 +590,15 @@ where
|
||||
|
||||
/// Keep reading the child's stdout to EOF in the background so a full pipe can't
|
||||
/// stall it; log notable events for diagnostics. When `notices` is `Some`, each
|
||||
/// parsed event is also forwarded to the caller (the core, which translates the
|
||||
/// `app_audio` ones into a UI warning); a send failure (receiver dropped) just
|
||||
/// stops forwarding, draining continues. The task ends on EOF (child exited).
|
||||
/// parsed event is also forwarded to the caller (the core), and when the stream
|
||||
/// ends — EOF or read error, i.e. the child exited or its event stream broke —
|
||||
/// a final [`HostNotice::Eof`] is sent so the caller learns the child is gone
|
||||
/// (a host that dies must not stay advertised as sharing). A send failure
|
||||
/// (receiver dropped) just stops forwarding, draining continues.
|
||||
fn drain_in_background<R>(
|
||||
mut lines: tokio::io::Lines<BufReader<R>>,
|
||||
role: &'static str,
|
||||
notices: Option<tokio::sync::mpsc::UnboundedSender<PixelpassEvent>>,
|
||||
notices: Option<tokio::sync::mpsc::UnboundedSender<HostNotice>>,
|
||||
) where
|
||||
R: tokio::io::AsyncRead + Unpin + Send + 'static,
|
||||
{
|
||||
@@ -587,10 +607,14 @@ fn drain_in_background<R>(
|
||||
if let Some(ev) = parse_pixelpass_event(&line) {
|
||||
crate::log_msg(&format!("pixelpass {role}: {}", event_for_log(&ev)));
|
||||
if let Some(tx) = ¬ices {
|
||||
let _ = tx.send(ev);
|
||||
let _ = tx.send(HostNotice::Event(ev));
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Some(tx) = ¬ices {
|
||||
crate::log_msg(&format!("pixelpass {role}: stdout ended"));
|
||||
let _ = tx.send(HostNotice::Eof);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@@ -1376,4 +1400,63 @@ Install hint: sudo apt install gstreamer1.0-plugins-bad
|
||||
#[cfg(not(windows))]
|
||||
assert_eq!(candidates, vec![dir.join("pixelpass")]);
|
||||
}
|
||||
|
||||
/// The host-fault contract, clean-exit half: events are forwarded in order
|
||||
/// and the stream ending yields exactly one terminal [`HostNotice::Eof`],
|
||||
/// after which the drain task drops its sender (the closed channel is what
|
||||
/// ends the core's forwarder). A host that dies silently — EOF swallowed —
|
||||
/// is the S2 defect: the dead share stays advertised in presence.
|
||||
#[tokio::test]
|
||||
async fn drain_forwards_events_then_synthesizes_eof_when_stdout_ends() {
|
||||
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
|
||||
let (read_half, mut write_half) = tokio::io::duplex(1024);
|
||||
drain_in_background(BufReader::new(read_half).lines(), "test", Some(tx));
|
||||
|
||||
use tokio::io::AsyncWriteExt;
|
||||
write_half
|
||||
.write_all(b"{\"event\":\"app_audio\",\"state\":\"routed\"}\nnot json\n")
|
||||
.await
|
||||
.unwrap();
|
||||
drop(write_half); // child exited: stdout EOF
|
||||
|
||||
assert_eq!(
|
||||
rx.recv().await,
|
||||
Some(HostNotice::Event(PixelpassEvent::AppAudioRouted))
|
||||
);
|
||||
// The non-JSON line is dropped, not forwarded.
|
||||
assert_eq!(rx.recv().await, Some(HostNotice::Eof));
|
||||
assert_eq!(rx.recv().await, None, "task ended and dropped the sender");
|
||||
}
|
||||
|
||||
/// The host-fault contract, broken-stream half: a read *error* (not a tidy
|
||||
/// EOF) must synthesize the same terminal `Eof` — the event stream is gone
|
||||
/// either way, and only the drain task can tell the core so.
|
||||
#[tokio::test]
|
||||
async fn drain_synthesizes_eof_on_a_read_error_too() {
|
||||
struct BrokenPipe;
|
||||
impl tokio::io::AsyncRead for BrokenPipe {
|
||||
fn poll_read(
|
||||
self: std::pin::Pin<&mut Self>,
|
||||
_cx: &mut std::task::Context<'_>,
|
||||
_buf: &mut tokio::io::ReadBuf<'_>,
|
||||
) -> std::task::Poll<std::io::Result<()>> {
|
||||
std::task::Poll::Ready(Err(std::io::Error::other("stream broke")))
|
||||
}
|
||||
}
|
||||
|
||||
use tokio::io::AsyncReadExt;
|
||||
// One good event line, then the stream breaks mid-read.
|
||||
let reader =
|
||||
std::io::Cursor::new(b"{\"event\":\"capture\",\"state\":\"started\"}\n".to_vec())
|
||||
.chain(BrokenPipe);
|
||||
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
|
||||
drain_in_background(BufReader::new(reader).lines(), "test", Some(tx));
|
||||
|
||||
assert_eq!(
|
||||
rx.recv().await,
|
||||
Some(HostNotice::Event(PixelpassEvent::CaptureStarted))
|
||||
);
|
||||
assert_eq!(rx.recv().await, Some(HostNotice::Eof));
|
||||
assert_eq!(rx.recv().await, None, "task ended and dropped the sender");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,589 @@
|
||||
//! S2 exit gate: a pixelpass host that dies mid-share must be torn down —
|
||||
//! reaped, pulled off presence, `ScreenShareStopped` emitted **before** the
|
||||
//! explanatory error — and a host stopped *deliberately* must NOT produce that
|
||||
//! error when its stdout EOF arrives late (the staleness gate).
|
||||
//!
|
||||
//! Drives the real core loop end to end through `CoreController`, with the
|
||||
//! pixelpass override pointed at fake shell scripts: one that emits a ticket
|
||||
//! and dies, one that emits a ticket and lives until signalled. This is the
|
||||
//! only harness that reaches the core's fault handler — the command loop has
|
||||
//! no unit seam — so these two halves are what kill the "forwarder drops the
|
||||
//! Eof" and "handler ignores the generation" mutants.
|
||||
//!
|
||||
//! Live: joins a real (solo) room, so it needs a working audio backend and
|
||||
//! network access for the endpoint bind.
|
||||
//! `cargo test --test screenshare_host_fault -- --ignored`
|
||||
|
||||
#![cfg(unix)]
|
||||
|
||||
use std::os::unix::fs::PermissionsExt;
|
||||
use std::path::PathBuf;
|
||||
use std::time::Duration;
|
||||
|
||||
use peerspeak::core::CoreController;
|
||||
use peerspeak::core::messages::{CoreCommand, UiEvent};
|
||||
|
||||
const EVENT_TIMEOUT: Duration = Duration::from_secs(20);
|
||||
/// How long to listen for events that must NOT arrive. Comfortably past the
|
||||
/// fake host's exit plus the drain/forwarder hop, so a stale fault that WOULD
|
||||
/// be mishandled has arrived by the end of it.
|
||||
const QUIET_WINDOW: Duration = Duration::from_secs(3);
|
||||
|
||||
/// Removes the fake-pixelpass dir even when an assertion panics mid-test
|
||||
/// (a plain trailing `remove_dir_all` never runs on an unwind).
|
||||
struct TempDir(PathBuf);
|
||||
|
||||
impl Drop for TempDir {
|
||||
fn drop(&mut self) {
|
||||
std::fs::remove_dir_all(&self.0).ok();
|
||||
}
|
||||
}
|
||||
|
||||
fn write_fake_pixelpass(dir: &std::path::Path, name: &str, body: &str) -> PathBuf {
|
||||
let path = dir.join(name);
|
||||
std::fs::write(&path, body).expect("write fake pixelpass");
|
||||
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))
|
||||
.expect("chmod fake pixelpass");
|
||||
path
|
||||
}
|
||||
|
||||
/// Skip events until `pick` matches, panicking after [`EVENT_TIMEOUT`].
|
||||
/// Unrelated events (identity, presence, chat plumbing) flow on this channel
|
||||
/// too, so gates scan rather than assert exact sequences.
|
||||
async fn wait_for<T>(
|
||||
rx: &mut tokio::sync::mpsc::Receiver<UiEvent>,
|
||||
what: &str,
|
||||
mut pick: impl FnMut(&UiEvent) -> Option<T>,
|
||||
) -> T {
|
||||
let deadline = tokio::time::Instant::now() + EVENT_TIMEOUT;
|
||||
loop {
|
||||
let ev = tokio::time::timeout_at(deadline, rx.recv())
|
||||
.await
|
||||
.unwrap_or_else(|_| panic!("timed out waiting for {what}"))
|
||||
.unwrap_or_else(|| panic!("ui channel closed waiting for {what}"));
|
||||
if let Some(v) = pick(&ev) {
|
||||
return v;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[ignore = "live: joins a real solo room (audio backend + network bind)"]
|
||||
async fn a_dead_host_is_torn_down_and_a_clean_stop_stays_clean() {
|
||||
let dir_guard =
|
||||
TempDir(std::env::temp_dir().join(format!("peerspeak-hostfault-{}", std::process::id())));
|
||||
let dir = dir_guard.0.clone();
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
|
||||
// Half 1's host: emits its ticket, then dies on its own — the S2 defect
|
||||
// scenario. Plain `sleep` (no exec) so the shell itself exits and closes
|
||||
// stdout with no orphan holding the pipe.
|
||||
let dying_host = write_fake_pixelpass(
|
||||
&dir,
|
||||
"pixelpass-dies",
|
||||
"#!/bin/sh\necho '{\"event\":\"ticket\",\"value\":\"fake-ticket-dies\"}'\nsleep 1\n",
|
||||
);
|
||||
// Half 2's host: lives until signalled. `exec` so the SIGINT from Stop
|
||||
// Share hits the sleep itself — the process dies AND its stdout closes,
|
||||
// which is exactly what makes the late Eof arrive and exercise the
|
||||
// staleness gate rather than vacuously never sending a fault.
|
||||
let living_host = write_fake_pixelpass(
|
||||
&dir,
|
||||
"pixelpass-lives",
|
||||
"#!/bin/sh\necho '{\"event\":\"ticket\",\"value\":\"fake-ticket-lives\"}'\nexec sleep 600\n",
|
||||
);
|
||||
|
||||
let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(256);
|
||||
let controller = CoreController::new(ui_tx);
|
||||
|
||||
assert!(controller.send(CoreCommand::SetPixelpassPath(Some(
|
||||
dying_host.to_string_lossy().into_owned()
|
||||
))));
|
||||
assert!(controller.send(CoreCommand::Join {
|
||||
name: "host-fault-gate".into(),
|
||||
ticket: "create".into(),
|
||||
room_name: "s2".into(),
|
||||
input_device: None,
|
||||
output_device: None,
|
||||
echo_cancellation: false,
|
||||
avatar: Default::default(),
|
||||
}));
|
||||
wait_for(&mut ui_rx, "RoomJoined", |ev| match ev {
|
||||
UiEvent::RoomJoined { .. } => Some(()),
|
||||
UiEvent::Error(e) => panic!("join failed: {e}"),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
// ── Half 1: the host dies mid-share ─────────────────────────────────────
|
||||
assert!(controller.send(CoreCommand::StartScreenShare {
|
||||
audio_app: None,
|
||||
settings: Default::default(),
|
||||
quality: Default::default(),
|
||||
}));
|
||||
wait_for(
|
||||
&mut ui_rx,
|
||||
"ScreenShareStarted (dying host)",
|
||||
|ev| match ev {
|
||||
UiEvent::ScreenShareStarted => Some(()),
|
||||
UiEvent::Error(e) => panic!("share start failed: {e}"),
|
||||
_ => None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
// The fake host exits ~1s in. The contract: ScreenShareStopped FIRST (it
|
||||
// clears the UI's sharing state), the explanatory error only after.
|
||||
wait_for(&mut ui_rx, "ScreenShareStopped after host death", |ev| {
|
||||
match ev {
|
||||
UiEvent::ScreenShareStopped => Some(()),
|
||||
// An error arriving first is the exact ordering defect S2 fixes:
|
||||
// the UI would show "sharing" next to the explanation.
|
||||
UiEvent::Error(e) => panic!("error arrived before ScreenShareStopped: {e}"),
|
||||
_ => None,
|
||||
}
|
||||
})
|
||||
.await;
|
||||
let err = wait_for(&mut ui_rx, "the host-death error", |ev| match ev {
|
||||
UiEvent::Error(e) => Some(e.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert!(
|
||||
err.contains("unexpectedly"),
|
||||
"the error should say the share ended unexpectedly, got: {err}"
|
||||
);
|
||||
|
||||
// ── Half 2: a deliberate stop must stay clean ───────────────────────────
|
||||
assert!(controller.send(CoreCommand::SetPixelpassPath(Some(
|
||||
living_host.to_string_lossy().into_owned()
|
||||
))));
|
||||
assert!(controller.send(CoreCommand::StartScreenShare {
|
||||
audio_app: None,
|
||||
settings: Default::default(),
|
||||
quality: Default::default(),
|
||||
}));
|
||||
wait_for(
|
||||
&mut ui_rx,
|
||||
"ScreenShareStarted (living host)",
|
||||
|ev| match ev {
|
||||
UiEvent::ScreenShareStarted => Some(()),
|
||||
UiEvent::Error(e) => panic!("second share start failed: {e}"),
|
||||
_ => None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
assert!(controller.send(CoreCommand::StopScreenShare));
|
||||
wait_for(
|
||||
&mut ui_rx,
|
||||
"ScreenShareStopped after Stop Share",
|
||||
|ev| match ev {
|
||||
UiEvent::ScreenShareStopped => Some(()),
|
||||
UiEvent::Error(e) => panic!("clean stop produced an error: {e}"),
|
||||
_ => None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
// The stopped host's stdout EOF is arriving about now as a *stale* fault
|
||||
// (its generation was retired when Stop Share cleared the share). Without
|
||||
// the staleness gate the handler would emit a second ScreenShareStopped
|
||||
// and a spurious "ended unexpectedly" error — listen long enough for that
|
||||
// mishandling to have shown up, and require silence.
|
||||
let deadline = tokio::time::Instant::now() + QUIET_WINDOW;
|
||||
while let Ok(Some(ev)) = tokio::time::timeout_at(deadline, ui_rx.recv()).await {
|
||||
match ev {
|
||||
UiEvent::ScreenShareStopped => {
|
||||
panic!("stale host fault re-emitted ScreenShareStopped after a clean stop")
|
||||
}
|
||||
UiEvent::Error(e) if e.contains("unexpectedly") => {
|
||||
panic!("stale host fault surfaced as an error after a clean stop: {e}")
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
// ── Half 3: a failed room switch while sharing must not cry "crash" ─────
|
||||
// Join tears the old session down (killing the host, deliberately) BEFORE
|
||||
// it validates the ticket, so an invalid ticket exits the Join arm early.
|
||||
// The share must be retired at the teardown itself — left advertised, the
|
||||
// killed host's EOF passes the staleness gate and a spurious "ended
|
||||
// unexpectedly" lands on top of the ticket error (Gemini review, P2-1).
|
||||
assert!(controller.send(CoreCommand::StartScreenShare {
|
||||
audio_app: None,
|
||||
settings: Default::default(),
|
||||
quality: Default::default(),
|
||||
}));
|
||||
wait_for(
|
||||
&mut ui_rx,
|
||||
"ScreenShareStarted (before failed switch)",
|
||||
|ev| match ev {
|
||||
UiEvent::ScreenShareStarted => Some(()),
|
||||
UiEvent::Error(e) => panic!("third share start failed: {e}"),
|
||||
_ => None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert!(controller.send(CoreCommand::Join {
|
||||
name: "host-fault-gate".into(),
|
||||
ticket: "definitely-not-a-ticket".into(),
|
||||
room_name: "s2".into(),
|
||||
input_device: None,
|
||||
output_device: None,
|
||||
echo_cancellation: false,
|
||||
avatar: Default::default(),
|
||||
}));
|
||||
wait_for(&mut ui_rx, "the invalid-ticket error", |ev| match ev {
|
||||
UiEvent::Error(e) if e.contains("invalid room ticket") => Some(()),
|
||||
UiEvent::Error(e) => panic!("unexpected error before the ticket error: {e}"),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
// The deliberately-killed host's EOF is arriving about now; it must be
|
||||
// dropped as stale, not reported as a crash.
|
||||
let deadline = tokio::time::Instant::now() + QUIET_WINDOW;
|
||||
while let Ok(Some(ev)) = tokio::time::timeout_at(deadline, ui_rx.recv()).await {
|
||||
match ev {
|
||||
UiEvent::ScreenShareStopped => {
|
||||
panic!("failed room switch re-emitted ScreenShareStopped for the torn-down share")
|
||||
}
|
||||
UiEvent::Error(e) if e.contains("unexpectedly") => {
|
||||
panic!("deliberate teardown during a failed room switch reported as a crash: {e}")
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// S2 presence gate: a host fault must pull the share ticket off PRESENCE —
|
||||
/// what remote peers actually see — and must do it BEFORE the reap wait, not
|
||||
/// after. Nothing on the sharer's own `UiEvent` channel can witness either
|
||||
/// half (presence is only observable from another node), so this test runs a
|
||||
/// real second core as an OBSERVER and asserts the sharer's `PeerState.sharing`
|
||||
/// goes `Some` → `None` on fault.
|
||||
///
|
||||
/// The observer runs in a SEPARATE PROCESS (`presence_probe_helper`, this same
|
||||
/// test binary re-invoked): two in-process cores would load the same
|
||||
/// `identity.key` and collapse into one node id, and swapping `XDG_CONFIG_HOME`
|
||||
/// between spawns in-process races other threads' getenv.
|
||||
///
|
||||
/// The fake host is a WEDGE — it closes stdout (the fault) but ignores SIGINT
|
||||
/// and lives until the SIGKILL fallback — so `stop_host` burns the full 2 s
|
||||
/// grace and TIME becomes the discriminator, exactly like the SIGINT gate:
|
||||
/// with presence-removal-first the observer sees the ticket clear ~1 s after
|
||||
/// it appeared (the wedge's pre-fault lifetime); with the old
|
||||
/// reap-then-presence ordering, only after ~3 s. The bound also makes the
|
||||
/// "presence removal deleted" mutant fail by timeout instead of passing
|
||||
/// vacuously.
|
||||
///
|
||||
/// Live: two real solo-room cores (audio backend + network bind each).
|
||||
#[tokio::test]
|
||||
#[ignore = "live: two real cores in one room (audio backend + network bind), observer subprocess"]
|
||||
async fn a_host_fault_pulls_the_ticket_off_presence_within_the_grace() {
|
||||
/// Mirrors `core::teardown::STOP_GRACE` (private): the wait the wedge
|
||||
/// forces before the SIGKILL fallback reaps it.
|
||||
const STOP_GRACE_MS: u128 = 2000;
|
||||
|
||||
let dir_guard = TempDir(
|
||||
std::env::temp_dir().join(format!("peerspeak-presence-gate-{}", std::process::id())),
|
||||
);
|
||||
let dir = dir_guard.0.clone();
|
||||
std::fs::create_dir_all(&dir).unwrap();
|
||||
|
||||
// Emits its ticket, shares for ~1 s, then closes stdout (the fault) while
|
||||
// staying alive and ignoring SIGINT, so the reap must wait out the grace.
|
||||
// The trailing sleep is NOT exec'd on purpose: it forks after stdout is
|
||||
// closed, so it holds no pipe (the vacuous-staleness trap doesn't apply),
|
||||
// and it merely idles out after the SIGKILL reaps the shell.
|
||||
//
|
||||
// The fake ticket must pass `screenshare::sanitize_ticket` (`endpoint` +
|
||||
// alphanumerics): the OBSERVER's gossip ingest sanitizes peer-advertised
|
||||
// tickets, and a garbage one is nulled to `sharing: None` there — the
|
||||
// probe would never see the share appear and the gate would go vacuous.
|
||||
let wedged_host = write_fake_pixelpass(
|
||||
&dir,
|
||||
"pixelpass-wedges",
|
||||
"#!/bin/sh\ntrap '' INT\n\
|
||||
echo '{\"event\":\"ticket\",\"value\":\"endpointaabwxjexzensznfvuudiapn5tyzws3angd2merarm\"}'\n\
|
||||
sleep 1\nexec 1>&-\nsleep 30\n",
|
||||
);
|
||||
|
||||
let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(256);
|
||||
let controller = CoreController::new(ui_tx);
|
||||
|
||||
assert!(controller.send(CoreCommand::SetPixelpassPath(Some(
|
||||
wedged_host.to_string_lossy().into_owned()
|
||||
))));
|
||||
assert!(controller.send(CoreCommand::Join {
|
||||
name: "presence-gate".into(),
|
||||
ticket: "create".into(),
|
||||
room_name: "s2-presence".into(),
|
||||
input_device: None,
|
||||
output_device: None,
|
||||
echo_cancellation: false,
|
||||
avatar: Default::default(),
|
||||
}));
|
||||
let room_ticket = wait_for(&mut ui_rx, "RoomJoined", |ev| match ev {
|
||||
UiEvent::RoomJoined { ticket, .. } => Some(ticket.clone()),
|
||||
UiEvent::Error(e) => panic!("join failed: {e}"),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
// The observer, in its own process with its own config dir (fresh
|
||||
// identity). It prints `PROBE …` lines this test parses.
|
||||
let probe_config = dir.join("probe-config");
|
||||
std::fs::create_dir_all(&probe_config).unwrap();
|
||||
let probe = tokio::process::Command::new(std::env::current_exe().unwrap())
|
||||
.kill_on_drop(true)
|
||||
.args([
|
||||
"presence_probe_helper",
|
||||
"--exact",
|
||||
"--ignored",
|
||||
"--nocapture",
|
||||
])
|
||||
.env("PEERSPEAK_PROBE_TICKET", &room_ticket)
|
||||
.env("XDG_CONFIG_HOME", &probe_config)
|
||||
.stdout(std::process::Stdio::piped())
|
||||
.stderr(std::process::Stdio::piped())
|
||||
.spawn()
|
||||
.expect("spawn the presence probe");
|
||||
|
||||
// Only share once the probe is in the room, so it witnesses the ticket
|
||||
// APPEARING before the fault clears it (otherwise `Some` → `None` could
|
||||
// both predate its join and the gate would go vacuous).
|
||||
wait_for(&mut ui_rx, "the probe's PeerJoined", |ev| match ev {
|
||||
UiEvent::PeerJoined { .. } => Some(()),
|
||||
UiEvent::Error(e) => panic!("waiting for the probe: {e}"),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
assert!(controller.send(CoreCommand::StartScreenShare {
|
||||
audio_app: None,
|
||||
settings: Default::default(),
|
||||
quality: Default::default(),
|
||||
}));
|
||||
wait_for(
|
||||
&mut ui_rx,
|
||||
"ScreenShareStarted (wedged host)",
|
||||
|ev| match ev {
|
||||
UiEvent::ScreenShareStarted => Some(()),
|
||||
UiEvent::Error(e) => panic!("share start failed: {e}"),
|
||||
_ => None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
// Sharer-side contract, unchanged by the reorder: Stopped first, the
|
||||
// explanatory error only after.
|
||||
wait_for(
|
||||
&mut ui_rx,
|
||||
"ScreenShareStopped after the wedge faults",
|
||||
|ev| match ev {
|
||||
UiEvent::ScreenShareStopped => Some(()),
|
||||
UiEvent::Error(e) => panic!("error arrived before ScreenShareStopped: {e}"),
|
||||
_ => None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let err = wait_for(&mut ui_rx, "the host-death error", |ev| match ev {
|
||||
UiEvent::Error(e) => Some(e.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
assert!(
|
||||
err.contains("unexpectedly"),
|
||||
"the error should say the share ended unexpectedly, got: {err}"
|
||||
);
|
||||
|
||||
let out = tokio::time::timeout(Duration::from_secs(60), probe.wait_with_output())
|
||||
.await
|
||||
.expect("probe process outlived its budget")
|
||||
.expect("probe process wait");
|
||||
let stdout = String::from_utf8_lossy(&out.stdout);
|
||||
let stderr = String::from_utf8_lossy(&out.stderr);
|
||||
assert!(
|
||||
out.status.success(),
|
||||
"probe failed ({}).\nstdout:\n{stdout}\nstderr:\n{stderr}",
|
||||
out.status
|
||||
);
|
||||
let cleared_ms: u128 = stdout
|
||||
.lines()
|
||||
.find_map(|l| l.strip_prefix("PROBE sharing-cleared "))
|
||||
.unwrap_or_else(|| {
|
||||
panic!("probe never saw the ticket clear from presence.\nstdout:\n{stdout}")
|
||||
})
|
||||
.trim()
|
||||
.parse()
|
||||
.expect("probe delta should be integer millis");
|
||||
// Presence-removal-first: ~1000 ms (the wedge's pre-fault lifetime).
|
||||
// Reap-then-presence: ~3000 ms (lifetime + the full stop grace). The
|
||||
// grace itself splits them with ~1 s of jitter headroom on each side.
|
||||
assert!(
|
||||
cleared_ms < STOP_GRACE_MS,
|
||||
"presence kept advertising the dead share for {cleared_ms} ms after it appeared — \
|
||||
at or past the wedge lifetime + stop grace, i.e. the ticket was only removed \
|
||||
AFTER the reap wait instead of before it"
|
||||
);
|
||||
|
||||
assert!(controller.send(CoreCommand::Leave));
|
||||
}
|
||||
|
||||
/// Observer half of `a_host_fault_pulls_the_ticket_off_presence_within_the_grace`,
|
||||
/// run BY that test as a subprocess. Standalone (no `PEERSPEAK_PROBE_TICKET` in
|
||||
/// the env — e.g. a plain `--ignored` sweep) it is a no-op pass.
|
||||
#[tokio::test]
|
||||
#[ignore = "helper: spawned by the presence gate as a subprocess; standalone it no-ops"]
|
||||
async fn presence_probe_helper() {
|
||||
let Ok(room_ticket) = std::env::var("PEERSPEAK_PROBE_TICKET") else {
|
||||
return;
|
||||
};
|
||||
|
||||
let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(256);
|
||||
let controller = CoreController::new(ui_tx);
|
||||
assert!(controller.send(CoreCommand::Join {
|
||||
name: "presence-probe".into(),
|
||||
ticket: room_ticket,
|
||||
room_name: String::new(),
|
||||
input_device: None,
|
||||
output_device: None,
|
||||
echo_cancellation: false,
|
||||
avatar: Default::default(),
|
||||
}));
|
||||
wait_for(&mut ui_rx, "RoomJoined (probe)", |ev| match ev {
|
||||
UiEvent::RoomJoined { .. } => Some(()),
|
||||
UiEvent::Error(e) => panic!("probe join failed: {e}"),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
// Watch the sharer's presence: record when its `sharing` ticket appears,
|
||||
// report the delta when it clears. Timings on both ends are local-loopback
|
||||
// arrival times, so the parent's bound compares like with like.
|
||||
let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
|
||||
let mut seen_at: Option<std::time::Instant> = None;
|
||||
loop {
|
||||
let ev = tokio::time::timeout_at(deadline, ui_rx.recv())
|
||||
.await
|
||||
.expect("probe timed out watching for the sharing transition")
|
||||
.expect("probe ui channel closed");
|
||||
let sharing = match &ev {
|
||||
UiEvent::PeerJoined { state, .. } | UiEvent::PeerUpdated { state, .. } => {
|
||||
state.sharing.is_some()
|
||||
}
|
||||
_ => continue,
|
||||
};
|
||||
match (&seen_at, sharing) {
|
||||
(None, true) => {
|
||||
seen_at = Some(std::time::Instant::now());
|
||||
println!("PROBE sharing-seen");
|
||||
}
|
||||
(Some(t0), false) => {
|
||||
println!("PROBE sharing-cleared {}", t0.elapsed().as_millis());
|
||||
break;
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
assert!(controller.send(CoreCommand::Leave));
|
||||
}
|
||||
|
||||
/// The long-owed Stop Share SIGINT gate (0c half (ii)), against the REAL
|
||||
/// pixelpass binary: a Stop Share must end the host through the graceful
|
||||
/// SIGINT path — child exits within [`STOP_GRACE`], no SIGKILL fallback, no
|
||||
/// "couldn't confirm" warning — because SIGKILL would skip pixelpass's own
|
||||
/// teardown (it unloads its capture sink on the way out in sink-owning modes).
|
||||
///
|
||||
/// The fallback is indistinguishable from success in the event stream (both
|
||||
/// end in a confirmed reap), so the discriminator is TIME: the fallback path
|
||||
/// first waits out the full 2 s grace, while a host honouring SIGINT exits in
|
||||
/// milliseconds. The bound asserts the stop completed inside the grace.
|
||||
///
|
||||
/// Live: needs `pixelpass` on `$PATH` plus a real solo room (audio + network).
|
||||
#[tokio::test]
|
||||
#[ignore = "live: real pixelpass host + a real solo room (audio backend, network bind)"]
|
||||
async fn stop_share_ends_the_real_host_via_sigint_within_the_grace() {
|
||||
/// Mirrors `core::teardown::STOP_GRACE` (private): the graceful wait
|
||||
/// before the SIGKILL fallback.
|
||||
const STOP_GRACE: Duration = Duration::from_secs(2);
|
||||
|
||||
let (ui_tx, mut ui_rx) = tokio::sync::mpsc::channel(256);
|
||||
let controller = CoreController::new(ui_tx);
|
||||
|
||||
// No override: resolve the real binary from $PATH.
|
||||
assert!(controller.send(CoreCommand::SetPixelpassPath(None)));
|
||||
assert!(controller.send(CoreCommand::Join {
|
||||
name: "sigint-gate".into(),
|
||||
ticket: "create".into(),
|
||||
room_name: "s2".into(),
|
||||
input_device: None,
|
||||
output_device: None,
|
||||
echo_cancellation: false,
|
||||
avatar: Default::default(),
|
||||
}));
|
||||
wait_for(&mut ui_rx, "RoomJoined", |ev| match ev {
|
||||
UiEvent::RoomJoined { .. } => Some(()),
|
||||
UiEvent::Error(e) => panic!("join failed: {e}"),
|
||||
_ => None,
|
||||
})
|
||||
.await;
|
||||
|
||||
// Whole-desktop share: no viewers ever connect, so the real host sits idle
|
||||
// after its ticket (capture starts on first viewer) — exactly the state a
|
||||
// Stop Share most often hits.
|
||||
assert!(controller.send(CoreCommand::StartScreenShare {
|
||||
audio_app: None,
|
||||
settings: Default::default(),
|
||||
quality: Default::default(),
|
||||
}));
|
||||
wait_for(
|
||||
&mut ui_rx,
|
||||
"ScreenShareStarted (real pixelpass)",
|
||||
|ev| match ev {
|
||||
UiEvent::ScreenShareStarted => Some(()),
|
||||
UiEvent::Error(e) => panic!("real pixelpass host failed to start: {e}"),
|
||||
_ => None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
let stop_started = std::time::Instant::now();
|
||||
assert!(controller.send(CoreCommand::StopScreenShare));
|
||||
wait_for(
|
||||
&mut ui_rx,
|
||||
"ScreenShareStopped (real pixelpass)",
|
||||
|ev| match ev {
|
||||
UiEvent::ScreenShareStopped => Some(()),
|
||||
// An Unconfirmed reap surfaces exactly this way; it means the
|
||||
// SIGINT AND the SIGKILL both failed to end the host.
|
||||
UiEvent::Error(e) => panic!("stop of the real host was not clean: {e}"),
|
||||
_ => None,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let elapsed = stop_started.elapsed();
|
||||
assert!(
|
||||
elapsed < STOP_GRACE,
|
||||
"stop took {elapsed:?} — at or past the {STOP_GRACE:?} grace, i.e. the \
|
||||
SIGKILL fallback fired instead of pixelpass honouring SIGINT"
|
||||
);
|
||||
|
||||
// And the late stdout EOF from the SIGINTed host must stay silent (same
|
||||
// staleness contract the fake-host half pins).
|
||||
let deadline = tokio::time::Instant::now() + QUIET_WINDOW;
|
||||
while let Ok(Some(ev)) = tokio::time::timeout_at(deadline, ui_rx.recv()).await {
|
||||
match ev {
|
||||
UiEvent::ScreenShareStopped => {
|
||||
panic!("stale fault from the SIGINTed real host re-emitted ScreenShareStopped")
|
||||
}
|
||||
UiEvent::Error(e) if e.contains("unexpectedly") => {
|
||||
panic!("stale fault from the SIGINTed real host surfaced as an error: {e}")
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
assert!(controller.send(CoreCommand::Leave));
|
||||
}
|
||||
Reference in New Issue
Block a user