fix(host): native sessions on the console + GPU-aware codecs + gamestream capability

The web console Dashboard read AppState.{streaming,launch,stream}, which only the
GameStream path writes, so a native punktfunk/1 session (the DEFAULT plane) showed
"Idle / no session" while actively streaming — only the Stats page (shared recorder)
reflected it. Add a plane-neutral per-session registry (session_status.rs) the native
video loop publishes to; /status now merges both planes, reports active_sessions, and
the Stop / Request-IDR buttons reach native sessions too (so surfacing them doesn't
leave dead buttons). LocalSummary (tray) gets the same fix.

Also on the management API:
- /host codecs derive from Codec::host_wire_caps() instead of a hardcoded
  [H264,H265,AV1], so codecs the GPU can't encode no longer appear.
- ApiCodec serializes HEVC as "hevc" (matching the wire/SDP/stats label) so the same
  codec reads identically across console pages.
- HostInfo.gamestream reports whether the GameStream planes run (--gamestream), so the
  console can hide the Moonlight-only pairing UI on the native-only default host.
- StatsStatus.elapsed_ms (host-monotonic) so the capture timer doesn't mix host/browser
  clocks.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-15 21:06:03 +02:00
parent c4645a8938
commit b8da32e8b6
7 changed files with 343 additions and 40 deletions
+24 -5
View File
@@ -2043,10 +2043,10 @@
}, },
"ApiCodec": { "ApiCodec": {
"type": "string", "type": "string",
"description": "Video codec identifier.", "description": "Video codec identifier. The wire token matches the codec's canonical name used across the\nstack (SDP/GameStream advertisement, the stats-capture `CaptureMeta.codec`, and the encoder's\n[`Codec::label`]) — notably `H.265` serializes as `\"hevc\"`, not `\"h265\"`, so the same codec\nreads identically on every console page.",
"enum": [ "enum": [
"h264", "h264",
"h265", "hevc",
"av1", "av1",
"pyrowave" "pyrowave"
] ]
@@ -2812,6 +2812,7 @@
"app_version", "app_version",
"gfe_version", "gfe_version",
"codecs", "codecs",
"gamestream",
"ports" "ports"
], ],
"properties": { "properties": {
@@ -2832,6 +2833,10 @@
}, },
"description": "Codecs the host can encode (NVENC)." "description": "Codecs the host can encode (NVENC)."
}, },
"gamestream": {
"type": "boolean",
"description": "Whether the GameStream/Moonlight-compat planes are running (`--gamestream`). `false` on the\nsecure default (native punktfunk/1 only) — a console can hide Moonlight-only UI (e.g. the\nMoonlight PIN pairing card, which could never receive a PIN when this is `false`)."
},
"gfe_version": { "gfe_version": {
"type": "string", "type": "string",
"description": "GFE version advertised to Moonlight clients." "description": "GFE version advertised to Moonlight clients."
@@ -3394,9 +3399,16 @@
"video_streaming", "video_streaming",
"audio_streaming", "audio_streaming",
"pin_pending", "pin_pending",
"paired_clients" "paired_clients",
"active_sessions"
], ],
"properties": { "properties": {
"active_sessions": {
"type": "integer",
"format": "int32",
"description": "Number of live streaming sessions across BOTH planes (GameStream + native punktfunk/1). The\nnative server admits concurrent sessions, so this can exceed 1; `session`/`stream` below\ndescribe a single representative session for the detail card.",
"minimum": 0
},
"audio_streaming": { "audio_streaming": {
"type": "boolean", "type": "boolean",
"description": "True while the audio stream thread is running." "description": "True while the audio stream thread is running."
@@ -3418,7 +3430,7 @@
}, },
{ {
"$ref": "#/components/schemas/SessionInfo", "$ref": "#/components/schemas/SessionInfo",
"description": "The active launch session (set by Moonlight's `/launch`, cleared on cancel/stop)." "description": "A representative active session. GameStream's launch (Moonlight `/launch`) when present, else\nthe first live native session. `null` when nothing is streaming."
} }
] ]
}, },
@@ -3429,7 +3441,7 @@
}, },
{ {
"$ref": "#/components/schemas/StreamInfo", "$ref": "#/components/schemas/StreamInfo",
"description": "The RTSP-negotiated stream parameters (present once a client has completed ANNOUNCE)." "description": "The active stream's parameters — RTSP-negotiated for GameStream, or the live native session's\nmode/codec/bitrate. `null` when nothing is streaming."
} }
] ]
}, },
@@ -3600,6 +3612,7 @@
"armed", "armed",
"sample_count", "sample_count",
"started_unix_ms", "started_unix_ms",
"elapsed_ms",
"kind" "kind"
], ],
"properties": { "properties": {
@@ -3607,6 +3620,12 @@
"type": "boolean", "type": "boolean",
"description": "Capture currently running." "description": "Capture currently running."
}, },
"elapsed_ms": {
"type": "integer",
"format": "int64",
"description": "Host-measured elapsed time of the in-progress capture, in ms (`0` if idle). Computed from the\nhost's MONOTONIC clock, so a console can show elapsed time without subtracting `started_unix_ms`\nfrom its own (possibly skewed) wall clock.",
"minimum": 0
},
"kind": { "kind": {
"type": "string", "type": "string",
"description": "Path of the in-progress capture (`\"\"` if idle)." "description": "Path of the in-progress capture (`\"\"` if idle)."
+2 -2
View File
@@ -252,7 +252,7 @@ pub fn serve(
); );
tokio::try_join!( tokio::try_join!(
nvhttp::run(state.clone()), nvhttp::run(state.clone()),
crate::mgmt::run(state.clone(), mgmt, Some(np.clone()), stats.clone()), crate::mgmt::run(state.clone(), mgmt, Some(np.clone()), stats.clone(), gamestream),
crate::punktfunk1::serve(native_opts, native.mgmt_port, np, stats.clone()), crate::punktfunk1::serve(native_opts, native.mgmt_port, np, stats.clone()),
)?; )?;
} else { } else {
@@ -263,7 +263,7 @@ pub fn serve(
(GameStream OFF — pass --gamestream for stock-Moonlight compat)" (GameStream OFF — pass --gamestream for stock-Moonlight compat)"
); );
tokio::try_join!( tokio::try_join!(
crate::mgmt::run(state.clone(), mgmt, Some(np.clone()), stats.clone()), crate::mgmt::run(state.clone(), mgmt, Some(np.clone()), stats.clone(), gamestream),
crate::punktfunk1::serve(native_opts, native.mgmt_port, np, stats.clone()), crate::punktfunk1::serve(native_opts, native.mgmt_port, np, stats.clone()),
)?; )?;
} }
+1
View File
@@ -68,6 +68,7 @@ mod send_pacing;
#[path = "windows/service.rs"] #[path = "windows/service.rs"]
mod service; mod service;
mod session_plan; mod session_plan;
mod session_status;
mod session_tuning; mod session_tuning;
mod spike; mod spike;
mod stats_recorder; mod stats_recorder;
+139 -15
View File
@@ -76,6 +76,10 @@ struct MgmtState {
/// Shared streaming-stats recorder — the same handle the streaming loops emit into, so an /// Shared streaming-stats recorder — the same handle the streaming loops emit into, so an
/// operator can arm/stop a capture here and review/list/delete saved recordings. /// operator can arm/stop a capture here and review/list/delete saved recordings.
stats: Arc<crate::stats_recorder::StatsRecorder>, stats: Arc<crate::stats_recorder::StatsRecorder>,
/// Whether this host runs the GameStream/Moonlight-compat planes (`--gamestream`). Surfaced in
/// [`HostInfo`] so the web console can hide the Moonlight-only pairing UI on the secure default
/// (native-only) host, where a Moonlight PIN can never arrive.
gamestream_enabled: bool,
token: Option<String>, token: Option<String>,
/// The port we serve on, echoed in [`PortMap`] so a client can persist a full endpoint map. /// The port we serve on, echoed in [`PortMap`] so a client can persist a full endpoint map.
port: u16, port: u16,
@@ -88,6 +92,7 @@ pub async fn run(
opts: Options, opts: Options,
native: Option<Arc<crate::native_pairing::NativePairing>>, native: Option<Arc<crate::native_pairing::NativePairing>>,
stats: Arc<crate::stats_recorder::StatsRecorder>, stats: Arc<crate::stats_recorder::StatsRecorder>,
gamestream_enabled: bool,
) -> Result<()> { ) -> Result<()> {
// The mgmt API is HTTPS + token-authenticated ALWAYS (even on loopback): `parse_serve` // The mgmt API is HTTPS + token-authenticated ALWAYS (even on loopback): `parse_serve`
// guarantees a token (CLI flag / env / persisted ~/.config/punktfunk/mgmt-token / generated). // guarantees a token (CLI flag / env / persisted ~/.config/punktfunk/mgmt-token / generated).
@@ -111,7 +116,14 @@ pub async fn run(
auth = "mTLS (paired cert) or bearer (required)", auth = "mTLS (paired cert) or bearer (required)",
"management API listening over HTTPS (docs at /api/docs, spec at /api/v1/openapi.json)" "management API listening over HTTPS (docs at /api/docs, spec at /api/v1/openapi.json)"
); );
let app = app(state, Some(token), opts.bind.port(), native, stats); let app = app(
state,
Some(token),
opts.bind.port(),
native,
stats,
gamestream_enabled,
);
serve_https(opts.bind, app, tls).await serve_https(opts.bind, app, tls).await
} }
@@ -122,11 +134,13 @@ fn app(
port: u16, port: u16,
native: Option<Arc<crate::native_pairing::NativePairing>>, native: Option<Arc<crate::native_pairing::NativePairing>>,
stats: Arc<crate::stats_recorder::StatsRecorder>, stats: Arc<crate::stats_recorder::StatsRecorder>,
gamestream_enabled: bool,
) -> Router { ) -> Router {
let shared = Arc::new(MgmtState { let shared = Arc::new(MgmtState {
app: state, app: state,
native, native,
stats, stats,
gamestream_enabled,
token, token,
port, port,
}); });
@@ -284,6 +298,10 @@ struct HostInfo {
gfe_version: String, gfe_version: String,
/// Codecs the host can encode (NVENC). /// Codecs the host can encode (NVENC).
codecs: Vec<ApiCodec>, codecs: Vec<ApiCodec>,
/// Whether the GameStream/Moonlight-compat planes are running (`--gamestream`). `false` on the
/// secure default (native punktfunk/1 only) — a console can hide Moonlight-only UI (e.g. the
/// Moonlight PIN pairing card, which could never receive a PIN when this is `false`).
gamestream: bool,
ports: PortMap, ports: PortMap,
} }
@@ -303,11 +321,15 @@ struct PortMap {
audio: u16, audio: u16,
} }
/// Video codec identifier. /// Video codec identifier. The wire token matches the codec's canonical name used across the
/// stack (SDP/GameStream advertisement, the stats-capture `CaptureMeta.codec`, and the encoder's
/// [`Codec::label`]) — notably `H.265` serializes as `"hevc"`, not `"h265"`, so the same codec
/// reads identically on every console page.
#[derive(Clone, Copy, Serialize, Deserialize, ToSchema, PartialEq, Eq, Debug)] #[derive(Clone, Copy, Serialize, Deserialize, ToSchema, PartialEq, Eq, Debug)]
#[serde(rename_all = "lowercase")] #[serde(rename_all = "lowercase")]
enum ApiCodec { enum ApiCodec {
H264, H264,
#[serde(rename = "hevc")]
H265, H265,
Av1, Av1,
/// PyroWave — the opt-in wired-LAN intra-only wavelet codec. /// PyroWave — the opt-in wired-LAN intra-only wavelet codec.
@@ -337,9 +359,15 @@ struct RuntimeStatus {
pin_pending: bool, pin_pending: bool,
/// Number of pinned (paired) client certificates. /// Number of pinned (paired) client certificates.
paired_clients: u32, paired_clients: u32,
/// The active launch session (set by Moonlight's `/launch`, cleared on cancel/stop). /// Number of live streaming sessions across BOTH planes (GameStream + native punktfunk/1). The
/// native server admits concurrent sessions, so this can exceed 1; `session`/`stream` below
/// describe a single representative session for the detail card.
active_sessions: u32,
/// A representative active session. GameStream's launch (Moonlight `/launch`) when present, else
/// the first live native session. `null` when nothing is streaming.
session: Option<SessionInfo>, session: Option<SessionInfo>,
/// The RTSP-negotiated stream parameters (present once a client has completed ANNOUNCE). /// The active stream's parameters — RTSP-negotiated for GameStream, or the live native session's
/// mode/codec/bitrate. `null` when nothing is streaming.
stream: Option<StreamInfo>, stream: Option<StreamInfo>,
} }
@@ -677,8 +705,25 @@ async fn get_host_info(State(st): State<Arc<MgmtState>>) -> Json<HostInfo> {
abi_version: punktfunk_core::ABI_VERSION, abi_version: punktfunk_core::ABI_VERSION,
app_version: APP_VERSION.into(), app_version: APP_VERSION.into(),
gfe_version: GFE_VERSION.into(), gfe_version: GFE_VERSION.into(),
// Everything NVENC encodes here (mirrors SERVER_CODEC_MODE_SUPPORT = 3843). // What this host can ACTUALLY encode on its resolved backend — GPU-aware, straight from the
codecs: vec![ApiCodec::H264, ApiCodec::H265, ApiCodec::Av1], // same capability mask that drives GameStream/QUIC negotiation ([`Codec::host_wire_caps`]).
// So an iGPU without AV1 encode won't advertise AV1, a software-only host reports H.264 only,
// and PyroWave appears only when its opt-in feature is built and the backend can open.
codecs: {
let caps = Codec::host_wire_caps();
use punktfunk_core::quic::{CODEC_AV1, CODEC_H264, CODEC_HEVC, CODEC_PYROWAVE};
[
(CODEC_H264, ApiCodec::H264),
(CODEC_HEVC, ApiCodec::H265),
(CODEC_AV1, ApiCodec::Av1),
(CODEC_PYROWAVE, ApiCodec::PyroWave),
]
.into_iter()
.filter(|(bit, _)| caps & bit != 0)
.map(|(_, codec)| codec)
.collect()
},
gamestream: st.gamestream_enabled,
ports: PortMap { ports: PortMap {
mgmt: st.port, mgmt: st.port,
http: h.http_port, http: h.http_port,
@@ -1397,12 +1442,32 @@ async fn delete_custom_preset(Path(id): Path<String>) -> Response {
) )
)] )]
async fn get_status(State(st): State<Arc<MgmtState>>) -> Json<RuntimeStatus> { async fn get_status(State(st): State<Arc<MgmtState>>) -> Json<RuntimeStatus> {
let session = st.app.launch.lock().unwrap().map(|l| SessionInfo { // GameStream plane (set by RTSP/nvhttp on the compat path).
let gs_launch = *st.app.launch.lock().unwrap();
let gs_stream = *st.app.stream.lock().unwrap();
let gs_video = st.app.streaming.load(Ordering::SeqCst);
let gs_audio = st.app.audio_streaming.load(Ordering::SeqCst);
// Native punktfunk/1 plane (published by the native video loop; the default plane). See
// [`crate::session_status`] for why this lives outside `AppState`.
let native = crate::session_status::snapshot();
// Detail card is singular: prefer a live GameStream session, else the first native one.
// `active_sessions` conveys the true count when several native clients stream at once.
let session = gs_launch
.map(|l| SessionInfo {
width: l.width, width: l.width,
height: l.height, height: l.height,
fps: l.fps, fps: l.fps,
})
.or_else(|| {
native.first().map(|s| SessionInfo {
width: s.width,
height: s.height,
fps: s.fps,
})
}); });
let stream = st.app.stream.lock().unwrap().as_ref().map(|c| StreamInfo { let stream = gs_stream
.map(|c| StreamInfo {
width: c.width, width: c.width,
height: c.height, height: c.height,
fps: c.fps, fps: c.fps,
@@ -1410,12 +1475,26 @@ async fn get_status(State(st): State<Arc<MgmtState>>) -> Json<RuntimeStatus> {
packet_size: c.packet_size as u32, packet_size: c.packet_size as u32,
min_fec: c.min_fec, min_fec: c.min_fec,
codec: c.codec.into(), codec: c.codec.into(),
})
.or_else(|| {
native.first().map(|s| StreamInfo {
width: s.width,
height: s.height,
fps: s.fps,
bitrate_kbps: s.bitrate_kbps,
// FEC/packetization are RTSP-negotiated (GameStream only); the native QUIC plane
// shards differently, so these are 0 (not applicable) for a native session.
packet_size: 0,
min_fec: 0,
codec: s.codec.into(),
})
}); });
Json(RuntimeStatus { Json(RuntimeStatus {
video_streaming: st.app.streaming.load(Ordering::SeqCst), video_streaming: gs_video || !native.is_empty(),
audio_streaming: st.app.audio_streaming.load(Ordering::SeqCst), audio_streaming: gs_audio || !native.is_empty(),
pin_pending: st.app.pairing.pin.awaiting_pin(), pin_pending: st.app.pairing.pin.awaiting_pin(),
paired_clients: st.app.paired.lock().unwrap().len() as u32, paired_clients: st.app.paired.lock().unwrap().len() as u32,
active_sessions: native.len() as u32 + u32::from(gs_video),
session, session,
stream, stream,
}) })
@@ -1438,10 +1517,24 @@ async fn get_status(State(st): State<Arc<MgmtState>>) -> Json<RuntimeStatus> {
) )
)] )]
async fn get_local_summary(State(st): State<Arc<MgmtState>>) -> Json<LocalSummary> { async fn get_local_summary(State(st): State<Arc<MgmtState>>) -> Json<LocalSummary> {
let session = st.app.launch.lock().unwrap().map(|l| SessionInfo { // GameStream launch, else the first live native session — so the tray reflects a native session
// too (same GameStream-only blind spot the Dashboard `/status` had; see `session_status`).
let session = st
.app
.launch
.lock()
.unwrap()
.map(|l| SessionInfo {
width: l.width, width: l.width,
height: l.height, height: l.height,
fps: l.fps, fps: l.fps,
})
.or_else(|| {
crate::session_status::snapshot().first().map(|s| SessionInfo {
width: s.width,
height: s.height,
fps: s.fps,
})
}); });
let (native_paired_clients, pending_approvals) = st let (native_paired_clients, pending_approvals) = st
.native .native
@@ -1907,7 +2000,11 @@ async fn stop_session(State(st): State<Arc<MgmtState>>) -> StatusCode {
st.app.audio_streaming.store(false, Ordering::SeqCst); st.app.audio_streaming.store(false, Ordering::SeqCst);
*st.app.launch.lock().unwrap() = None; *st.app.launch.lock().unwrap() = None;
*st.app.stream.lock().unwrap() = None; *st.app.stream.lock().unwrap() = None;
tracing::info!(was_streaming, "management API: session stopped"); // Native plane: the GameStream flags above don't reach it (it runs its own loops off the shared
// session registry), so signal every live native session to tear down too.
let native = crate::session_status::count();
crate::session_status::stop_all();
tracing::info!(was_streaming, native_sessions = native, "management API: session stopped");
StatusCode::NO_CONTENT StatusCode::NO_CONTENT
} }
@@ -1927,10 +2024,16 @@ async fn stop_session(State(st): State<Arc<MgmtState>>) -> StatusCode {
) )
)] )]
async fn request_idr(State(st): State<Arc<MgmtState>>) -> Response { async fn request_idr(State(st): State<Arc<MgmtState>>) -> Response {
if !st.app.streaming.load(Ordering::SeqCst) { let gs = st.app.streaming.load(Ordering::SeqCst);
let native = crate::session_status::count();
if !gs && native == 0 {
return api_error(StatusCode::CONFLICT, "no active video stream"); return api_error(StatusCode::CONFLICT, "no active video stream");
} }
if gs {
st.app.force_idr.store(true, Ordering::SeqCst); st.app.force_idr.store(true, Ordering::SeqCst);
}
// Native sessions get the keyframe request through their registry flag (see `session_status`).
crate::session_status::force_idr_all();
StatusCode::ACCEPTED.into_response() StatusCode::ACCEPTED.into_response()
} }
@@ -2332,6 +2435,8 @@ mod tests {
DEFAULT_PORT, DEFAULT_PORT,
None, None,
stats, stats,
// GameStream-compat planes off (the secure default the native-only tests model).
false,
) )
} }
@@ -2348,6 +2453,7 @@ mod tests {
DEFAULT_PORT, DEFAULT_PORT,
Some(np), Some(np),
stats, stats,
false,
) )
} }
@@ -2655,7 +2761,25 @@ mod tests {
assert_eq!(body["uniqueid"], "deadbeef"); assert_eq!(body["uniqueid"], "deadbeef");
assert_eq!(body["ports"]["http"], HTTP_PORT); assert_eq!(body["ports"]["http"], HTTP_PORT);
assert_eq!(body["ports"]["mgmt"], DEFAULT_PORT); assert_eq!(body["ports"]["mgmt"], DEFAULT_PORT);
assert_eq!(body["codecs"], serde_json::json!(["h264", "h265", "av1"])); // Codecs are GPU-aware (derived from `Codec::host_wire_caps`), so assert against that mask
// rather than a fixed set — and confirm HEVC serializes as "hevc" (the unified codec label),
// never "h265".
use punktfunk_core::quic::{CODEC_AV1, CODEC_H264, CODEC_HEVC, CODEC_PYROWAVE};
let caps = Codec::host_wire_caps();
let expected: Vec<&str> = [
(CODEC_H264, "h264"),
(CODEC_HEVC, "hevc"),
(CODEC_AV1, "av1"),
(CODEC_PYROWAVE, "pyrowave"),
]
.into_iter()
.filter(|(bit, _)| caps & bit != 0)
.map(|(_, name)| name)
.collect();
assert_eq!(body["codecs"], serde_json::json!(expected));
assert!(caps & CODEC_H264 != 0, "H.264 is always encodable");
// test_app models the secure default (GameStream-compat off).
assert_eq!(body["gamestream"], false);
} }
#[tokio::test] #[tokio::test]
@@ -2800,7 +2924,7 @@ mod tests {
bind: "127.0.0.1:0".parse().unwrap(), bind: "127.0.0.1:0".parse().unwrap(),
token: Some(" ".into()), token: Some(" ".into()),
}; };
let err = run(test_state(), opts, None, test_stats()) let err = run(test_state(), opts, None, test_stats(), false)
.await .await
.unwrap_err(); .unwrap_err();
assert!(err.to_string().contains("no token"), "{err}"); assert!(err.to_string().contains("no token"), "{err}");
+23 -1
View File
@@ -3584,7 +3584,7 @@ fn pack_mode(width: u32, height: u32, refresh_hz: u32) -> u64 {
} }
/// Unpack a [`pack_mode`] word back into `(width, height, refresh_hz)`. /// Unpack a [`pack_mode`] word back into `(width, height, refresh_hz)`.
fn unpack_mode(packed: u64) -> (u32, u32, u32) { pub(crate) fn unpack_mode(packed: u64) -> (u32, u32, u32) {
( (
((packed >> 32) & 0xffff) as u32, ((packed >> 32) & 0xffff) as u32,
((packed >> 16) & 0xffff) as u32, ((packed >> 16) & 0xffff) as u32,
@@ -4210,6 +4210,11 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
mode.height, mode.height,
interval_hz(interval), interval_hz(interval),
))); )));
// One-shot force-keyframe flag driven by the management API (`POST /session/idr`, the web-console
// Dashboard's "Request IDR" button) — drained in the encode loop below exactly like a client
// decode-recovery request. Registered with `session_status` so the mgmt handler can reach THIS
// session (the native plane never touches the GameStream `AppState.force_idr`).
let force_idr = Arc::new(AtomicBool::new(false));
// The send thread emits the web-console stats sample (it owns `session.stats()`); clone the // The send thread emits the web-console stats sample (it owns `session.stats()`); clone the
// recorder so the capture loop keeps its own handle for the per-frame `is_armed()` gate. // recorder so the capture loop keeps its own handle for the per-frame `is_armed()` gate.
let send_stats = SendStats { let send_stats = SendStats {
@@ -4241,6 +4246,18 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
}) })
.context("spawn send thread")?; .context("spawn send thread")?;
// Publish this session to the plane-neutral live-session registry so the web-console Dashboard
// (`GET /status`) shows the native stream — resolution/fps/codec/bitrate resolve live from the
// same handles a mid-stream mode switch / adaptive-bitrate change updates. The guard clears the
// entry when this loop exits (return / `?` / panic), so the Dashboard tracks the session's life.
let _live_session = crate::session_status::register(
live_mode.clone(),
live_bitrate.clone(),
plan.codec,
stop.clone(),
force_idr.clone(),
);
// Mid-stream session-switch watcher (opt-in via PUNKTFUNK_SESSION_WATCH; never under an explicit // Mid-stream session-switch watcher (opt-in via PUNKTFUNK_SESSION_WATCH; never under an explicit
// PUNKTFUNK_COMPOSITOR pin). It self-baselines and signals the loop below to swap the backend in // PUNKTFUNK_COMPOSITOR pin). It self-baselines and signals the loop below to swap the backend in
// place when the box flips Gaming↔Desktop. When not spawned, session_rx just stays empty. // place when the box flips Gaming↔Desktop. When not spawned, session_rx just stays empty.
@@ -4574,6 +4591,11 @@ fn virtual_stream(ctx: SessionContext) -> Result<()> {
while keyframe.try_recv().is_ok() { while keyframe.try_recv().is_ok() {
want_kf = true; want_kf = true;
} }
// Management API `POST /session/idr` (web-console Dashboard) targets this session's registry
// flag; drain it into the same forced-keyframe path a client decode-recovery request takes.
if force_idr.swap(false, Ordering::Relaxed) {
want_kf = true;
}
// Client LTR-RFI recovery: prefer re-referencing a known-good older frame (a clean recovery // Client LTR-RFI recovery: prefer re-referencing a known-good older frame (a clean recovery
// P-frame — no 20-40× IDR spike) over a full keyframe when the encoder supports it (native // P-frame — no 20-40× IDR spike) over a full keyframe when the encoder supports it (native
// AMF LTR / Windows NVENC). Drain the backlog (the client re-requests until the recovery // AMF LTR / Windows NVENC). Drain the backlog (the client re-requests until the recovery
+131
View File
@@ -0,0 +1,131 @@
//! Plane-neutral live-session status for the management `/status` (web-console Dashboard) view.
//!
//! The GameStream media pipeline records its session in `AppState.{launch, stream, streaming}`
//! (consumed by RTSP/media), but the native punktfunk/1 plane never touches `AppState` — by design
//! it is handed only the shared stats recorder ([`crate::punktfunk1::serve`]). So a native session,
//! which is the DEFAULT plane (GameStream is opt-in, `--gamestream`), was invisible on the Dashboard:
//! `GET /status` reported `video_streaming: false` and no session/stream card while a client was
//! actively streaming (the Stats page worked because it shares the recorder — hence the confusing
//! "stats move but the dashboard says idle").
//!
//! This module is the small shared surface the native video loop ([`crate::punktfunk1::virtual_stream`])
//! publishes a live snapshot to, keyed per session so CONCURRENT native sessions each get an entry
//! (the native server admits up to `max_sessions`, unbounded by default). The loop registers on
//! stream start and the returned [`LiveSessionGuard`] removes the entry on ANY scope exit (return,
//! `?`, panic — RAII). `/status` reads [`snapshot`]/[`count`]; the Dashboard's session-control
//! buttons reach a native session through [`stop_all`] (stop) and [`force_idr_all`] (request IDR),
//! so surfacing the session doesn't leave those buttons dead.
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use crate::encode::Codec;
/// A single live native session. Holds the SAME live Arc handles the video loop already maintains,
/// so a mid-stream mode switch / adaptive-bitrate change is reflected on the Dashboard with no
/// second update path — plus the flags the mgmt API flips to control the session.
struct LiveSession {
id: u64,
/// Packed `w:16|h:16|hz:16` ([`crate::punktfunk1::pack_mode`]); updated on a mode switch.
mode: Arc<AtomicU64>,
/// Live encoder target (kbps); updated on an adaptive-bitrate change.
bitrate_kbps: Arc<AtomicU32>,
codec: Codec,
/// The session's teardown flag ([`stop_all`] → mgmt `DELETE /session`).
stop: Arc<AtomicBool>,
/// One-shot force-keyframe flag ([`force_idr_all`] → mgmt `POST /session/idr`); the encode loop
/// drains it alongside a client's decode-recovery keyframe request.
force_idr: Arc<AtomicBool>,
}
/// A resolved read of one live session, for the `/status` view.
#[derive(Clone, Copy)]
pub struct SessionSnapshot {
pub width: u32,
pub height: u32,
pub fps: u32,
pub bitrate_kbps: u32,
pub codec: Codec,
}
fn registry() -> &'static Mutex<Vec<LiveSession>> {
static REG: OnceLock<Mutex<Vec<LiveSession>>> = OnceLock::new();
REG.get_or_init(|| Mutex::new(Vec::new()))
}
fn next_id() -> u64 {
static ID: AtomicU64 = AtomicU64::new(1);
ID.fetch_add(1, Ordering::Relaxed)
}
/// Registers a live native session; the returned guard removes it on drop (session end).
pub fn register(
mode: Arc<AtomicU64>,
bitrate_kbps: Arc<AtomicU32>,
codec: Codec,
stop: Arc<AtomicBool>,
force_idr: Arc<AtomicBool>,
) -> LiveSessionGuard {
let id = next_id();
registry().lock().unwrap().push(LiveSession {
id,
mode,
bitrate_kbps,
codec,
stop,
force_idr,
});
LiveSessionGuard { id }
}
/// Removes its session from the registry when dropped (any scope exit of the native video loop).
pub struct LiveSessionGuard {
id: u64,
}
impl Drop for LiveSessionGuard {
fn drop(&mut self) {
registry().lock().unwrap().retain(|s| s.id != self.id);
}
}
/// The number of live native sessions.
pub fn count() -> usize {
registry().lock().unwrap().len()
}
/// A resolved snapshot of every live native session (mode/bitrate read live), newest last.
pub fn snapshot() -> Vec<SessionSnapshot> {
registry()
.lock()
.unwrap()
.iter()
.map(|s| {
let (width, height, fps) =
crate::punktfunk1::unpack_mode(s.mode.load(Ordering::Relaxed));
SessionSnapshot {
width,
height,
fps,
bitrate_kbps: s.bitrate_kbps.load(Ordering::Relaxed),
codec: s.codec,
}
})
.collect()
}
/// Signals every live native session to tear down (mgmt `DELETE /session`). Best-effort: the video
/// + send loops observe the flag and exit, ending the stream; the guard then clears the entry.
pub fn stop_all() {
for s in registry().lock().unwrap().iter() {
s.stop.store(true, Ordering::SeqCst);
}
}
/// Requests a forced keyframe on every live native session (mgmt `POST /session/idr`). The encode
/// loop drains the flag exactly like a client decode-recovery request.
pub fn force_idr_all() {
for s in registry().lock().unwrap().iter() {
s.force_idr.store(true, Ordering::Relaxed);
}
}
@@ -96,6 +96,10 @@ pub struct StatsStatus {
pub sample_count: u32, pub sample_count: u32,
/// Unix start time of the in-progress capture (`0` if idle). /// Unix start time of the in-progress capture (`0` if idle).
pub started_unix_ms: u64, pub started_unix_ms: u64,
/// Host-measured elapsed time of the in-progress capture, in ms (`0` if idle). Computed from the
/// host's MONOTONIC clock, so a console can show elapsed time without subtracting `started_unix_ms`
/// from its own (possibly skewed) wall clock.
pub elapsed_ms: u64,
/// Path of the in-progress capture (`""` if idle). /// Path of the in-progress capture (`""` if idle).
pub kind: String, pub kind: String,
} }
@@ -376,12 +380,14 @@ fn status_of(live: Option<&Live>) -> StatsStatus {
armed: true, armed: true,
sample_count: l.samples.len() as u32, sample_count: l.samples.len() as u32,
started_unix_ms: l.started_unix_ms, started_unix_ms: l.started_unix_ms,
elapsed_ms: l.started.elapsed().as_millis() as u64,
kind: l.meta.as_ref().map(|m| m.kind.clone()).unwrap_or_default(), kind: l.meta.as_ref().map(|m| m.kind.clone()).unwrap_or_default(),
}, },
None => StatsStatus { None => StatsStatus {
armed: false, armed: false,
sample_count: 0, sample_count: 0,
started_unix_ms: 0, started_unix_ms: 0,
elapsed_ms: 0,
kind: String::new(), kind: String::new(),
}, },
} }