From 2688b3260e0452bc77d96520fb53f7129db0130e Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 19 Oct 2026 13:16:16 -0400 Subject: [PATCH 1/6] feat(dgw): let a recorder push its session events log A recorder can now push a session events log (`.slog`) next to its recording stream, over the same push WebSocket: /jet/jrec/push/{id}?fileType=slog&artifact=events The log is listed under `artifacts.events` in the manifest, not in `files`, so it stays out of the clip list, the shadow stream and the recording policy. A reconnecting recorder appends to the same log. The session stays active (token reuse, delete conflict) while either push is connected. Co-Authored-By: Claude Opus 5.5 (1M context) --- devolutions-gateway/src/api/jrec.rs | 24 ++- devolutions-gateway/src/artifacts.rs | 25 ++- devolutions-gateway/src/recording.rs | 274 +++++++++++++++++++++++++-- 3 files changed, 298 insertions(+), 25 deletions(-) diff --git a/devolutions-gateway/src/api/jrec.rs b/devolutions-gateway/src/api/jrec.rs index 508d32ff5..48f9a3776 100644 --- a/devolutions-gateway/src/api/jrec.rs +++ b/devolutions-gateway/src/api/jrec.rs @@ -31,7 +31,7 @@ use crate::api::heartbeat::recording_storage_health; use crate::artifacts::JrecArtifacts; use crate::extract::{JrecToken, RecordingDeleteScope, RecordingsReadScope}; use crate::http::{HttpError, HttpErrorBuilder}; -use crate::recording::{PushOutcome, RecordingMessageSender}; +use crate::recording::{PushOutcome, PushTarget, RecordingMessageSender}; use crate::token::{JrecTokenClaims, RecordingFileType, RecordingOperation}; /// Read chunk size when streaming a finished session ZIP from the temp file. @@ -66,6 +66,14 @@ pub fn make_router(state: DgwState) -> Router { #[serde(rename_all = "camelCase")] struct JrecPushQueryParam { file_type: RecordingFileType, + artifact: Option, +} + +/// An artifact a recorder pushes next to its recording, instead of a recording clip. +#[derive(Deserialize)] +#[serde(rename_all = "lowercase")] +enum PushArtifact { + Events, } #[derive(Deserialize)] @@ -92,6 +100,14 @@ async fn jrec_push( return Err(HttpError::forbidden().msg("expected push operation")); } + let target = match (query.artifact, query.file_type) { + (None, file_type) => PushTarget::Clip(file_type), + (Some(PushArtifact::Events), RecordingFileType::SessionRecordingLog) => PushTarget::Events, + (Some(PushArtifact::Events), _) => { + return Err(HttpError::bad_request().msg("the events artifact is a session recording log")); + } + }; + let conf = conf_handle.get_conf(); // Pre-flight: refuse the upgrade up-front when the recording storage cannot accept @@ -138,7 +154,7 @@ async fn jrec_push( recordings, shutdown_signal, claims, - query.file_type, + target, session_id, source_addr, Duration::from_secs(conf_handle.get_conf().debug.ws_keep_alive_interval), @@ -154,7 +170,7 @@ async fn handle_jrec_push( recordings: RecordingMessageSender, shutdown_signal: ShutdownSignal, claims: JrecTokenClaims, - file_type: RecordingFileType, + target: PushTarget, session_id: Uuid, source_addr: SocketAddr, keep_alive_interval: Duration, @@ -169,7 +185,7 @@ async fn handle_jrec_push( .client_stream(stream) .recordings(recordings) .claims(claims) - .file_type(file_type) + .target(target) .session_id(session_id) .shutdown_signal(shutdown_signal) .build() diff --git a/devolutions-gateway/src/artifacts.rs b/devolutions-gateway/src/artifacts.rs index a2ecd5fa8..e46c7a548 100644 --- a/devolutions-gateway/src/artifacts.rs +++ b/devolutions-gateway/src/artifacts.rs @@ -9,22 +9,25 @@ use crate::token::RecordingFileType; pub(crate) struct JrecArtifacts { #[serde(default, skip_serializing_if = "Vec::is_empty")] ai_analysis: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + events: Vec, } impl JrecArtifacts { pub(crate) fn is_empty(&self) -> bool { - let Self { ai_analysis } = self; - ai_analysis.is_empty() + let Self { ai_analysis, events } = self; + ai_analysis.is_empty() && events.is_empty() } pub(crate) fn into_file_names(self) -> impl IntoIterator { - let Self { ai_analysis } = self; - ai_analysis.into_iter().map(|artifact| artifact.file_name) + let Self { ai_analysis, events } = self; + ai_analysis.into_iter().chain(events).map(|artifact| artifact.file_name) } pub(crate) fn of_kind_mut(&mut self, kind: ArtifactKind) -> &mut Vec { match kind { ArtifactKind::AiAnalysis => &mut self.ai_analysis, + ArtifactKind::Events => &mut self.events, } } } @@ -39,18 +42,21 @@ pub(crate) struct JrecArtifact { #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub enum ArtifactKind { AiAnalysis, + /// The session events log a recorder pushes next to its recording. + Events, } impl ArtifactKind { pub const fn as_str(self) -> &'static str { match self { ArtifactKind::AiAnalysis => "ai-analysis", + ArtifactKind::Events => "events", } } pub(crate) const fn file_type(self) -> RecordingFileType { match self { - ArtifactKind::AiAnalysis => RecordingFileType::SessionRecordingLog, + ArtifactKind::AiAnalysis | ArtifactKind::Events => RecordingFileType::SessionRecordingLog, } } @@ -62,11 +68,14 @@ impl ArtifactKind { #[cfg(test)] mod tests { + use rstest::rstest; + use super::*; - #[test] - fn manifest_key_is_the_kind_name() { - let kind = ArtifactKind::AiAnalysis; + #[rstest] + #[case::ai_analysis(ArtifactKind::AiAnalysis)] + #[case::events(ArtifactKind::Events)] + fn manifest_key_is_the_kind_name(#[case] kind: ArtifactKind) { let mut artifacts = JrecArtifacts::default(); artifacts.of_kind_mut(kind).push(JrecArtifact { file_name: "file".to_owned(), diff --git a/devolutions-gateway/src/recording.rs b/devolutions-gateway/src/recording.rs index 8ece9b9b6..6943054e5 100644 --- a/devolutions-gateway/src/recording.rs +++ b/devolutions-gateway/src/recording.rs @@ -79,11 +79,20 @@ pub struct ClientPush { recordings: RecordingMessageSender, claims: JrecTokenClaims, client_stream: S, - file_type: RecordingFileType, + target: PushTarget, session_id: Uuid, shutdown_signal: ShutdownSignal, } +/// What a push writes. +#[derive(Debug, Clone, Copy)] +pub enum PushTarget { + /// A clip of the recording, listed in `files`. + Clip(RecordingFileType), + /// The session events log, an artifact the recorder pushes next to its recording. + Events, +} + impl ClientPush where S: AsyncRead + AsyncWrite + Unpin + Send + 'static, @@ -93,7 +102,7 @@ where recordings, claims, mut client_stream, - file_type, + target, session_id, mut shutdown_signal, } = self; @@ -109,7 +118,15 @@ where } }; - let (recording_file, progress) = match recordings.connect(session_id, file_type, disconnected_ttl).await { + let connected = match target { + PushTarget::Clip(file_type) => recordings + .connect(session_id, file_type, disconnected_ttl) + .await + .map(|(path, progress)| (path, Some(progress))), + PushTarget::Events => recordings.connect_events(session_id).await.map(|path| (path, None)), + }; + + let (recording_file, progress) = match connected { Ok(connected) => connected, Err(e) => { warn!(error = format!("{e:#}"), "Unable to start recording"); @@ -122,7 +139,13 @@ where let mut open_options = fs::OpenOptions::new(); - open_options.read(false).write(true).truncate(true).create(true); + open_options.read(false).write(true).create(true); + + // A reconnecting recorder continues the session's single events log. + match target { + PushTarget::Clip(_) => open_options.truncate(true), + PushTarget::Events => open_options.append(true), + }; #[cfg(windows)] { @@ -135,7 +158,9 @@ where let res = match open_options.open(&recording_file).await { Ok(file) => { - progress.mark_recording(); + if let Some(progress) = &progress { + progress.mark_recording(); + } // Wrap WriteProgressWriter inside a BufWriter to reduce the number of flushes. let (file, flush_signal) = WriteProgressWriter::new(file); // larger buffer size to reduce the number of flushes @@ -148,7 +173,9 @@ where loop { tokio::select! { _ = flush_signal.notified() => { - progress.notify_data(); + if let Some(progress) = &progress { + progress.notify_data(); + } }, _ = shutdown_signal_clone.wait() => { break; @@ -179,7 +206,7 @@ where let _ = signal_loop.await; let flush_result = file.flush().await; - if flush_result.is_ok() { + if let (Ok(()), Some(progress)) = (&flush_result, &progress) { progress.notify_data(); } @@ -198,7 +225,13 @@ where info!(?res, "Recording finished"); - recordings.disconnect(session_id).await.context("disconnect")?; + match target { + PushTarget::Clip(_) => recordings.disconnect(session_id).await.context("disconnect")?, + PushTarget::Events => recordings + .disconnect_events(session_id) + .await + .context("disconnect events")?, + } res } @@ -360,6 +393,13 @@ enum RecordingManagerMessage { Disconnect { id: Uuid, }, + ConnectEvents { + id: Uuid, + channel: oneshot::Sender>, + }, + DisconnectEvents { + id: Uuid, + }, GetState { id: Uuid, channel: oneshot::Sender>, @@ -397,6 +437,12 @@ impl fmt::Debug for RecordingManagerMessage { .field("kind", kind) .finish_non_exhaustive(), RecordingManagerMessage::Disconnect { id } => f.debug_struct("Disconnect").field("id", id).finish(), + RecordingManagerMessage::ConnectEvents { id, channel: _ } => { + f.debug_struct("ConnectEvents").field("id", id).finish_non_exhaustive() + } + RecordingManagerMessage::DisconnectEvents { id } => { + f.debug_struct("DisconnectEvents").field("id", id).finish() + } RecordingManagerMessage::GetState { id, channel: _ } => { f.debug_struct("GetState").field("id", id).finish_non_exhaustive() } @@ -467,6 +513,25 @@ impl RecordingMessageSender { .context("couldn't send Remove message") } + /// Returns the session events log to append to. + async fn connect_events(&self, id: Uuid) -> anyhow::Result { + let (tx, rx) = oneshot::channel(); + self.channel + .send(RecordingManagerMessage::ConnectEvents { id, channel: tx }) + .await + .ok() + .context("couldn't send ConnectEvents message")?; + rx.await.context("couldn't receive ConnectEvents result")? + } + + async fn disconnect_events(&self, id: Uuid) -> anyhow::Result<()> { + self.channel + .send(RecordingManagerMessage::DisconnectEvents { id }) + .await + .ok() + .context("couldn't send DisconnectEvents message") + } + pub async fn get_state(&self, id: Uuid) -> anyhow::Result> { let (tx, rx) = oneshot::channel(); self.channel @@ -573,6 +638,8 @@ impl Ord for DisconnectedTtl { pub struct RecordingManagerTask { rx: RecordingMessageReceiver, ongoing_recordings: HashMap, + /// Sessions whose recorder is pushing its events log. + events_pushes: HashSet, recordings_path: Utf8PathBuf, session_manager_handle: SessionMessageSender, job_queue_handle: JobQueueHandle, @@ -588,6 +655,7 @@ impl RecordingManagerTask { Self { rx, ongoing_recordings: HashMap::new(), + events_pushes: HashSet::new(), recordings_path, session_manager_handle, job_queue_handle, @@ -819,6 +887,60 @@ impl RecordingManagerTask { Ok(artifact_path) } + async fn handle_connect_events(&mut self, id: Uuid) -> anyhow::Result { + if self.events_pushes.contains(&id) { + anyhow::bail!("concurrent events push for the same session is not supported"); + } + + let recording_path = self.recordings_path.join(id.to_string()); + let manifest_path = recording_path.join("recording.json"); + + // The events log can start before the recording it accompanies. + let mut manifest = if manifest_path.exists() { + JrecManifest::read_from_file(&manifest_path).context("read manifest from disk")? + } else { + fs::create_dir_all(&recording_path) + .await + .with_context(|| format!("failed to create recording path: {recording_path}"))?; + + JrecManifest { + session_id: id, + start_time: time::OffsetDateTime::now_utc().unix_timestamp(), + duration: 0, + files: Vec::new(), + artifacts: JrecArtifacts::default(), + } + }; + + let events = manifest.artifacts.of_kind_mut(ArtifactKind::Events); + let file_name = match events.last() { + Some(log) => log.file_name.clone(), + None => { + let file_name = ArtifactKind::Events.file_name(0); + events.push(JrecArtifact { + file_name: file_name.clone(), + }); + manifest + .save_to_file(&manifest_path) + .context("write manifest to disk")?; + file_name + } + }; + + self.events_pushes.insert(id); + self.rx.active_recordings.insert(id); + + Ok(recording_path.join(file_name)) + } + + fn handle_disconnect_events(&mut self, id: Uuid) { + self.events_pushes.remove(&id); + + if !self.ongoing_recordings.contains_key(&id) { + self.rx.active_recordings.remove(id); + } + } + fn handle_remove(&mut self, id: Uuid) { if let Some(ongoing) = self.ongoing_recordings.get(&id) { match ongoing.state { @@ -826,7 +948,9 @@ impl RecordingManagerTask { if reconnect_deadline <= tokio::time::Instant::now() => { debug!(%id, "Mark recording as terminated"); - self.rx.active_recordings.remove(id); + if !self.events_pushes.contains(&id) { + self.rx.active_recordings.remove(id); + } ongoing.stream_state.send_modify(RecordingStreamState::mark_ended); // Check the recording policy of the associated session and kill it if necessary. @@ -960,6 +1084,12 @@ async fn recording_manager_task( } } } + RecordingManagerMessage::ConnectEvents { id, channel } => { + let _ = channel.send(manager.handle_connect_events(id).await); + } + RecordingManagerMessage::DisconnectEvents { id } => { + manager.handle_disconnect_events(id); + } RecordingManagerMessage::GetState { id, channel } => { let response = manager.ongoing_recordings.get(&id).map(|ongoing| ongoing.state.clone()); let _ = channel.send(response); @@ -1008,11 +1138,15 @@ async fn recording_manager_task( }; debug!(?msg, "Received message"); - if let RecordingManagerMessage::Disconnect { id } = msg { - if let Err(e) = manager.handle_disconnect(id).await { - error!(error = format!("{e:#}"), "handle_disconnect"); + match msg { + RecordingManagerMessage::Disconnect { id } => { + if let Err(e) = manager.handle_disconnect(id).await { + error!(error = format!("{e:#}"), "handle_disconnect"); + } + manager.ongoing_recordings.remove(&id); } - manager.ongoing_recordings.remove(&id); + RecordingManagerMessage::DisconnectEvents { id } => manager.handle_disconnect_events(id), + _ => {} } } @@ -1179,6 +1313,21 @@ mod tests { fn read_file(&self, id: Uuid, file_name: &str) -> String { std::fs::read_to_string(self.recordings_path.join(id.to_string()).join(file_name)).expect("read file") } + + async fn try_connect_events(&self, id: Uuid) -> anyhow::Result { + let path = self.sender.connect_events(id).await?; + Ok(path.file_name().expect("file name").to_owned()) + } + + async fn connect_events(&self, id: Uuid) -> String { + self.try_connect_events(id).await.expect("connect events") + } + + async fn disconnect_events(&self, id: Uuid) { + self.sender.disconnect_events(id).await.expect("disconnect events"); + // Messages are processed in order, so this waits for the disconnect to be handled. + self.sender.get_count().await.expect("sync with manager"); + } } #[tokio::test] @@ -1421,6 +1570,105 @@ mod tests { assert_eq!(harness.sender.get_count().await.expect("count"), 1); } + #[tokio::test] + async fn events_log_goes_to_artifacts() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + let clip = harness.connect(id, WEBM).await; + let log = harness.connect_events(id).await; + + assert_eq!(clip, "recording-0.webm"); + assert_eq!(log, "events-0.slog"); + let manifest = harness.read_manifest(id); + let files = manifest["files"].as_array().expect("files"); + assert_eq!(files.len(), 1); + assert_eq!(files[0]["fileName"], "recording-0.webm"); + assert_eq!( + manifest["artifacts"], + json!({ "events": [{ "fileName": "events-0.slog" }] }) + ); + } + + #[tokio::test] + async fn events_log_can_start_before_the_recording() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + let log = harness.connect_events(id).await; + let clip = harness.connect(id, WEBM).await; + + assert_eq!(log, "events-0.slog"); + assert_eq!(clip, "recording-0.webm"); + let manifest = harness.read_manifest(id); + assert_eq!(manifest["files"][0]["fileName"], "recording-0.webm"); + assert_eq!( + manifest["artifacts"], + json!({ "events": [{ "fileName": "events-0.slog" }] }) + ); + } + + #[tokio::test] + async fn a_second_events_push_is_refused() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + harness.connect_events(id).await; + + assert!(harness.try_connect_events(id).await.is_err()); + } + + #[tokio::test] + async fn a_reconnecting_events_push_continues_the_same_log() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + let first = harness.connect_events(id).await; + harness.disconnect_events(id).await; + let second = harness.connect_events(id).await; + + assert_eq!(first, second); + assert_eq!( + harness.read_manifest(id)["artifacts"], + json!({ "events": [{ "fileName": "events-0.slog" }] }) + ); + } + + #[tokio::test] + async fn an_events_push_keeps_the_session_active() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + harness.connect_events(id).await; + assert!(harness.sender.active_recordings.contains(id)); + assert!(harness.sender.get_state(id).await.expect("get state").is_none()); + assert_eq!(harness.sender.get_count().await.expect("count"), 0); + + harness.connect(id, WEBM).await; + harness.disconnect_events(id).await; + assert!(harness.sender.active_recordings.contains(id)); + + harness.disconnect(id).await; + assert!(harness.sender.active_recordings.contains(id)); + } + + #[tokio::test] + async fn the_session_stays_active_until_its_events_push_ends() { + let harness = Harness::start(); + let id = Uuid::new_v4(); + + harness.connect_events(id).await; + harness.sender.connect(id, WEBM, Duration::ZERO).await.expect("connect"); + harness.disconnect(id).await; + while harness.sender.get_state(id).await.expect("get state").is_some() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(harness.sender.active_recordings.contains(id)); + + harness.disconnect_events(id).await; + assert!(!harness.sender.active_recordings.contains(id)); + } + #[test] fn manifest_without_artifacts_round_trips_byte_for_byte() { let dir = tempfile::tempdir().expect("temp dir"); From 755eddb0fde08ef67cc562cbb1daceea89c6da56 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 19 Oct 2026 14:19:08 -0400 Subject: [PATCH 2/6] fix(dgw): start the session with its first clip when the events log came first An events push that connects before any clip creates the manifest, with its own connect time as the session start. The first clip now resets that start, so `startTime` and `duration` describe the recording, not the wait for it. Co-Authored-By: Claude Opus 5.5 (1M context) --- devolutions-gateway/src/recording.rs | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/devolutions-gateway/src/recording.rs b/devolutions-gateway/src/recording.rs index 6943054e5..256611cee 100644 --- a/devolutions-gateway/src/recording.rs +++ b/devolutions-gateway/src/recording.rs @@ -688,6 +688,11 @@ impl RecordingManagerTask { let start_time = time::OffsetDateTime::now_utc().unix_timestamp(); + // An events log pushed first created the manifest, but the session starts with its first clip. + if existing_manifest.files.is_empty() { + existing_manifest.start_time = start_time; + } + let file_name = format!("recording-{next_file_idx}.{}", file_type.extension()); let recording_file = recording_path.join(&file_name); @@ -1596,12 +1601,19 @@ mod tests { let id = Uuid::new_v4(); let log = harness.connect_events(id).await; + let mut manifest = harness.read_manifest(id); + manifest["startTime"] = json!(0); + harness.write_manifest(id, &manifest.to_string()); let clip = harness.connect(id, WEBM).await; assert_eq!(log, "events-0.slog"); assert_eq!(clip, "recording-0.webm"); let manifest = harness.read_manifest(id); assert_eq!(manifest["files"][0]["fileName"], "recording-0.webm"); + assert_eq!( + manifest["startTime"], manifest["files"][0]["startTime"], + "the session starts with its first clip" + ); assert_eq!( manifest["artifacts"], json!({ "events": [{ "fileName": "events-0.slog" }] }) From c89b378bad0204c0e4e3fc9de66cfd0aba656f8e Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 19 Oct 2026 15:29:39 -0400 Subject: [PATCH 3/6] docs(dgw): state the events log push in the recording intent Co-Authored-By: Claude Opus 5.5 (1M context) --- devolutions-gateway/src/recording.intent.md | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/devolutions-gateway/src/recording.intent.md b/devolutions-gateway/src/recording.intent.md index 14bb0371e..721a777ed 100644 --- a/devolutions-gateway/src/recording.intent.md +++ b/devolutions-gateway/src/recording.intent.md @@ -1,10 +1,15 @@ # Recording -Gateway recording is a continuous stream of bytes pushed from the client who pocesses the valid recording token. +Gateway recording is a continuous stream of bytes pushed from the client who possesses the valid recording token. The url is `/jet/jrec/push/{sessionId}?fileType={fileType}`. We expect the `fileType` to be one of the following file types: `webm`, `cast`, `trp` and `slog`, which must be specified. When connection is established with request of recordings for the session, if recording is not enabled within a short period of time, the connection will be closed with indication of violation of the recording policy. +Gateway only allows one recording to be pushed at a time per session. # Artifacts -Artifacts are files that are not recordings, currently only have `ai-analysis` with combination to `slog` file type. -Currently, we do not have an api endpoint to push artifacts, but we will have one in the future. \ No newline at end of file +Artifacts are files that are not recordings, listed under `artifacts` in manifest instead of `files`. +The url is `/jet/jrec/push/{sessionId}?fileType=slog&artifact=events`; `events` is the only artifact that can be pushed, as the session events log. +`ai-analysis` is not pushed: it is only kept so older manifests still list it. +Artifacts of file type slog can be pushed by producers alongside the recordings, one artifact at a time. +Pushing an artifact does not satisfy the recording policy. + From 6cd32abdfd31a30d68ef162014ef0bf59f080045 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 19 Oct 2026 16:29:13 -0400 Subject: [PATCH 4/6] fix(dgw): keep the events push out of the recording lifecycle An open events push kept the session in the active recordings, so the push token stayed reusable for new clips past its reconnect window, until it expired. The events push now leaves the recording state alone. Co-Authored-By: Claude Opus 5.5 (1M context) --- devolutions-gateway/src/recording.rs | 31 ++++------------------------ 1 file changed, 4 insertions(+), 27 deletions(-) diff --git a/devolutions-gateway/src/recording.rs b/devolutions-gateway/src/recording.rs index 256611cee..c21495d03 100644 --- a/devolutions-gateway/src/recording.rs +++ b/devolutions-gateway/src/recording.rs @@ -933,17 +933,12 @@ impl RecordingManagerTask { }; self.events_pushes.insert(id); - self.rx.active_recordings.insert(id); Ok(recording_path.join(file_name)) } fn handle_disconnect_events(&mut self, id: Uuid) { self.events_pushes.remove(&id); - - if !self.ongoing_recordings.contains_key(&id) { - self.rx.active_recordings.remove(id); - } } fn handle_remove(&mut self, id: Uuid) { @@ -953,9 +948,7 @@ impl RecordingManagerTask { if reconnect_deadline <= tokio::time::Instant::now() => { debug!(%id, "Mark recording as terminated"); - if !self.events_pushes.contains(&id) { - self.rx.active_recordings.remove(id); - } + self.rx.active_recordings.remove(id); ongoing.stream_state.send_modify(RecordingStreamState::mark_ended); // Check the recording policy of the associated session and kill it if necessary. @@ -1647,37 +1640,21 @@ mod tests { } #[tokio::test] - async fn an_events_push_keeps_the_session_active() { + async fn an_events_push_never_keeps_the_recording_alive() { let harness = Harness::start(); let id = Uuid::new_v4(); harness.connect_events(id).await; - assert!(harness.sender.active_recordings.contains(id)); + assert!(!harness.sender.active_recordings.contains(id)); assert!(harness.sender.get_state(id).await.expect("get state").is_none()); assert_eq!(harness.sender.get_count().await.expect("count"), 0); - harness.connect(id, WEBM).await; - harness.disconnect_events(id).await; - assert!(harness.sender.active_recordings.contains(id)); - - harness.disconnect(id).await; - assert!(harness.sender.active_recordings.contains(id)); - } - - #[tokio::test] - async fn the_session_stays_active_until_its_events_push_ends() { - let harness = Harness::start(); - let id = Uuid::new_v4(); - - harness.connect_events(id).await; harness.sender.connect(id, WEBM, Duration::ZERO).await.expect("connect"); + assert!(harness.sender.active_recordings.contains(id)); harness.disconnect(id).await; while harness.sender.get_state(id).await.expect("get state").is_some() { tokio::time::sleep(Duration::from_millis(10)).await; } - assert!(harness.sender.active_recordings.contains(id)); - - harness.disconnect_events(id).await; assert!(!harness.sender.active_recordings.contains(id)); } From 932731e63299fece5d6e0c72947b0519555b7edc Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 19 Oct 2026 16:53:14 -0400 Subject: [PATCH 5/6] fix(dgw): refuse an events push before the recording starts The events push created the session folder and manifest when it came before the recording, and the first clip then had to reset the session start for it. The folder and manifest belong to the recording: an events push now needs the recording of the session to have started, and is refused before. The events push has its own push token, separate from the recording's, so an events push refused this way never touches the recording's token. Co-Authored-By: Claude Opus 5.5 (1M context) --- devolutions-gateway/src/recording.rs | 57 +++++++--------------------- 1 file changed, 13 insertions(+), 44 deletions(-) diff --git a/devolutions-gateway/src/recording.rs b/devolutions-gateway/src/recording.rs index c21495d03..5a6477c07 100644 --- a/devolutions-gateway/src/recording.rs +++ b/devolutions-gateway/src/recording.rs @@ -688,11 +688,6 @@ impl RecordingManagerTask { let start_time = time::OffsetDateTime::now_utc().unix_timestamp(); - // An events log pushed first created the manifest, but the session starts with its first clip. - if existing_manifest.files.is_empty() { - existing_manifest.start_time = start_time; - } - let file_name = format!("recording-{next_file_idx}.{}", file_type.extension()); let recording_file = recording_path.join(&file_name); @@ -900,22 +895,12 @@ impl RecordingManagerTask { let recording_path = self.recordings_path.join(id.to_string()); let manifest_path = recording_path.join("recording.json"); - // The events log can start before the recording it accompanies. - let mut manifest = if manifest_path.exists() { - JrecManifest::read_from_file(&manifest_path).context("read manifest from disk")? - } else { - fs::create_dir_all(&recording_path) - .await - .with_context(|| format!("failed to create recording path: {recording_path}"))?; + // The session folder and manifest belong to the recording, so the events push never creates them. + if !manifest_path.exists() { + anyhow::bail!("an events push needs the recording of the session to start first"); + } - JrecManifest { - session_id: id, - start_time: time::OffsetDateTime::now_utc().unix_timestamp(), - duration: 0, - files: Vec::new(), - artifacts: JrecArtifacts::default(), - } - }; + let mut manifest = JrecManifest::read_from_file(&manifest_path).context("read manifest from disk")?; let events = manifest.artifacts.of_kind_mut(ArtifactKind::Events); let file_name = match events.last() { @@ -1589,28 +1574,12 @@ mod tests { } #[tokio::test] - async fn events_log_can_start_before_the_recording() { + async fn an_events_push_before_the_recording_is_refused() { let harness = Harness::start(); let id = Uuid::new_v4(); - let log = harness.connect_events(id).await; - let mut manifest = harness.read_manifest(id); - manifest["startTime"] = json!(0); - harness.write_manifest(id, &manifest.to_string()); - let clip = harness.connect(id, WEBM).await; - - assert_eq!(log, "events-0.slog"); - assert_eq!(clip, "recording-0.webm"); - let manifest = harness.read_manifest(id); - assert_eq!(manifest["files"][0]["fileName"], "recording-0.webm"); - assert_eq!( - manifest["startTime"], manifest["files"][0]["startTime"], - "the session starts with its first clip" - ); - assert_eq!( - manifest["artifacts"], - json!({ "events": [{ "fileName": "events-0.slog" }] }) - ); + assert!(harness.try_connect_events(id).await.is_err()); + assert!(!harness.recordings_path.join(id.to_string()).exists()); } #[tokio::test] @@ -1618,6 +1587,7 @@ mod tests { let harness = Harness::start(); let id = Uuid::new_v4(); + harness.connect(id, WEBM).await; harness.connect_events(id).await; assert!(harness.try_connect_events(id).await.is_err()); @@ -1628,6 +1598,7 @@ mod tests { let harness = Harness::start(); let id = Uuid::new_v4(); + harness.connect(id, WEBM).await; let first = harness.connect_events(id).await; harness.disconnect_events(id).await; let second = harness.connect_events(id).await; @@ -1644,13 +1615,11 @@ mod tests { let harness = Harness::start(); let id = Uuid::new_v4(); - harness.connect_events(id).await; - assert!(!harness.sender.active_recordings.contains(id)); - assert!(harness.sender.get_state(id).await.expect("get state").is_none()); - assert_eq!(harness.sender.get_count().await.expect("count"), 0); - harness.sender.connect(id, WEBM, Duration::ZERO).await.expect("connect"); + harness.connect_events(id).await; assert!(harness.sender.active_recordings.contains(id)); + assert_eq!(harness.sender.get_count().await.expect("count"), 1); + harness.disconnect(id).await; while harness.sender.get_state(id).await.expect("get state").is_some() { tokio::time::sleep(Duration::from_millis(10)).await; From 3c50ce2c70031b0c402dcc0032f134d008b61a77 Mon Sep 17 00:00:00 2001 From: Junyi Ou Date: Mon, 19 Oct 2026 17:08:55 -0400 Subject: [PATCH 6/6] fix(dgw): name the push target in its logs and mark the events append untested - The push logs say whether a clip or the events log started or ended. - The append of a reconnecting events push has no test; an `UNTESTED:` comment keeps it from turning into a truncate. - `handle_connect_events` no longer awaits anything, so it is not async. - The test waiting for a recording to end gives up after 5 s instead of hanging. Co-Authored-By: Claude Opus 5.5 (1M context) --- devolutions-gateway/src/recording.rs | 22 ++++++++++++++-------- 1 file changed, 14 insertions(+), 8 deletions(-) diff --git a/devolutions-gateway/src/recording.rs b/devolutions-gateway/src/recording.rs index 5a6477c07..f806d9984 100644 --- a/devolutions-gateway/src/recording.rs +++ b/devolutions-gateway/src/recording.rs @@ -129,7 +129,7 @@ where let (recording_file, progress) = match connected { Ok(connected) => connected, Err(e) => { - warn!(error = format!("{e:#}"), "Unable to start recording"); + warn!(error = format!("{e:#}"), ?target, "Unable to start recording"); client_stream.shutdown().await.context("shutdown")?; return Ok(PushOutcome::Done); } @@ -141,7 +141,9 @@ where open_options.read(false).write(true).create(true); - // A reconnecting recorder continues the session's single events log. + // UNTESTED: a reconnecting events push appends to the session's single events log. + // Truncating it would drop every line the recorder pushed before the reconnect. + // When changing these options, keep `append` for `PushTarget::Events` and `truncate` for clips. match target { PushTarget::Clip(_) => open_options.truncate(true), PushTarget::Events => open_options.append(true), @@ -223,7 +225,7 @@ where Err(e) => Err(anyhow::Error::new(e).context(format!("failed to open file at {recording_file}"))), }; - info!(?res, "Recording finished"); + info!(?target, ?res, "Recording finished"); match target { PushTarget::Clip(_) => recordings.disconnect(session_id).await.context("disconnect")?, @@ -887,7 +889,7 @@ impl RecordingManagerTask { Ok(artifact_path) } - async fn handle_connect_events(&mut self, id: Uuid) -> anyhow::Result { + fn handle_connect_events(&mut self, id: Uuid) -> anyhow::Result { if self.events_pushes.contains(&id) { anyhow::bail!("concurrent events push for the same session is not supported"); } @@ -1068,7 +1070,7 @@ async fn recording_manager_task( } } RecordingManagerMessage::ConnectEvents { id, channel } => { - let _ = channel.send(manager.handle_connect_events(id).await); + let _ = channel.send(manager.handle_connect_events(id)); } RecordingManagerMessage::DisconnectEvents { id } => { manager.handle_disconnect_events(id); @@ -1621,9 +1623,13 @@ mod tests { assert_eq!(harness.sender.get_count().await.expect("count"), 1); harness.disconnect(id).await; - while harness.sender.get_state(id).await.expect("get state").is_some() { - tokio::time::sleep(Duration::from_millis(10)).await; - } + tokio::time::timeout(Duration::from_secs(5), async { + while harness.sender.get_state(id).await.expect("get state").is_some() { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("the recording ends"); assert!(!harness.sender.active_recordings.contains(id)); }