diff --git a/README.md b/README.md index 654556ba..57fafd6a 100644 --- a/README.md +++ b/README.md @@ -208,6 +208,9 @@ For wire protocol details, frame encoding, and transport internals, see [ARCHITE | `BLIT_TARGET` | unset | Default remote: a URI or named remote (overrides `target` in `blit.conf`) | | `BLIT_REMOTES` | `~/.config/blit/blit.remotes` | Gateway remotes file path (overrides default location) | | `BLIT_SCROLLBACK` | `10000` | Scrollback rows per PTY | +| `BLIT_EVENTS_SIZE` | `1048576` | Binary server event-ring capacity | +| `BLIT_EVENTS` | low-throughput lifecycle events | Fine-grained event names/selectors (`default,+frame.*,+pty.*`, or `all`) | +| `BLIT_EVENTS_FILE` | unset | Stream binary events to this server-side file from startup | | `BLIT_HUB` | `hub.blit.sh` | Signaling hub URL for WebRTC sharing. On `blit gateway`, sets the default hub for `share:` remotes when `BLIT_GATEWAY_WEBRTC=1`. | | `BLIT_GATEWAY_WEBRTC` | unset | Set to `1` on `blit gateway` to proxy `share:` remotes via WebRTC. The gateway connects as a WebRTC consumer and bridges terminals to browsers over WebSocket/WebTransport. Without this, `share:` entries in `blit.remotes` are ignored. | | `BLIT_PASSPHRASE` | unset | Browser passphrase for `blit gateway`; may be plaintext or an argon2 PHC hash generated by `blit hash-passphrase`. Browsers still enter the plaintext passphrase. | diff --git a/crates/cli/src/agent.rs b/crates/cli/src/agent.rs index fc0f67e6..3fbf1f1a 100644 --- a/crates/cli/src/agent.rs +++ b/crates/cli/src/agent.rs @@ -218,6 +218,16 @@ impl AgentConn { Ok(data) } + pub(crate) async fn recv_unbounded(&mut self) -> Result, String> { + let data = read_message(&mut self.reader, &mut self.fragment_buf) + .await + .ok_or_else(|| "server closed connection".to_string())?; + if let Some(error) = self.note_kick(&data) { + return Err(error); + } + Ok(data) + } + pub(crate) fn has_pty(&self, id: u16) -> bool { self.ptys.iter().any(|p| p.id == id) } diff --git a/crates/cli/src/cli.rs b/crates/cli/src/cli.rs index f29c579c..c0c9aa17 100644 --- a/crates/cli/src/cli.rs +++ b/crates/cli/src/cli.rs @@ -323,6 +323,12 @@ pub enum Command { command: Option, }, + /// Inspect and configure the server's binary event journal + Events { + #[command(subcommand)] + command: EventsCommand, + }, + /// Manage compositor surfaces #[command(alias = "s")] Surface { @@ -671,6 +677,93 @@ pub enum Command { External(Vec), } +// ── Event journal subcommands ──────────────────────────────────────────── + +#[derive(Subcommand)] +pub enum EventsCommand { + /// Show config revision, ring use/loss counters, and active event types + Config, + + /// Replace ring size and/or activation bitset + Set { + /// Ring capacity in bytes (4 KiB through just under 64 MiB) + #[arg(long)] + size: Option, + + /// Activation selectors: default, all, none, exact names, category.*, + /// with optional + or - prefixes + #[arg(long, value_name = "SPEC")] + events: Option, + + /// Apply only while the server configuration is still at this revision + #[arg(long, value_name = "REVISION")] + if_revision: Option, + }, + + /// Write one retained binary snapshot to stdout or a local file + Dump { + /// Local output path; defaults to stdout. Use - for stdout explicitly. + #[arg(short, long, value_name = "LOCAL_PATH")] + output: Option, + }, + + /// Follow events in the foreground, writing stdout or a local file + /// + /// The stream ends with this client connection. A connection may own at + /// most four concurrent event tails. + Tail { + /// Local output path; defaults to stdout. Use - for stdout explicitly. + #[arg(short, long, value_name = "LOCAL_PATH")] + output: Option, + + /// Append to --output instead of truncating it + #[arg(long, requires = "output")] + append: bool, + + /// Follow only events produced after this command starts + #[arg(long)] + from_now: bool, + }, + + /// Manage detached recordings written by the server + Record { + #[command(subcommand)] + command: EventsRecordCommand, + }, +} + +#[derive(Subcommand)] +pub enum EventsRecordCommand { + /// Start a detached recording to a path on the server + /// + /// Returns an ID only after the initial header/history has been flushed. + /// The server permits at most eight detached recordings process-wide. + Start { + /// Destination path on the server + #[arg(value_name = "SERVER_PATH")] + path: String, + + /// Append instead of truncating the server file + #[arg(long)] + append: bool, + + /// Record only events produced after this command starts + #[arg(long)] + from_now: bool, + }, + + /// List recordings as TSV: ID, STATE, RECORDS, BYTES, LOST, HISTORY, MODE, PATH, ERROR + List, + + /// Stop and flush a server recording + /// + /// Returns an error if any recording write or final flush failed. + Stop { + /// Recording ID from `blit events record start` or `list` + id: u32, + }, +} + // ── Terminal subcommands ───────────────────────────────────────────────── #[derive(Subcommand)] diff --git a/crates/cli/src/events.rs b/crates/cli/src/events.rs new file mode 100644 index 00000000..bf6a80c9 --- /dev/null +++ b/crates/cli/src/events.rs @@ -0,0 +1,399 @@ +use std::path::Path; + +use blit_remote::events::{ + EVENT_TYPE_CATALOG, EVENTS_CONFIG_REVISION_ANY, EVENTS_STREAM_APPEND, EVENTS_STREAM_HISTORY, + EVENTS_STREAM_STATE_FAILED, EVENTS_STREAM_STATE_RUNNING, EVENTS_STREAM_STATE_STOPPED, + EVENTS_TARGET_CLIENT, EVENTS_TARGET_FILE, EventsMessage, FEATURE_EVENTS, msg_events_config_get, + msg_events_config_set, msg_events_dump, msg_events_stream_list, msg_events_stream_start, + msg_events_stream_stop, parse_activation_spec, parse_events_message, +}; +use tokio::io::{AsyncWrite, AsyncWriteExt}; + +use crate::agent::AgentConn; +use crate::cli::{EventsCommand, EventsRecordCommand}; +use crate::transport::Transport; + +const NONCE: u16 = 0xE701; + +async fn connect(transport: Transport) -> Result { + let connection = AgentConn::connect(transport).await?; + if connection.features & FEATURE_EVENTS == 0 { + return Err("server does not support blit.events.v1".into()); + } + Ok(connection) +} + +async fn get_config( + conn: &mut AgentConn, +) -> Result< + ( + u64, + u64, + u64, + u64, + u64, + u64, + blit_remote::events::ActivationSet, + ), + String, +> { + conn.send(&msg_events_config_get(NONCE)).await?; + loop { + let packet = conn.recv().await?; + let Ok(message) = parse_events_message(&packet) else { + continue; + }; + match message { + EventsMessage::Config { + nonce: NONCE, + revision, + size, + used, + records, + dropped, + next_sequence, + activations, + } => { + return Ok(( + revision, + size, + used, + records, + dropped, + next_sequence, + activations, + )); + } + EventsMessage::Result { + nonce: NONCE, + status, + detail, + .. + } => return Err(format!("{}: {detail}", blit_remote::status_text(status))), + _ => {} + } + } +} + +fn print_config( + revision: u64, + size: u64, + used: u64, + records: u64, + dropped: u64, + next_sequence: u64, + activations: blit_remote::events::ActivationSet, +) { + println!("protocol\t{}", blit_remote::events::EVENTS_PROTOCOL); + println!("revision\t{revision}"); + println!("size\t{size}"); + println!("used\t{used}"); + println!("records\t{records}"); + println!("dropped\t{dropped}"); + println!("next_sequence\t{next_sequence}"); + let enabled = EVENT_TYPE_CATALOG + .iter() + .filter_map(|&(kind, name)| activations.enabled(kind).then_some(name)) + .collect::>() + .join(","); + println!("events\t{enabled}"); + println!( + "bitset\t{}", + activations + .0 + .iter() + .map(|word| format!("{word:016x}")) + .collect::>() + .join(":") + ); +} + +async fn wait_result(conn: &mut AgentConn, started: bool) -> Result { + loop { + let packet = conn.recv().await?; + let Ok(message) = parse_events_message(&packet) else { + continue; + }; + match message { + EventsMessage::Result { + nonce: NONCE, + status, + stream_id, + detail, + } => { + if status == blit_remote::STATUS_OK { + return Ok(stream_id); + } + return Err(format!("{}: {detail}", blit_remote::status_text(status))); + } + EventsMessage::StreamStarted { + nonce: NONCE, + status, + stream_id, + detail, + } if started => { + if status == blit_remote::STATUS_OK { + return Ok(stream_id); + } + return Err(format!("{}: {detail}", blit_remote::status_text(status))); + } + _ => {} + } + } +} + +async fn open_output( + path: &str, + append: bool, +) -> Result, String> { + if path == "-" { + return Ok(Box::new(tokio::io::stdout())); + } + let mut options = tokio::fs::OpenOptions::new(); + options + .create(true) + .write(true) + .append(append) + .truncate(!append); + Ok(Box::new( + options + .open(Path::new(path)) + .await + .map_err(|error| format!("cannot open {path}: {error}"))?, + )) +} + +pub(crate) async fn run(transport: Transport, command: EventsCommand) -> Result<(), String> { + let mut conn = connect(transport).await?; + match command { + EventsCommand::Config => { + let (revision, size, used, records, dropped, next_sequence, activations) = + get_config(&mut conn).await?; + print_config( + revision, + size, + used, + records, + dropped, + next_sequence, + activations, + ); + } + EventsCommand::Set { + size, + events, + if_revision, + } => { + let (_, current_size, _, _, _, _, current_activations) = get_config(&mut conn).await?; + let size = size.unwrap_or(current_size); + let activations = events + .as_deref() + .map(parse_activation_spec) + .transpose()? + .unwrap_or(current_activations); + conn.send(&msg_events_config_set( + NONCE, + if_revision.unwrap_or(EVENTS_CONFIG_REVISION_ANY), + size, + activations, + )) + .await?; + loop { + let packet = conn.recv().await?; + let Ok(message) = parse_events_message(&packet) else { + continue; + }; + match message { + EventsMessage::Config { + nonce: NONCE, + revision, + size, + used, + records, + dropped, + next_sequence, + activations, + } => { + print_config( + revision, + size, + used, + records, + dropped, + next_sequence, + activations, + ); + break; + } + EventsMessage::Result { + nonce: NONCE, + status, + detail, + .. + } => return Err(format!("{}: {detail}", blit_remote::status_text(status))), + _ => {} + } + } + } + EventsCommand::Dump { output } => { + conn.send(&msg_events_dump(NONCE)).await?; + loop { + let packet = conn.recv().await?; + if let Ok(EventsMessage::Dump { + nonce: NONCE, + bytes, + }) = parse_events_message(&packet) + { + let path = output.as_deref().unwrap_or("-"); + let mut writer = open_output(path, false).await?; + writer + .write_all(bytes) + .await + .map_err(|error| format!("cannot write {path}: {error}"))?; + writer + .flush() + .await + .map_err(|error| format!("cannot flush {path}: {error}"))?; + break; + } + } + } + EventsCommand::Tail { + output, + append, + from_now, + } => { + let mut flags = 0; + if !from_now { + flags |= EVENTS_STREAM_HISTORY; + } + conn.send(&msg_events_stream_start( + NONCE, + EVENTS_TARGET_CLIENT, + flags, + "", + )) + .await?; + let stream_id = wait_result(&mut conn, true).await?; + let path = output.as_deref().unwrap_or("-"); + let mut writer = open_output(path, append).await?; + loop { + tokio::select! { + result = conn.recv_unbounded() => { + let packet = result?; + let Ok(message) = parse_events_message(&packet) else { + continue; + }; + match message { + EventsMessage::Dump { nonce: NONCE, bytes } => { + writer.write_all(bytes).await.map_err(|e| e.to_string())?; + } + EventsMessage::Records { stream_id: id, records, .. } + if id == stream_id => { + writer.write_all(records).await.map_err(|e| e.to_string())?; + } + EventsMessage::StreamGap { stream_id: id, lost } if id == stream_id => { + eprintln!("blit: event stream lost {lost} records"); + } + EventsMessage::StreamStopped { stream_id: id, status, detail } + if id == stream_id => { + if status != blit_remote::STATUS_OK { + return Err(format!("{}: {detail}", blit_remote::status_text(status))); + } + break; + } + _ => {} + } + } + _ = tokio::signal::ctrl_c() => { + conn.send(&msg_events_stream_stop(NONCE, stream_id)).await?; + let _ = wait_result(&mut conn, false).await?; + break; + } + } + } + writer.flush().await.map_err(|error| error.to_string())?; + } + EventsCommand::Record { command } => match command { + EventsRecordCommand::Start { + path, + append, + from_now, + } => { + let mut flags = 0; + if !from_now { + flags |= EVENTS_STREAM_HISTORY; + } + if append { + flags |= EVENTS_STREAM_APPEND; + } + conn.send(&msg_events_stream_start( + NONCE, + EVENTS_TARGET_FILE, + flags, + &path, + )) + .await?; + println!("{}", wait_result(&mut conn, true).await?); + } + EventsRecordCommand::List => { + conn.send(&msg_events_stream_list(NONCE)).await?; + loop { + let packet = conn.recv().await?; + let Ok(message) = parse_events_message(&packet) else { + continue; + }; + match message { + EventsMessage::Streams { + nonce: NONCE, + streams, + } => { + for stream in streams { + let state = match stream.state { + EVENTS_STREAM_STATE_RUNNING => "running", + EVENTS_STREAM_STATE_STOPPED => "stopped", + EVENTS_STREAM_STATE_FAILED => "failed", + _ => "unknown", + }; + let history = if stream.flags & EVENTS_STREAM_HISTORY != 0 { + "history" + } else { + "from-now" + }; + let mode = if stream.flags & EVENTS_STREAM_APPEND != 0 { + "append" + } else { + "truncate" + }; + println!( + "{}\t{state}\t{}\t{}\t{}\t{history}\t{mode}\t{}\t{}", + stream.stream_id, + stream.records, + stream.bytes, + stream.lost, + stream.path, + stream.error, + ); + } + break; + } + EventsMessage::Result { + nonce: NONCE, + status, + detail, + .. + } => { + return Err(format!("{}: {detail}", blit_remote::status_text(status))); + } + _ => {} + } + } + } + EventsRecordCommand::Stop { id } => { + conn.send(&msg_events_stream_stop(NONCE, id)).await?; + let _ = wait_result(&mut conn, false).await?; + } + }, + } + Ok(()) +} diff --git a/crates/cli/src/learn.md b/crates/cli/src/learn.md index 5cfe81b5..02e5ecb8 100644 --- a/crates/cli/src/learn.md +++ b/crates/cli/src/learn.md @@ -79,6 +79,31 @@ continue from if it truncated. Both `show` and line-based `history` accept `--cols`/`--rows` to resize before reading, and `--ansi` to preserve colors. +## Server event journal + +The binary event journal is disabled by event type rather than globally. Its +default 1 MiB ring retains low-throughput lifecycle events; enable hot paths +only for targeted captures. + +```bash +blit events config +blit events set --events 'default,+frame.*,+pty.*' --size 8388608 +blit events set --if-revision 12 --events default --size 1048576 +blit events dump > snapshot.events # one binary snapshot +blit events tail > live.events # history, then follow +blit events tail --from-now -o live.events # follow new events locally +ID=$(blit events record start /var/log/blit.events) # detached server writer +blit events record list +blit events record stop "$ID" +``` + +`dump` and `tail` write binary `BLITEVT1` data to stdout unless `--output` +names a local path. `record` paths are on the server and continue after the +starting client disconnects. A recording id is returned only after its header +and history are flushed; `record list` shows state, counters, and delayed write +errors. Event operations from extensions use the same access and stream budgets +as direct clients. + ## Commands in a live shell A shell that emits OSC 133 records each command (see diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index 3e0b3c30..50bdf6fc 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -2,6 +2,7 @@ mod agent; mod attach; mod cli; mod completion; +mod events; mod extension; mod forward; mod fs; @@ -460,6 +461,20 @@ async fn async_main() { std::process::exit(1); } } + Command::Events { command } => { + let conn = &cli.connect; + let transport = match transport::connect(&conn.on, &conn.hub).await { + Ok(t) => t, + Err(e) => { + eprintln!("blit: {e}"); + std::process::exit(1); + } + }; + if let Err(error) = events::run(transport, command).await { + eprintln!("blit: {error}"); + std::process::exit(1); + } + } Command::Surface { command } => { let cmd = command.unwrap_or(SurfaceCommand::List); let conn = &cli.connect; diff --git a/crates/remote/src/events.rs b/crates/remote/src/events.rs new file mode 100644 index 00000000..a205f78b --- /dev/null +++ b/crates/remote/src/events.rs @@ -0,0 +1,985 @@ +//! Binary server event journal wire protocol (`blit.events.v1`). +//! +//! The family uses one direction-local envelope opcode. Every request is +//! correlated; live records use a stream id after the correlated start. + +use std::fmt; + +/// `S2C_HELLO` feature bit for `blit.events.v1`. +pub const FEATURE_EVENTS: u32 = 1 << 31; +/// Direction-local family envelope opcode. +pub const EVENTS: u8 = 0xD0; +pub const EVENTS_VERSION: u8 = 1; +pub const EVENTS_PROTOCOL: &str = "blit.events.v1"; + +pub const ACTIVATION_WORDS: usize = 4; +pub const ACTIVATION_BYTES: usize = ACTIVATION_WORDS * 8; + +pub const EVENTS_CONFIG_GET: u8 = 1; +pub const EVENTS_CONFIG_SET: u8 = 2; +pub const EVENTS_DUMP: u8 = 3; +pub const EVENTS_STREAM_START: u8 = 4; +pub const EVENTS_STREAM_STOP: u8 = 5; +pub const EVENTS_STREAM_LIST: u8 = 6; + +pub const EVENTS_CONFIG: u8 = 1; +pub const EVENTS_RESULT: u8 = 2; +pub const EVENTS_DUMPED: u8 = 3; +pub const EVENTS_STREAM_STARTED: u8 = 4; +pub const EVENTS_RECORD: u8 = 5; +pub const EVENTS_STREAM_STOPPED: u8 = 6; +pub const EVENTS_STREAM_GAP: u8 = 7; +pub const EVENTS_STREAMS: u8 = 8; + +pub const EVENTS_TARGET_CLIENT: u8 = 0; +pub const EVENTS_TARGET_FILE: u8 = 1; + +/// Sentinel in CONFIG_SET for an unconditional configuration replacement. +pub const EVENTS_CONFIG_REVISION_ANY: u64 = u64::MAX; + +pub const EVENTS_STREAM_STATE_RUNNING: u8 = 1; +pub const EVENTS_STREAM_STATE_STOPPED: u8 = 2; +pub const EVENTS_STREAM_STATE_FAILED: u8 = 3; + +/// Start a stream with the current retained history before live records. +pub const EVENTS_STREAM_HISTORY: u8 = 1 << 0; +/// Open a file target for append instead of truncating it. +pub const EVENTS_STREAM_APPEND: u8 = 1 << 1; +pub const EVENTS_STREAM_FLAGS: u8 = EVENTS_STREAM_HISTORY | EVENTS_STREAM_APPEND; + +/// Magic at the front of every self-describing binary dump. +pub const EVENT_DUMP_MAGIC: &[u8; 8] = b"BLITEVT1"; +pub const EVENT_DUMP_HEADER_LEN: usize = 84; +/// Bytes before an event's type-specific binary payload. +pub const EVENT_RECORD_HEADER_LEN: usize = 32; +/// Synthetic record type used only by file streams when their live receiver +/// lagged. Its payload is one little-endian `u64` count. +pub const EVENT_TYPE_STREAM_GAP: u16 = u16::MAX; + +/// Stages in a correlated `pty.create` payload. The fixed payload is +/// `[connection_id:u64][nonce:u16][stage:u8][status:u8][pty_id:u16]`. +pub const PTY_CREATE_REQUEST_RECEIVED: u8 = 1; +pub const PTY_CREATE_SESSION_ACQUIRED: u8 = 2; +pub const PTY_CREATE_SPAWN_BEGIN: u8 = 3; +pub const PTY_CREATE_SPAWN_END: u8 = 4; +pub const PTY_CREATE_REGISTERED: u8 = 5; +pub const PTY_CREATE_REFUSED: u8 = 6; +pub const PTY_CREATE_REPLY_WRITTEN: u8 = 7; + +/// Stable event ids. The numeric value is also its activation-bit index. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[repr(u16)] +pub enum EventType { + ServerStart = 0, + ServerStop = 1, + TaskStart = 2, + TaskStop = 3, + ClientConnect = 4, + ClientDisconnect = 5, + ClientReject = 6, + ConfigChange = 7, + StreamStart = 8, + StreamStop = 9, + ProtocolError = 10, + PtyCreate = 11, + PtyExit = 12, + PtyRemove = 13, + Deadline = 14, + Capacity = 15, + FrameRead = 16, + FrameWrite = 17, + MessageRead = 18, + MessageWrite = 19, + TickStart = 20, + TickStop = 21, + TickNudge = 22, + SessionLock = 23, + PtyRead = 24, + PtyWrite = 25, + PtyParse = 26, + PtySnapshot = 27, + PtyResize = 28, + PtyInput = 29, + CompositorEvent = 30, + CompositorCommand = 31, + SurfaceEncode = 32, + SurfaceFrame = 33, + AudioFrame = 34, + FsRequest = 35, + GitRequest = 36, + LspRequest = 37, + KvRequest = 38, + NetRequest = 39, + ProcessRequest = 40, + ExtensionRequest = 41, + ChannelRequest = 42, + ClientControl = 43, + OutboxQueue = 44, + Supervisor = 45, + ConnectionAccept = 46, + Error = 47, +} + +impl EventType { + pub const fn id(self) -> u16 { + self as u16 + } + + pub const fn name(self) -> &'static str { + EVENT_TYPE_CATALOG[self as usize].1 + } + + pub fn from_name(name: &str) -> Option { + EVENT_TYPE_CATALOG + .iter() + .find_map(|&(kind, candidate)| (candidate == name).then_some(kind)) + } +} + +pub const EVENT_TYPE_CATALOG: &[(EventType, &str)] = &[ + (EventType::ServerStart, "server.start"), + (EventType::ServerStop, "server.stop"), + (EventType::TaskStart, "task.start"), + (EventType::TaskStop, "task.stop"), + (EventType::ClientConnect, "client.connect"), + (EventType::ClientDisconnect, "client.disconnect"), + (EventType::ClientReject, "client.reject"), + (EventType::ConfigChange, "config.change"), + (EventType::StreamStart, "stream.start"), + (EventType::StreamStop, "stream.stop"), + (EventType::ProtocolError, "protocol.error"), + (EventType::PtyCreate, "pty.create"), + (EventType::PtyExit, "pty.exit"), + (EventType::PtyRemove, "pty.remove"), + (EventType::Deadline, "pty.deadline"), + (EventType::Capacity, "server.capacity"), + (EventType::FrameRead, "frame.read"), + (EventType::FrameWrite, "frame.write"), + (EventType::MessageRead, "message.read"), + (EventType::MessageWrite, "message.write"), + (EventType::TickStart, "tick.start"), + (EventType::TickStop, "tick.stop"), + (EventType::TickNudge, "tick.nudge"), + (EventType::SessionLock, "session.lock"), + (EventType::PtyRead, "pty.read"), + (EventType::PtyWrite, "pty.write"), + (EventType::PtyParse, "pty.parse"), + (EventType::PtySnapshot, "pty.snapshot"), + (EventType::PtyResize, "pty.resize"), + (EventType::PtyInput, "pty.input"), + (EventType::CompositorEvent, "compositor.event"), + (EventType::CompositorCommand, "compositor.command"), + (EventType::SurfaceEncode, "surface.encode"), + (EventType::SurfaceFrame, "surface.frame"), + (EventType::AudioFrame, "audio.frame"), + (EventType::FsRequest, "fs.request"), + (EventType::GitRequest, "git.request"), + (EventType::LspRequest, "lsp.request"), + (EventType::KvRequest, "kv.request"), + (EventType::NetRequest, "net.request"), + (EventType::ProcessRequest, "process.request"), + (EventType::ExtensionRequest, "extension.request"), + (EventType::ChannelRequest, "channel.request"), + (EventType::ClientControl, "client.control"), + (EventType::OutboxQueue, "outbox.queue"), + (EventType::Supervisor, "supervisor.event"), + (EventType::ConnectionAccept, "connection.accept"), + (EventType::Error, "server.error"), +]; + +/// The exact 256-bit activation set carried by the protocol. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct ActivationSet(pub [u64; ACTIVATION_WORDS]); + +impl ActivationSet { + pub const fn low_throughput() -> Self { + // IDs 0 through 15 are intentionally the low-volume lifecycle set. + Self([u16::MAX as u64, 0, 0, 0]) + } + + pub const fn all() -> Self { + Self([u64::MAX; ACTIVATION_WORDS]) + } + + pub const fn enabled(self, kind: EventType) -> bool { + let id = kind.id() as usize; + self.0[id / 64] & (1u64 << (id % 64)) != 0 + } + + pub fn set(&mut self, kind: EventType, enabled: bool) { + let id = kind.id() as usize; + let bit = 1u64 << (id % 64); + if enabled { + self.0[id / 64] |= bit; + } else { + self.0[id / 64] &= !bit; + } + } + + pub fn to_bytes(self) -> [u8; ACTIVATION_BYTES] { + let mut bytes = [0; ACTIVATION_BYTES]; + for (index, word) in self.0.into_iter().enumerate() { + bytes[index * 8..index * 8 + 8].copy_from_slice(&word.to_le_bytes()); + } + bytes + } + + pub fn from_bytes(bytes: &[u8]) -> Option { + if bytes.len() != ACTIVATION_BYTES { + return None; + } + let mut words = [0; ACTIVATION_WORDS]; + for (index, word) in words.iter_mut().enumerate() { + *word = u64::from_le_bytes(bytes[index * 8..index * 8 + 8].try_into().ok()?); + } + Some(Self(words)) + } +} + +/// Parse `all`, `none`, `default`, exact names, and `category.*` selectors. +/// A leading `+`/`-` expression edits the default set; otherwise the set +/// starts empty and is built left-to-right. +pub fn parse_activation_spec(spec: &str) -> Result { + let first = spec.split(',').map(str::trim).find(|part| !part.is_empty()); + let mut set = if first.is_some_and(|part| part.starts_with(['+', '-'])) { + ActivationSet::low_throughput() + } else { + ActivationSet::default() + }; + for raw in spec + .split(',') + .map(str::trim) + .filter(|part| !part.is_empty()) + { + let (enabled, selector) = match raw.as_bytes().first() { + Some(b'+') => (true, &raw[1..]), + Some(b'-') => (false, &raw[1..]), + _ => (true, raw), + }; + match selector { + "all" => { + set = if enabled { + ActivationSet::all() + } else { + ActivationSet::default() + } + } + "none" => { + if enabled { + set = ActivationSet::default(); + } + } + "default" => { + if enabled { + set = ActivationSet::low_throughput(); + } else { + for &(kind, _) in EVENT_TYPE_CATALOG { + if ActivationSet::low_throughput().enabled(kind) { + set.set(kind, false); + } + } + } + } + _ => { + let mut matched = false; + if let Some(prefix) = selector.strip_suffix(".*") { + for &(kind, name) in EVENT_TYPE_CATALOG { + if name + .strip_prefix(prefix) + .is_some_and(|tail| tail.starts_with('.')) + { + set.set(kind, enabled); + matched = true; + } + } + } else if let Some(kind) = EventType::from_name(selector) { + set.set(kind, enabled); + matched = true; + } + if !matched { + return Err(format!("unknown event selector {selector:?}")); + } + } + } + } + Ok(set) +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum EventsRequest<'a> { + ConfigGet { + nonce: u16, + }, + ConfigSet { + nonce: u16, + expected_revision: u64, + size: u64, + activations: ActivationSet, + }, + Dump { + nonce: u16, + }, + StreamStart { + nonce: u16, + target: u8, + flags: u8, + path: &'a str, + }, + StreamStop { + nonce: u16, + stream_id: u32, + }, + StreamList { + nonce: u16, + }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum EventsDecodeError { + NotEvents, + UnsupportedVersion, + Truncated, + TrailingBytes, + InvalidTarget, + InvalidFlags, + InvalidPath, + InvalidOperation, +} + +impl fmt::Display for EventsDecodeError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(match self { + Self::NotEvents => "not an events packet", + Self::UnsupportedVersion => "unsupported events protocol version", + Self::Truncated => "events packet is truncated", + Self::TrailingBytes => "events packet has trailing bytes", + Self::InvalidTarget => "invalid events stream target", + Self::InvalidFlags => "invalid events stream flags", + Self::InvalidPath => "invalid events stream path", + Self::InvalidOperation => "invalid events operation", + }) + } +} + +pub fn parse_events_request(packet: &[u8]) -> Result, EventsDecodeError> { + if packet.first() != Some(&EVENTS) { + return Err(EventsDecodeError::NotEvents); + } + if packet.len() < 5 { + return Err(EventsDecodeError::Truncated); + } + if packet[1] != EVENTS_VERSION { + return Err(EventsDecodeError::UnsupportedVersion); + } + let op = packet[2]; + let nonce = u16::from_le_bytes([packet[3], packet[4]]); + let body = &packet[5..]; + match op { + EVENTS_CONFIG_GET if body.is_empty() => Ok(EventsRequest::ConfigGet { nonce }), + EVENTS_CONFIG_GET => Err(EventsDecodeError::TrailingBytes), + EVENTS_CONFIG_SET => { + if body.len() < 16 + ACTIVATION_BYTES { + return Err(EventsDecodeError::Truncated); + } + if body.len() != 16 + ACTIVATION_BYTES { + return Err(EventsDecodeError::TrailingBytes); + } + Ok(EventsRequest::ConfigSet { + nonce, + expected_revision: u64::from_le_bytes( + body[..8].try_into().expect("checked length"), + ), + size: u64::from_le_bytes(body[8..16].try_into().expect("checked length")), + activations: ActivationSet::from_bytes(&body[16..]).expect("checked length"), + }) + } + EVENTS_DUMP if body.is_empty() => Ok(EventsRequest::Dump { nonce }), + EVENTS_DUMP => Err(EventsDecodeError::TrailingBytes), + EVENTS_STREAM_START => { + if body.len() < 4 { + return Err(EventsDecodeError::Truncated); + } + let target = body[0]; + let flags = body[1]; + if target > EVENTS_TARGET_FILE { + return Err(EventsDecodeError::InvalidTarget); + } + if flags & !EVENTS_STREAM_FLAGS != 0 + || target == EVENTS_TARGET_CLIENT && flags & EVENTS_STREAM_APPEND != 0 + { + return Err(EventsDecodeError::InvalidFlags); + } + let path_len = u16::from_le_bytes([body[2], body[3]]) as usize; + if body.len() < 4 + path_len { + return Err(EventsDecodeError::Truncated); + } + if body.len() != 4 + path_len { + return Err(EventsDecodeError::TrailingBytes); + } + let path = + std::str::from_utf8(&body[4..]).map_err(|_| EventsDecodeError::InvalidPath)?; + if path.as_bytes().contains(&0) + || (target == EVENTS_TARGET_CLIENT && !path.is_empty()) + || (target == EVENTS_TARGET_FILE && path.is_empty()) + { + return Err(EventsDecodeError::InvalidPath); + } + Ok(EventsRequest::StreamStart { + nonce, + target, + flags, + path, + }) + } + EVENTS_STREAM_STOP => { + if body.len() < 4 { + return Err(EventsDecodeError::Truncated); + } + if body.len() != 4 { + return Err(EventsDecodeError::TrailingBytes); + } + Ok(EventsRequest::StreamStop { + nonce, + stream_id: u32::from_le_bytes(body.try_into().expect("checked length")), + }) + } + EVENTS_STREAM_LIST if body.is_empty() => Ok(EventsRequest::StreamList { nonce }), + EVENTS_STREAM_LIST => Err(EventsDecodeError::TrailingBytes), + _ => Err(EventsDecodeError::InvalidOperation), + } +} + +fn request(op: u8, nonce: u16) -> Vec { + let mut msg = vec![EVENTS, EVENTS_VERSION, op]; + msg.extend_from_slice(&nonce.to_le_bytes()); + msg +} + +pub fn msg_events_config_get(nonce: u16) -> Vec { + request(EVENTS_CONFIG_GET, nonce) +} + +pub fn msg_events_config_set( + nonce: u16, + expected_revision: u64, + size: u64, + activations: ActivationSet, +) -> Vec { + let mut msg = request(EVENTS_CONFIG_SET, nonce); + msg.extend_from_slice(&expected_revision.to_le_bytes()); + msg.extend_from_slice(&size.to_le_bytes()); + msg.extend_from_slice(&activations.to_bytes()); + msg +} + +pub fn msg_events_dump(nonce: u16) -> Vec { + request(EVENTS_DUMP, nonce) +} + +pub fn msg_events_stream_start(nonce: u16, target: u8, flags: u8, path: &str) -> Vec { + let path = path.as_bytes(); + let len = path.len().min(u16::MAX as usize); + let mut msg = request(EVENTS_STREAM_START, nonce); + msg.push(target); + msg.push(flags); + msg.extend_from_slice(&(len as u16).to_le_bytes()); + msg.extend_from_slice(&path[..len]); + msg +} + +pub fn msg_events_stream_stop(nonce: u16, stream_id: u32) -> Vec { + let mut msg = request(EVENTS_STREAM_STOP, nonce); + msg.extend_from_slice(&stream_id.to_le_bytes()); + msg +} + +pub fn msg_events_stream_list(nonce: u16) -> Vec { + request(EVENTS_STREAM_LIST, nonce) +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct EventStreamInfo<'a> { + pub stream_id: u32, + pub state: u8, + pub flags: u8, + pub records: u64, + pub bytes: u64, + pub lost: u64, + pub path: &'a str, + pub error: &'a str, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct EventConfig { + pub revision: u64, + pub size: u64, + pub used: u64, + pub records: u64, + pub dropped: u64, + pub next_sequence: u64, + pub activations: ActivationSet, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum EventsMessage<'a> { + Config { + nonce: u16, + revision: u64, + size: u64, + used: u64, + records: u64, + dropped: u64, + next_sequence: u64, + activations: ActivationSet, + }, + Result { + nonce: u16, + status: u8, + stream_id: u32, + detail: &'a str, + }, + Dump { + nonce: u16, + bytes: &'a [u8], + }, + StreamStarted { + nonce: u16, + status: u8, + stream_id: u32, + detail: &'a str, + }, + Records { + stream_id: u32, + count: u16, + records: &'a [u8], + }, + StreamStopped { + stream_id: u32, + status: u8, + detail: &'a str, + }, + StreamGap { + stream_id: u32, + lost: u64, + }, + Streams { + nonce: u16, + streams: Vec>, + }, +} + +fn response(kind: u8, nonce: u16) -> Vec { + let mut msg = vec![EVENTS, EVENTS_VERSION, kind]; + msg.extend_from_slice(&nonce.to_le_bytes()); + msg +} + +pub fn msg_events_config(nonce: u16, config: EventConfig) -> Vec { + let mut msg = response(EVENTS_CONFIG, nonce); + for value in [ + config.revision, + config.size, + config.used, + config.records, + config.dropped, + config.next_sequence, + ] { + msg.extend_from_slice(&value.to_le_bytes()); + } + msg.extend_from_slice(&config.activations.to_bytes()); + msg +} + +pub fn msg_events_result(nonce: u16, status: u8, stream_id: u32, detail: &str) -> Vec { + let mut msg = response(EVENTS_RESULT, nonce); + msg.push(status); + msg.extend_from_slice(&stream_id.to_le_bytes()); + msg.extend_from_slice(detail.as_bytes()); + msg +} + +pub fn msg_events_dumped(nonce: u16, dump: &[u8]) -> Vec { + let mut msg = response(EVENTS_DUMPED, nonce); + msg.extend_from_slice(dump); + msg +} + +pub fn msg_events_stream_started(nonce: u16, status: u8, stream_id: u32, detail: &str) -> Vec { + let mut msg = response(EVENTS_STREAM_STARTED, nonce); + msg.push(status); + msg.extend_from_slice(&stream_id.to_le_bytes()); + msg.extend_from_slice(detail.as_bytes()); + msg +} + +pub fn msg_events_records>(stream_id: u32, records: &[T]) -> Vec { + let count = records.len().min(u16::MAX as usize); + let mut msg = vec![EVENTS, EVENTS_VERSION, EVENTS_RECORD]; + msg.extend_from_slice(&stream_id.to_le_bytes()); + msg.extend_from_slice(&(count as u16).to_le_bytes()); + for record in &records[..count] { + msg.extend_from_slice(record.as_ref()); + } + msg +} + +pub fn msg_events_stream_stopped(stream_id: u32, status: u8, detail: &str) -> Vec { + let mut msg = vec![EVENTS, EVENTS_VERSION, EVENTS_STREAM_STOPPED]; + msg.extend_from_slice(&stream_id.to_le_bytes()); + msg.push(status); + msg.extend_from_slice(detail.as_bytes()); + msg +} + +pub fn msg_events_stream_gap(stream_id: u32, lost: u64) -> Vec { + let mut msg = vec![EVENTS, EVENTS_VERSION, EVENTS_STREAM_GAP]; + msg.extend_from_slice(&stream_id.to_le_bytes()); + msg.extend_from_slice(&lost.to_le_bytes()); + msg +} + +pub fn msg_events_streams(nonce: u16, streams: &[EventStreamInfo<'_>]) -> Vec { + let count = streams.len().min(u16::MAX as usize); + let mut msg = response(EVENTS_STREAMS, nonce); + msg.extend_from_slice(&(count as u16).to_le_bytes()); + for stream in &streams[..count] { + let mut path_len = stream.path.len().min(u16::MAX as usize); + while !stream.path.is_char_boundary(path_len) { + path_len -= 1; + } + msg.extend_from_slice(&stream.stream_id.to_le_bytes()); + let mut error_len = stream.error.len().min(u16::MAX as usize); + while !stream.error.is_char_boundary(error_len) { + error_len -= 1; + } + msg.push(stream.state); + msg.push(stream.flags & EVENTS_STREAM_FLAGS); + msg.extend_from_slice(&stream.records.to_le_bytes()); + msg.extend_from_slice(&stream.bytes.to_le_bytes()); + msg.extend_from_slice(&stream.lost.to_le_bytes()); + msg.extend_from_slice(&(path_len as u16).to_le_bytes()); + msg.extend_from_slice(&(error_len as u16).to_le_bytes()); + msg.extend_from_slice(&stream.path.as_bytes()[..path_len]); + msg.extend_from_slice(&stream.error.as_bytes()[..error_len]); + } + msg +} + +pub fn parse_events_message(packet: &[u8]) -> Result, EventsDecodeError> { + if packet.first() != Some(&EVENTS) { + return Err(EventsDecodeError::NotEvents); + } + if packet.len() < 3 { + return Err(EventsDecodeError::Truncated); + } + if packet[1] != EVENTS_VERSION { + return Err(EventsDecodeError::UnsupportedVersion); + } + match packet[2] { + EVENTS_CONFIG => { + let expected = 5 + 6 * 8 + ACTIVATION_BYTES; + if packet.len() < expected { + return Err(EventsDecodeError::Truncated); + } + if packet.len() != expected { + return Err(EventsDecodeError::TrailingBytes); + } + let nonce = u16::from_le_bytes([packet[3], packet[4]]); + let mut at = 5; + let mut take_u64 = || { + let value = u64::from_le_bytes(packet[at..at + 8].try_into().expect("checked")); + at += 8; + value + }; + Ok(EventsMessage::Config { + nonce, + revision: take_u64(), + size: take_u64(), + used: take_u64(), + records: take_u64(), + dropped: take_u64(), + next_sequence: take_u64(), + activations: ActivationSet::from_bytes(&packet[at..]).expect("checked"), + }) + } + EVENTS_RESULT | EVENTS_STREAM_STARTED => { + if packet.len() < 10 { + return Err(EventsDecodeError::Truncated); + } + let nonce = u16::from_le_bytes([packet[3], packet[4]]); + let status = packet[5]; + let stream_id = u32::from_le_bytes(packet[6..10].try_into().expect("checked")); + let detail = + std::str::from_utf8(&packet[10..]).map_err(|_| EventsDecodeError::InvalidPath)?; + if packet[2] == EVENTS_RESULT { + Ok(EventsMessage::Result { + nonce, + status, + stream_id, + detail, + }) + } else { + Ok(EventsMessage::StreamStarted { + nonce, + status, + stream_id, + detail, + }) + } + } + EVENTS_DUMPED => { + if packet.len() < 5 { + return Err(EventsDecodeError::Truncated); + } + Ok(EventsMessage::Dump { + nonce: u16::from_le_bytes([packet[3], packet[4]]), + bytes: &packet[5..], + }) + } + EVENTS_RECORD => { + if packet.len() < 9 { + return Err(EventsDecodeError::Truncated); + } + let count = u16::from_le_bytes(packet[7..9].try_into().expect("checked")); + if count == 0 { + return Err(EventsDecodeError::InvalidFlags); + } + let mut at = 9; + for _ in 0..count { + if packet.len().saturating_sub(at) < EVENT_RECORD_HEADER_LEN { + return Err(EventsDecodeError::Truncated); + } + let record_len = + u32::from_le_bytes(packet[at..at + 4].try_into().expect("checked")) as usize; + if record_len < EVENT_RECORD_HEADER_LEN + || packet.len().saturating_sub(at) < record_len + { + return Err(EventsDecodeError::Truncated); + } + at += record_len; + } + if at != packet.len() { + return Err(EventsDecodeError::TrailingBytes); + } + Ok(EventsMessage::Records { + stream_id: u32::from_le_bytes(packet[3..7].try_into().expect("checked")), + count, + records: &packet[9..], + }) + } + EVENTS_STREAM_STOPPED => { + if packet.len() < 8 { + return Err(EventsDecodeError::Truncated); + } + Ok(EventsMessage::StreamStopped { + stream_id: u32::from_le_bytes(packet[3..7].try_into().expect("checked")), + status: packet[7], + detail: std::str::from_utf8(&packet[8..]) + .map_err(|_| EventsDecodeError::InvalidPath)?, + }) + } + EVENTS_STREAM_GAP => { + if packet.len() < 15 { + return Err(EventsDecodeError::Truncated); + } + if packet.len() != 15 { + return Err(EventsDecodeError::TrailingBytes); + } + Ok(EventsMessage::StreamGap { + stream_id: u32::from_le_bytes(packet[3..7].try_into().expect("checked")), + lost: u64::from_le_bytes(packet[7..15].try_into().expect("checked")), + }) + } + EVENTS_STREAMS => { + if packet.len() < 7 { + return Err(EventsDecodeError::Truncated); + } + let nonce = u16::from_le_bytes([packet[3], packet[4]]); + let count = u16::from_le_bytes([packet[5], packet[6]]) as usize; + let mut at = 7; + let mut streams = Vec::with_capacity(count); + for _ in 0..count { + if packet.len().saturating_sub(at) < 34 { + return Err(EventsDecodeError::Truncated); + } + let stream_id = u32::from_le_bytes(packet[at..at + 4].try_into().expect("checked")); + let state = packet[at + 4]; + if !matches!( + state, + EVENTS_STREAM_STATE_RUNNING + | EVENTS_STREAM_STATE_STOPPED + | EVENTS_STREAM_STATE_FAILED + ) { + return Err(EventsDecodeError::InvalidFlags); + } + let flags = packet[at + 5]; + if flags & !EVENTS_STREAM_FLAGS != 0 { + return Err(EventsDecodeError::InvalidFlags); + } + let records = + u64::from_le_bytes(packet[at + 6..at + 14].try_into().expect("checked")); + let bytes = + u64::from_le_bytes(packet[at + 14..at + 22].try_into().expect("checked")); + let lost = + u64::from_le_bytes(packet[at + 22..at + 30].try_into().expect("checked")); + let path_len = u16::from_le_bytes([packet[at + 30], packet[at + 31]]) as usize; + let error_len = u16::from_le_bytes([packet[at + 32], packet[at + 33]]) as usize; + at += 34; + if packet.len().saturating_sub(at) < path_len.saturating_add(error_len) { + return Err(EventsDecodeError::Truncated); + } + let path = std::str::from_utf8(&packet[at..at + path_len]) + .map_err(|_| EventsDecodeError::InvalidPath)?; + at += path_len; + let error = std::str::from_utf8(&packet[at..at + error_len]) + .map_err(|_| EventsDecodeError::InvalidPath)?; + at += error_len; + streams.push(EventStreamInfo { + stream_id, + state, + flags, + records, + bytes, + lost, + path, + error, + }); + } + if at != packet.len() { + return Err(EventsDecodeError::TrailingBytes); + } + Ok(EventsMessage::Streams { nonce, streams }) + } + _ => Err(EventsDecodeError::InvalidOperation), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn activation_selectors_are_composable() { + let set = parse_activation_spec("default,+frame.*,-client.disconnect").unwrap(); + assert!(set.enabled(EventType::ServerStart)); + assert!(set.enabled(EventType::FrameRead)); + assert!(set.enabled(EventType::FrameWrite)); + assert!(!set.enabled(EventType::ClientDisconnect)); + assert!(!set.enabled(EventType::PtyRead)); + assert_eq!( + parse_activation_spec("default,-none").unwrap(), + ActivationSet::low_throughput() + ); + } + + #[test] + fn request_codecs_round_trip() { + let activations = ActivationSet([1, 2, 3, 4]); + assert_eq!( + parse_events_request(&msg_events_config_set(9, 17, 1024, activations)).unwrap(), + EventsRequest::ConfigSet { + nonce: 9, + expected_revision: 17, + size: 1024, + activations + } + ); + assert_eq!( + parse_events_request(&msg_events_stream_start( + 7, + EVENTS_TARGET_FILE, + EVENTS_STREAM_HISTORY | EVENTS_STREAM_APPEND, + "/tmp/events.bin", + )) + .unwrap(), + EventsRequest::StreamStart { + nonce: 7, + target: EVENTS_TARGET_FILE, + flags: EVENTS_STREAM_HISTORY | EVENTS_STREAM_APPEND, + path: "/tmp/events.bin", + } + ); + assert_eq!( + parse_events_request(&msg_events_stream_stop(8, 42)).unwrap(), + EventsRequest::StreamStop { + nonce: 8, + stream_id: 42 + } + ); + assert_eq!( + parse_events_request(&msg_events_stream_list(10)).unwrap(), + EventsRequest::StreamList { nonce: 10 } + ); + } + + #[test] + fn response_codecs_round_trip() { + let activations = ActivationSet([5, 6, 7, 8]); + let msg = msg_events_config( + 2, + EventConfig { + revision: 17, + size: 100, + used: 80, + records: 3, + dropped: 4, + next_sequence: 9, + activations, + }, + ); + assert_eq!( + parse_events_message(&msg).unwrap(), + EventsMessage::Config { + nonce: 2, + revision: 17, + size: 100, + used: 80, + records: 3, + dropped: 4, + next_sequence: 9, + activations, + } + ); + let result = msg_events_stream_started(4, crate::STATUS_OK, 11, ""); + assert_eq!( + parse_events_message(&result).unwrap(), + EventsMessage::StreamStarted { + nonce: 4, + status: crate::STATUS_OK, + stream_id: 11, + detail: "", + } + ); + let streams = [EventStreamInfo { + stream_id: 12, + state: EVENTS_STREAM_STATE_FAILED, + flags: EVENTS_STREAM_HISTORY, + records: 19, + bytes: 2048, + lost: 3, + path: "/tmp/events.bin", + error: "disk full", + }]; + let message = msg_events_streams(5, &streams); + assert_eq!( + parse_events_message(&message).unwrap(), + EventsMessage::Streams { + nonce: 5, + streams: streams.to_vec(), + } + ); + let record = [EVENT_RECORD_HEADER_LEN as u8; EVENT_RECORD_HEADER_LEN]; + let mut record = record.to_vec(); + record[..4].copy_from_slice(&(EVENT_RECORD_HEADER_LEN as u32).to_le_bytes()); + let records = [&record[..], &record[..]]; + let message = msg_events_records(9, &records); + assert_eq!( + parse_events_message(&message).unwrap(), + EventsMessage::Records { + stream_id: 9, + count: 2, + records: &[record.as_slice(), record.as_slice()].concat(), + } + ); + } +} diff --git a/crates/remote/src/lib.rs b/crates/remote/src/lib.rs index 984142b9..6b2fd7e9 100644 --- a/crates/remote/src/lib.rs +++ b/crates/remote/src/lib.rs @@ -44,6 +44,9 @@ pub mod media; /// cursor a client feeds back to read only what is new. pub mod journal; +/// Process-wide binary server event journal (`blit.events.v1`). +pub mod events; + /// Cap on any single LZ4-decompressed payload, protocol-wide /// (docs/protocol.md "Compressed payloads"). Receivers check the prepended /// size against it *before* allocating, so a hostile or corrupt length diff --git a/crates/server/src/events.rs b/crates/server/src/events.rs new file mode 100644 index 00000000..c2bf3779 --- /dev/null +++ b/crates/server/src/events.rs @@ -0,0 +1,929 @@ +//! Process-wide bounded binary event journal. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU8, AtomicU32, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; + +use blit_remote::events::{ + ACTIVATION_WORDS, ActivationSet, EVENT_DUMP_HEADER_LEN, EVENT_DUMP_MAGIC, + EVENT_RECORD_HEADER_LEN, EVENT_TYPE_STREAM_GAP, EVENTS_STREAM_APPEND, EVENTS_STREAM_HISTORY, + EventType, parse_activation_spec, +}; +use tokio::io::AsyncWriteExt; +use tokio::sync::{broadcast, oneshot}; + +pub(crate) const DEFAULT_RING_SIZE: usize = 1024 * 1024; +pub(crate) const MIN_RING_SIZE: usize = 4 * 1024; +pub(crate) const MAX_RING_SIZE: usize = blit_remote::MAX_LOGICAL_MESSAGE - 4096; +const LIVE_CHANNEL_RECORDS: usize = 4096; +pub(crate) const LIVE_BATCH_MAX_BYTES: usize = 256 * 1024; + +#[derive(Clone, Debug)] +pub(crate) struct StartupFile { + pub path: String, + pub flags: u8, +} + +#[derive(Clone, Copy, Debug)] +pub(crate) struct EventStats { + pub revision: u64, + pub capacity: usize, + pub used: usize, + pub records: u64, + pub dropped: u64, + pub next_sequence: u64, +} + +struct Ring { + bytes: Box<[u8]>, + head: usize, + used: usize, + records: u64, + dropped: u64, +} + +impl Ring { + fn new(capacity: usize) -> Self { + Self { + bytes: vec![0; capacity].into_boxed_slice(), + head: 0, + used: 0, + records: 0, + dropped: 0, + } + } + + fn capacity(&self) -> usize { + self.bytes.len() + } + + fn copy_out(&self, at: usize, len: usize, out: &mut Vec) { + let first = len.min(self.capacity() - at); + out.extend_from_slice(&self.bytes[at..at + first]); + if first < len { + out.extend_from_slice(&self.bytes[..len - first]); + } + } + + fn bytes_at(&self, at: usize, len: usize) -> Vec { + let mut out = Vec::with_capacity(len); + self.copy_out(at, len, &mut out); + out + } + + fn read_u32(&self, at: usize) -> u32 { + let mut bytes = [0; 4]; + let first = bytes.len().min(self.capacity() - at); + bytes[..first].copy_from_slice(&self.bytes[at..at + first]); + if first < bytes.len() { + let remaining = bytes.len() - first; + bytes[first..].copy_from_slice(&self.bytes[..remaining]); + } + u32::from_le_bytes(bytes) + } + + fn write_at(&mut self, at: usize, data: &[u8]) { + let first = data.len().min(self.capacity() - at); + self.bytes[at..at + first].copy_from_slice(&data[..first]); + if first < data.len() { + self.bytes[..data.len() - first].copy_from_slice(&data[first..]); + } + } + + fn oldest_len(&self) -> Option { + if self.used < 4 { + return None; + } + let len = self.read_u32(self.head) as usize; + (len >= EVENT_RECORD_HEADER_LEN && len <= self.used).then_some(len) + } + + fn evict_oldest(&mut self) { + let Some(len) = self.oldest_len() else { + // A corrupt in-memory prefix cannot be recovered record-by-record. + self.head = 0; + self.used = 0; + self.records = 0; + self.dropped = self.dropped.saturating_add(1); + return; + }; + self.head = (self.head + len) % self.capacity(); + self.used -= len; + self.records = self.records.saturating_sub(1); + self.dropped = self.dropped.saturating_add(1); + } + + fn append(&mut self, record: &[u8]) -> bool { + self.append_parts(record, &[]) + } + + fn append_parts(&mut self, header: &[u8], payload: &[u8]) -> bool { + let Some(len) = header.len().checked_add(payload.len()) else { + self.dropped = self.dropped.saturating_add(1); + return false; + }; + if len > self.capacity() { + self.dropped = self.dropped.saturating_add(1); + return false; + } + while self.capacity() - self.used < len { + self.evict_oldest(); + } + let tail = (self.head + self.used) % self.capacity(); + self.write_at(tail, header); + self.write_at((tail + header.len()) % self.capacity(), payload); + self.used += len; + self.records = self.records.saturating_add(1); + true + } + + fn record_vecs(&self) -> Vec> { + let mut result = Vec::with_capacity(self.records.min(usize::MAX as u64) as usize); + let mut at = self.head; + let mut left = self.used; + while left >= EVENT_RECORD_HEADER_LEN { + let len = self.read_u32(at) as usize; + if len < EVENT_RECORD_HEADER_LEN || len > left { + break; + } + result.push(self.bytes_at(at, len)); + at = (at + len) % self.capacity(); + left -= len; + } + result + } + + fn resize(&mut self, capacity: usize) { + if capacity == self.capacity() { + return; + } + let records = self.record_vecs(); + let mut replacement = Ring::new(capacity); + replacement.dropped = self.dropped; + for record in records { + replacement.append(&record); + } + *self = replacement; + } +} + +pub(crate) struct EventLog { + activations: [AtomicU64; ACTIVATION_WORDS], + ring: Mutex, + config_revision: AtomicU64, + next_sequence: AtomicU64, + next_stream_id: AtomicU32, + started: Instant, + started_unix_ns: u64, + live_tx: broadcast::Sender>, + file_streams: Mutex>, +} + +struct FileStreamTask { + stop: oneshot::Sender<()>, + join: tokio::task::JoinHandle>, + path: String, + flags: u8, + progress: Arc, +} + +type ClientEventStream = (u32, Option>, broadcast::Receiver>); +type EventSubscription = (Option>, u64, broadcast::Receiver>); + +struct FileStreamProgress { + state: AtomicU8, + records: AtomicU64, + bytes: AtomicU64, + lost: AtomicU64, + error: Mutex>, +} + +impl FileStreamProgress { + fn new(records: u64, bytes: u64) -> Arc { + Arc::new(Self { + state: AtomicU8::new(blit_remote::events::EVENTS_STREAM_STATE_RUNNING), + records: AtomicU64::new(records), + bytes: AtomicU64::new(bytes), + lost: AtomicU64::new(0), + error: Mutex::new(None), + }) + } + + fn fail(&self, error: String) -> String { + self.state.store( + blit_remote::events::EVENTS_STREAM_STATE_FAILED, + Ordering::Release, + ); + *self.error.lock().expect("event file progress poisoned") = Some(error.clone()); + error + } + + fn stop(&self) { + self.state.store( + blit_remote::events::EVENTS_STREAM_STATE_STOPPED, + Ordering::Release, + ); + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ConfigureError { + InvalidSize, + Conflict, +} + +impl std::fmt::Display for ConfigureError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Self::InvalidSize => "ring size is outside the supported range", + Self::Conflict => "event configuration revision changed", + }) + } +} + +#[derive(Debug)] +pub(crate) enum StartFileStreamError { + Io(String), +} + +impl std::fmt::Display for StartFileStreamError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Io(error) => f.write_str(error), + } + } +} + +pub(crate) struct FileStreamInfo { + pub id: u32, + pub state: u8, + pub flags: u8, + pub records: u64, + pub bytes: u64, + pub lost: u64, + pub path: String, + pub error: String, +} + +impl EventLog { + pub(crate) fn from_env() -> (Arc, Option) { + let size = std::env::var("BLIT_EVENTS_SIZE") + .ok() + .and_then(|value| value.parse::().ok()) + .filter(|size| (MIN_RING_SIZE..=MAX_RING_SIZE).contains(size)) + .unwrap_or(DEFAULT_RING_SIZE); + let activations = match std::env::var("BLIT_EVENTS") { + Ok(spec) => match parse_activation_spec(&spec) { + Ok(set) => set, + Err(error) => { + eprintln!("blit-server: ignoring invalid BLIT_EVENTS: {error}"); + ActivationSet::low_throughput() + } + }, + Err(_) => ActivationSet::low_throughput(), + }; + let file = std::env::var("BLIT_EVENTS_FILE") + .ok() + .filter(|path| !path.is_empty()) + .map(|path| { + let mut flags = 0; + if !std::env::var("BLIT_EVENTS_FILE_HISTORY").is_ok_and(|value| value == "0") { + flags |= EVENTS_STREAM_HISTORY; + } + if std::env::var("BLIT_EVENTS_FILE_APPEND").is_ok_and(|value| value == "1") { + flags |= EVENTS_STREAM_APPEND; + } + StartupFile { path, flags } + }); + let (live_tx, _) = broadcast::channel(LIVE_CHANNEL_RECORDS); + let started_unix_ns = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + .min(u64::MAX as u128) as u64; + let log = Arc::new(Self { + activations: std::array::from_fn(|index| AtomicU64::new(activations.0[index])), + ring: Mutex::new(Ring::new(size)), + config_revision: AtomicU64::new(1), + next_sequence: AtomicU64::new(0), + next_stream_id: AtomicU32::new(1), + started: Instant::now(), + started_unix_ns, + live_tx, + file_streams: Mutex::new(HashMap::new()), + }); + (log, file) + } + + #[cfg(test)] + pub(crate) fn new(size: usize, activations: ActivationSet) -> Arc { + let (live_tx, _) = broadcast::channel(16); + Arc::new(Self { + activations: std::array::from_fn(|index| AtomicU64::new(activations.0[index])), + ring: Mutex::new(Ring::new(size)), + config_revision: AtomicU64::new(1), + next_sequence: AtomicU64::new(0), + next_stream_id: AtomicU32::new(1), + started: Instant::now(), + started_unix_ns: 0, + live_tx, + file_streams: Mutex::new(HashMap::new()), + }) + } + + #[inline] + pub(crate) fn enabled(&self, kind: EventType) -> bool { + let id = kind.id() as usize; + self.activations[id / 64].load(Ordering::Relaxed) & (1u64 << (id % 64)) != 0 + } + + pub(crate) fn activations(&self) -> ActivationSet { + ActivationSet(std::array::from_fn(|index| { + self.activations[index].load(Ordering::Acquire) + })) + } + + pub(crate) fn configure( + &self, + size: usize, + activations: ActivationSet, + expected_revision: Option, + ) -> Result { + if !(MIN_RING_SIZE..=MAX_RING_SIZE).contains(&size) { + return Err(ConfigureError::InvalidSize); + } + let mut ring = self.ring.lock().expect("event ring poisoned"); + let revision = self.config_revision.load(Ordering::Acquire); + if expected_revision.is_some_and(|expected| expected != revision) { + return Err(ConfigureError::Conflict); + } + ring.resize(size); + for (word, value) in self.activations.iter().zip(activations.0) { + word.store(value, Ordering::Release); + } + let revision = if revision >= blit_remote::events::EVENTS_CONFIG_REVISION_ANY - 1 { + 1 + } else { + revision + 1 + }; + self.config_revision.store(revision, Ordering::Release); + Ok(self.stats_locked(&ring, revision)) + } + + #[cfg(test)] + pub(crate) fn stats(&self) -> EventStats { + let ring = self.ring.lock().expect("event ring poisoned"); + let revision = self.config_revision.load(Ordering::Acquire); + self.stats_locked(&ring, revision) + } + + fn stats_locked(&self, ring: &Ring, revision: u64) -> EventStats { + EventStats { + revision, + capacity: ring.capacity(), + used: ring.used, + records: ring.records, + dropped: ring.dropped, + next_sequence: self.next_sequence.load(Ordering::Acquire), + } + } + + pub(crate) fn configuration(&self) -> (EventStats, ActivationSet) { + let ring = self.ring.lock().expect("event ring poisoned"); + let revision = self.config_revision.load(Ordering::Acquire); + (self.stats_locked(&ring, revision), self.activations()) + } + + pub(crate) fn record(&self, kind: EventType, flags: u16, payload: &[u8]) { + let sequence = self.next_sequence.fetch_add(1, Ordering::Relaxed); + let monotonic_ns = self.started.elapsed().as_nanos().min(u64::MAX as u128) as u64; + let unix_ns = self.started_unix_ns.saturating_add(monotonic_ns); + let max_payload = blit_remote::MAX_LOGICAL_MESSAGE + .saturating_sub(EVENT_RECORD_HEADER_LEN) + .saturating_sub(9); + let payload = &payload[..payload.len().min(max_payload)]; + let len = EVENT_RECORD_HEADER_LEN + payload.len(); + let mut header = [0; EVENT_RECORD_HEADER_LEN]; + header[0..4].copy_from_slice(&(len as u32).to_le_bytes()); + header[4..6].copy_from_slice(&kind.id().to_le_bytes()); + header[6..8].copy_from_slice(&flags.to_le_bytes()); + header[8..16].copy_from_slice(&sequence.to_le_bytes()); + header[16..24].copy_from_slice(&monotonic_ns.to_le_bytes()); + header[24..32].copy_from_slice(&unix_ns.to_le_bytes()); + let mut ring = self.ring.lock().expect("event ring poisoned"); + ring.append_parts(&header, payload); + // Live streams still see a record that is larger than the configured + // ring. The ring's dropped counter reports that it was not retained. + if self.live_tx.receiver_count() != 0 { + let mut record = Vec::with_capacity(len); + record.extend_from_slice(&header); + record.extend_from_slice(payload); + let _ = self.live_tx.send(record.into()); + } + } + + fn dump_locked(&self, ring: &Ring) -> Vec { + let activations = self.activations(); + let mut dump = Vec::with_capacity(EVENT_DUMP_HEADER_LEN + ring.used); + dump.extend_from_slice(EVENT_DUMP_MAGIC); + dump.extend_from_slice(&(EVENT_DUMP_HEADER_LEN as u16).to_le_bytes()); + dump.extend_from_slice(&(blit_remote::events::EVENTS_VERSION as u16).to_le_bytes()); + dump.extend_from_slice(&(ring.capacity() as u64).to_le_bytes()); + dump.extend_from_slice(&(ring.used as u64).to_le_bytes()); + dump.extend_from_slice(&ring.records.to_le_bytes()); + dump.extend_from_slice(&ring.dropped.to_le_bytes()); + dump.extend_from_slice(&self.next_sequence.load(Ordering::Acquire).to_le_bytes()); + dump.extend_from_slice(&activations.to_bytes()); + debug_assert_eq!(dump.len(), EVENT_DUMP_HEADER_LEN); + for record in ring.record_vecs() { + dump.extend_from_slice(&record); + } + dump + } + + pub(crate) fn dump(&self) -> Vec { + let ring = self.ring.lock().expect("event ring poisoned"); + self.dump_locked(&ring) + } + + fn empty_dump_locked(&self, ring: &Ring) -> Vec { + let mut dump = Vec::with_capacity(EVENT_DUMP_HEADER_LEN); + dump.extend_from_slice(EVENT_DUMP_MAGIC); + dump.extend_from_slice(&(EVENT_DUMP_HEADER_LEN as u16).to_le_bytes()); + dump.extend_from_slice(&(blit_remote::events::EVENTS_VERSION as u16).to_le_bytes()); + dump.extend_from_slice(&(ring.capacity() as u64).to_le_bytes()); + dump.extend_from_slice(&0u64.to_le_bytes()); + dump.extend_from_slice(&0u64.to_le_bytes()); + dump.extend_from_slice(&ring.dropped.to_le_bytes()); + dump.extend_from_slice(&self.next_sequence.load(Ordering::Acquire).to_le_bytes()); + dump.extend_from_slice(&self.activations().to_bytes()); + debug_assert_eq!(dump.len(), EVENT_DUMP_HEADER_LEN); + dump + } + + pub(crate) fn snapshot_and_subscribe( + &self, + history: bool, + empty_header: bool, + ) -> EventSubscription { + let ring = self.ring.lock().expect("event ring poisoned"); + let receiver = self.live_tx.subscribe(); + let records = if history { ring.records } else { 0 }; + let dump = if history { + Some(self.dump_locked(&ring)) + } else if empty_header { + Some(self.empty_dump_locked(&ring)) + } else { + None + }; + (dump, records, receiver) + } + + fn allocate_stream_id(&self) -> u32 { + loop { + let id = self.next_stream_id.fetch_add(1, Ordering::Relaxed); + if id != 0 { + return id; + } + } + } + + pub(crate) fn client_stream(&self, history: bool) -> ClientEventStream { + let id = self.allocate_stream_id(); + let (dump, _, receiver) = self.snapshot_and_subscribe(history, true); + (id, dump, receiver) + } + + pub(crate) async fn start_file_stream( + self: &Arc, + path: &str, + flags: u8, + ) -> Result { + let append = flags & EVENTS_STREAM_APPEND != 0; + let history = flags & EVENTS_STREAM_HISTORY != 0; + let mut options = tokio::fs::OpenOptions::new(); + options + .create(true) + .write(true) + .append(append) + .truncate(!append); + let mut file = options + .open(path) + .await + .map_err(|error| StartFileStreamError::Io(format!("cannot open {path}: {error}")))?; + let id = self.allocate_stream_id(); + // Every recording invocation starts a self-describing segment, even + // when appending and/or starting from now. This is also the initial + // write whose flush gates the successful START response. + let (dump, history_records, mut receiver) = self.snapshot_and_subscribe(history, true); + let initial_bytes = dump.as_ref().map_or(0, Vec::len) as u64; + if let Some(dump) = dump { + file.write_all(&dump).await.map_err(|error| { + StartFileStreamError::Io(format!("cannot initialize {path}: {error}")) + })?; + file.flush().await.map_err(|error| { + StartFileStreamError::Io(format!("cannot flush initial data to {path}: {error}")) + })?; + } + let progress = FileStreamProgress::new(history_records, initial_bytes); + let task_progress = progress.clone(); + let task_path = path.to_owned(); + let (stop, mut stopped) = oneshot::channel(); + let task = tokio::spawn(async move { + let result = async { + loop { + let next = tokio::select! { + biased; + _ = &mut stopped => break, + next = receiver.recv() => next, + }; + match next { + Ok(record) => { + write_stream_bytes(&mut file, &record, 0, &task_progress, &task_path) + .await?; + } + Err(broadcast::error::RecvError::Lagged(lost)) => { + let record = gap_record(lost); + write_stream_bytes( + &mut file, + &record, + lost, + &task_progress, + &task_path, + ) + .await?; + } + Err(broadcast::error::RecvError::Closed) => break, + } + } + loop { + match receiver.try_recv() { + Ok(record) => { + write_stream_bytes(&mut file, &record, 0, &task_progress, &task_path) + .await?; + } + Err(broadcast::error::TryRecvError::Lagged(lost)) => { + write_stream_bytes( + &mut file, + &gap_record(lost), + lost, + &task_progress, + &task_path, + ) + .await?; + } + Err(broadcast::error::TryRecvError::Empty) + | Err(broadcast::error::TryRecvError::Closed) => break, + } + } + file.flush() + .await + .map_err(|error| format!("cannot flush {task_path}: {error}"))?; + Ok(()) + } + .await; + match result { + Ok(()) => task_progress.stop(), + Err(error) => { + return Err(task_progress.fail(error)); + } + } + Ok(()) + }); + self.file_streams + .lock() + .expect("event file streams poisoned") + .insert( + id, + FileStreamTask { + stop, + join: task, + path: path.to_owned(), + flags, + progress, + }, + ); + Ok(id) + } + + pub(crate) fn file_streams(&self) -> Vec { + let streams = self + .file_streams + .lock() + .expect("event file streams poisoned"); + let mut result = streams + .iter() + .map(|(&id, task)| FileStreamInfo { + id, + state: task.progress.state.load(Ordering::Acquire), + flags: task.flags, + records: task.progress.records.load(Ordering::Acquire), + bytes: task.progress.bytes.load(Ordering::Acquire), + lost: task.progress.lost.load(Ordering::Acquire), + path: task.path.clone(), + error: task + .progress + .error + .lock() + .expect("event file progress poisoned") + .clone() + .unwrap_or_default(), + }) + .collect::>(); + result.sort_unstable_by_key(|stream| stream.id); + result + } + + pub(crate) async fn stop_file_stream(&self, id: u32) -> Result { + let task = self + .file_streams + .lock() + .expect("event file streams poisoned") + .remove(&id); + if let Some(task) = task { + let _ = task.stop.send(()); + match task.join.await { + Ok(Ok(())) => Ok(true), + Ok(Err(error)) => Err(error), + Err(error) => Err(format!("event recording task failed: {error}")), + } + } else { + Ok(false) + } + } + + pub(crate) async fn shutdown_file_streams(&self) { + let tasks: Vec<_> = self + .file_streams + .lock() + .expect("event file streams poisoned") + .drain() + .map(|(_, task)| task) + .collect(); + let mut joins = Vec::with_capacity(tasks.len()); + for task in tasks { + let _ = task.stop.send(()); + joins.push(task.join); + } + for join in joins { + let _ = join.await; + } + } +} + +async fn write_stream_bytes( + file: &mut tokio::fs::File, + record: &[u8], + lost: u64, + progress: &FileStreamProgress, + path: &str, +) -> Result<(), String> { + file.write_all(record) + .await + .map_err(|error| format!("cannot write {path}: {error}"))?; + progress.records.fetch_add(1, Ordering::Relaxed); + progress + .bytes + .fetch_add(record.len() as u64, Ordering::Relaxed); + progress.lost.fetch_add(lost, Ordering::Relaxed); + Ok(()) +} + +fn gap_record(lost: u64) -> Vec { + let len = EVENT_RECORD_HEADER_LEN + 8; + let mut record = Vec::with_capacity(len); + record.extend_from_slice(&(len as u32).to_le_bytes()); + record.extend_from_slice(&EVENT_TYPE_STREAM_GAP.to_le_bytes()); + record.extend_from_slice(&0u16.to_le_bytes()); + record.extend_from_slice(&0u64.to_le_bytes()); + record.extend_from_slice(&0u64.to_le_bytes()); + record.extend_from_slice(&0u64.to_le_bytes()); + record.extend_from_slice(&lost.to_le_bytes()); + record +} + +pub(crate) fn payload_name(name: &str) -> Vec { + let bytes = name.as_bytes(); + let len = bytes.len().min(u16::MAX as usize); + let mut payload = Vec::with_capacity(2 + len); + payload.extend_from_slice(&(len as u16).to_le_bytes()); + payload.extend_from_slice(&bytes[..len]); + payload +} + +pub(crate) fn payload_client(client_id: u64) -> Vec { + client_id.to_le_bytes().to_vec() +} + +pub(crate) fn payload_pty_create_stage( + client_id: u64, + nonce: u16, + stage: u8, + status: u8, + pty_id: u16, +) -> [u8; 14] { + let mut payload = [0; 14]; + payload[..8].copy_from_slice(&client_id.to_le_bytes()); + payload[8..10].copy_from_slice(&nonce.to_le_bytes()); + payload[10] = stage; + payload[11] = status; + payload[12..14].copy_from_slice(&pty_id.to_le_bytes()); + payload +} + +pub(crate) fn payload_frame(client_id: u64, frame: &[u8]) -> Vec { + let mut payload = Vec::with_capacity(12 + frame.len()); + payload.extend_from_slice(&client_id.to_le_bytes()); + payload.extend_from_slice(&(frame.len().min(u32::MAX as usize) as u32).to_le_bytes()); + payload.extend_from_slice(&frame[..frame.len().min(u32::MAX as usize)]); + payload +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn ring_wrap_preserves_complete_newest_records() { + let log = EventLog::new(160, ActivationSet::all()); + for value in 0..8u8 { + log.record(EventType::Error, 0, &[value; 8]); + } + let dump = log.dump(); + assert_eq!(&dump[..8], EVENT_DUMP_MAGIC); + let records = &dump[EVENT_DUMP_HEADER_LEN..]; + assert_eq!(records.len() % (EVENT_RECORD_HEADER_LEN + 8), 0); + assert!(records.len() <= 160); + assert_eq!(records.last().copied(), Some(7)); + assert!(log.stats().dropped > 0); + } + + #[test] + fn disabled_event_building_can_be_guarded() { + let log = EventLog::new(4096, ActivationSet::default()); + assert!(!log.enabled(EventType::FrameRead)); + assert_eq!(log.stats().records, 0); + } + + #[test] + fn oversized_records_are_live_even_when_not_retained() { + let log = EventLog::new(64, ActivationSet::all()); + let (_, _, mut receiver) = log.client_stream(false); + log.record(EventType::Error, 0, &[7; 64]); + let record = receiver.try_recv().unwrap(); + assert_eq!(record.last().copied(), Some(7)); + assert_eq!(log.stats().records, 0); + assert_eq!(log.stats().dropped, 1); + } + + #[test] + fn shrinking_keeps_the_newest_records() { + let log = EventLog::new(MIN_RING_SIZE * 2, ActivationSet::all()); + for value in 0..100u8 { + log.record(EventType::Error, 0, &[value; 64]); + } + log.configure(MIN_RING_SIZE, ActivationSet::all(), None) + .unwrap(); + let dump = log.dump(); + assert_eq!(dump.last().copied(), Some(99)); + assert!(log.stats().used <= MIN_RING_SIZE); + } + + #[test] + fn configuration_revision_rejects_a_stale_replace() { + let log = EventLog::new(MIN_RING_SIZE, ActivationSet::low_throughput()); + assert_eq!(log.stats().revision, 1); + let changed = log + .configure(MIN_RING_SIZE * 2, ActivationSet::all(), Some(1)) + .unwrap(); + assert_eq!(changed.revision, 2); + assert!(matches!( + log.configure(MIN_RING_SIZE, ActivationSet::default(), Some(1)), + Err(ConfigureError::Conflict) + )); + let (current, activations) = log.configuration(); + assert_eq!(current.revision, 2); + assert_eq!(current.capacity, MIN_RING_SIZE * 2); + assert_eq!(activations, ActivationSet::all()); + } + + #[tokio::test] + async fn file_stream_flushes_history_and_live_records() { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let path = + std::env::temp_dir().join(format!("blit-events-{}-{unique}.bin", std::process::id())); + let path_text = path.to_str().unwrap(); + let log = EventLog::new(MIN_RING_SIZE, ActivationSet::all()); + log.record(EventType::ServerStart, 0, &[1]); + let stream_id = log + .start_file_stream(path_text, EVENTS_STREAM_HISTORY) + .await + .unwrap(); + let initialized = tokio::fs::read(&path).await.unwrap(); + assert_eq!(&initialized[..EVENT_DUMP_MAGIC.len()], EVENT_DUMP_MAGIC); + assert_eq!(initialized.last().copied(), Some(1)); + let streams = log.file_streams(); + assert_eq!(streams.len(), 1); + assert_eq!(streams[0].id, stream_id); + assert_eq!( + streams[0].state, + blit_remote::events::EVENTS_STREAM_STATE_RUNNING + ); + assert_eq!(streams[0].flags, EVENTS_STREAM_HISTORY); + assert_eq!(streams[0].path, path_text); + log.record(EventType::Error, 0, &[9]); + assert!(log.stop_file_stream(stream_id).await.unwrap()); + assert!(log.file_streams().is_empty()); + + let bytes = tokio::fs::read(&path).await.unwrap(); + tokio::fs::remove_file(&path).await.unwrap(); + assert_eq!(&bytes[..EVENT_DUMP_MAGIC.len()], EVENT_DUMP_MAGIC); + assert_eq!(bytes.last().copied(), Some(9)); + } + + #[tokio::test] + async fn from_now_file_stream_flushes_its_header_before_start_returns() { + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let path = std::env::temp_dir().join(format!( + "blit-events-now-{}-{unique}.bin", + std::process::id() + )); + let path_text = path.to_str().unwrap(); + let log = EventLog::new(MIN_RING_SIZE, ActivationSet::all()); + + let stream_id = log.start_file_stream(path_text, 0).await.unwrap(); + let initialized = tokio::fs::read(&path).await.unwrap(); + assert_eq!(initialized.len(), EVENT_DUMP_HEADER_LEN); + assert_eq!(&initialized[..EVENT_DUMP_MAGIC.len()], EVENT_DUMP_MAGIC); + assert!(log.stop_file_stream(stream_id).await.unwrap()); + + tokio::fs::remove_file(path).await.unwrap(); + } + + #[tokio::test] + async fn file_stream_status_and_stop_preserve_write_failure() { + let log = EventLog::new(MIN_RING_SIZE, ActivationSet::all()); + let (stop, stopped) = oneshot::channel(); + drop(stopped); + let progress = FileStreamProgress::new(7, 99); + progress.lost.store(2, Ordering::Relaxed); + progress.fail("disk full".into()); + let join = tokio::spawn(async { Err::<(), String>("disk full".into()) }); + log.file_streams.lock().unwrap().insert( + 42, + FileStreamTask { + stop, + join, + path: "/tmp/failed.bin".into(), + flags: EVENTS_STREAM_HISTORY, + progress, + }, + ); + + let streams = log.file_streams(); + assert_eq!(streams.len(), 1); + assert_eq!( + streams[0].state, + blit_remote::events::EVENTS_STREAM_STATE_FAILED + ); + assert_eq!(streams[0].records, 7); + assert_eq!(streams[0].bytes, 99); + assert_eq!(streams[0].lost, 2); + assert_eq!(streams[0].error, "disk full"); + assert_eq!(log.stop_file_stream(42).await, Err("disk full".into())); + assert!(log.file_streams().is_empty()); + } + + #[tokio::test] + async fn file_streams_have_no_process_wide_admission_cap() { + let log = EventLog::new(MIN_RING_SIZE, ActivationSet::all()); + let unique = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let mut streams = Vec::new(); + for index in 0..9 { + let path = std::env::temp_dir().join(format!( + "blit-events-uncapped-{}-{unique}-{index}.bin", + std::process::id() + )); + let id = log + .start_file_stream(path.to_str().unwrap(), 0) + .await + .unwrap(); + streams.push((id, path)); + } + assert_eq!(log.file_streams().len(), 9); + for (id, path) in streams { + assert!(log.stop_file_stream(id).await.unwrap()); + tokio::fs::remove_file(path).await.unwrap(); + } + } +} diff --git a/crates/server/src/extension/mod.rs b/crates/server/src/extension/mod.rs index 5d2e1a5b..2b45495e 100644 --- a/crates/server/src/extension/mod.rs +++ b/crates/server/src/extension/mod.rs @@ -5848,6 +5848,10 @@ mod tests { allow_forward_insecure: false, allow_persistent_extensions: true, }, + events: super::super::events::EventLog::new( + super::super::events::DEFAULT_RING_SIZE, + blit_remote::events::ActivationSet::default(), + ), #[cfg(any(unix, windows))] process_server: super::super::process::Server::new(false, false), boot_generation, diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 6d9300e9..7b1afc37 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -12,6 +12,7 @@ use blit_remote::desktop::{ msg_notification_update, msg_tray_menu, msg_tray_snapshot, msg_tray_update, parse_desktop_subscribe, parse_notification_event, parse_tray_event, }; +use blit_remote::events::EventType; #[cfg(target_os = "linux")] use blit_remote::media::{ ACTIVE_CAMERA, ACTIVE_MICROPHONE, ACTIVE_SCREENCAST, C2S_MEDIA_CONTROL, C2S_MEDIA_DATA, @@ -39,14 +40,15 @@ use blit_remote::{ FEATURE_CREATE_NO_SUBSCRIBE, FEATURE_CREATE_NONCE, FEATURE_CREATE_STATUS, FEATURE_KILL_MODE, FEATURE_PTY_DEADLINE, FEATURE_RESIZE_BATCH, FEATURE_RESTART, FEATURE_SCROLL_BY, FrameState, KICK_REASON_MAX, KILL_LEADER_ONLY, READ_ANSI, READ_TAIL, REMOTE_INPUT_POINTER, - REMOTE_INPUT_TOUCH, S2C_CLOSED, S2C_CREATED, S2C_CREATED_N, S2C_LIST, S2C_PING, S2C_QUIT, - S2C_READY, S2C_SEARCH_RESULTS, S2C_SURFACE_CAPTURE, S2C_SURFACE_LIST, S2C_TEXT, S2C_TITLE, - STATUS_BUDGET, STATUS_INVALID, STATUS_NOT_FOUND, STATUS_OK, STATUS_OTHER, STATUS_TOO_LARGE, - SURFACE_FRAME_CODEC_H264, SURFACE_FRAME_FLAG_KEYFRAME, SURFACE_POINTER_AXIS2_LEN, - SURFACE_POINTER_DOWN, SURFACE_POINTER_LEAVE, SURFACE_POINTER_MOVE, SURFACE_POINTER_UP, - SURFACE_TOUCH_CANCEL, SURFACE_TOUCH_DISABLE, SURFACE_TOUCH_DOWN, SURFACE_TOUCH_ENABLE, - SURFACE_TOUCH_MOTION, SURFACE_TOUCH_UP, build_update_msg, clamp_cursor_rect, msg_copy_failed, - msg_hello, msg_kick_result, msg_kicked, msg_s2c_client_list, msg_s2c_client_list2, + REMOTE_INPUT_TOUCH, S2C_CLOSED, S2C_CREATE_FAILED, S2C_CREATED, S2C_CREATED_N, S2C_LIST, + S2C_PING, S2C_QUIT, S2C_READY, S2C_SEARCH_RESULTS, S2C_SURFACE_CAPTURE, S2C_SURFACE_LIST, + S2C_TEXT, S2C_TITLE, STATUS_BUDGET, STATUS_CONFLICT, STATUS_INVALID, STATUS_NOT_FOUND, + STATUS_OK, STATUS_OTHER, STATUS_TOO_LARGE, SURFACE_FRAME_CODEC_H264, + SURFACE_FRAME_FLAG_KEYFRAME, SURFACE_POINTER_AXIS2_LEN, SURFACE_POINTER_DOWN, + SURFACE_POINTER_LEAVE, SURFACE_POINTER_MOVE, SURFACE_POINTER_UP, SURFACE_TOUCH_CANCEL, + SURFACE_TOUCH_DISABLE, SURFACE_TOUCH_DOWN, SURFACE_TOUCH_ENABLE, SURFACE_TOUCH_MOTION, + SURFACE_TOUCH_UP, build_update_msg, clamp_cursor_rect, msg_copy_failed, msg_hello, + msg_kick_result, msg_kicked, msg_s2c_client_list, msg_s2c_client_list2, msg_s2c_clipboard_content, msg_s2c_clipboard_list, msg_s2c_clipboard_owner, msg_s2c_scroll_offset, msg_s2c_surface_remote_input, msg_s2c_used_rows, msg_surface_activated, msg_surface_app_id, msg_surface_created, msg_surface_destroyed, msg_surface_encoder, @@ -78,6 +80,7 @@ mod capacity_diagnostics; mod channel; #[cfg(target_os = "linux")] mod desktop_bus; +mod events; mod extension; pub mod extension_catalog; mod extension_jobs; @@ -124,6 +127,31 @@ pub use surface_encoder::{SurfaceBandwidth, SurfaceEncoding, SurfaceSpeed}; type PtyFds = Arc>>; +/// Conditional event emission. `$payload` is evaluated only while the event's +/// activation bit is set, which is the invariant that keeps full byte capture +/// free when operators leave the low-throughput default enabled. +macro_rules! blit_event { + ($log:expr, $kind:expr) => {{ + let log = &$log; + let kind = $kind; + if log.enabled(kind) { + log.record(kind, 0, &[]); + } + }}; + ($log:expr, $kind:expr, $payload:expr) => {{ + let log = &$log; + let kind = $kind; + if log.enabled(kind) { + let payload = $payload; + log.record(kind, 0, &payload); + } + }}; +} + +tokio::task_local! { + static EVENT_WRITE_CONTEXT: (Arc, Arc); +} + /// Command-line overrides for extension and channel deployment policy. /// /// The CLI fills this without mutating the process environment. Calling @@ -1584,6 +1612,77 @@ async fn write_frame_counted_with_timeout( ) -> Result<(), FrameWriteError> { write_frame_with_timeout(writer, payload, no_progress_timeout).await?; outbound_bytes.fetch_add((payload.len() as u64).saturating_add(4), Ordering::Relaxed); + // A live event-record envelope is deliberately excluded: recording its + // delivery would create another record to deliver forever. Other event + // control/results, and physical fragments of large packets, stay visible. + let recursive_event_record = payload.first() == Some(&blit_remote::events::EVENTS) + && payload.get(2) == Some(&blit_remote::events::EVENTS_RECORD); + if !recursive_event_record { + let _ = EVENT_WRITE_CONTEXT.try_with(|(log, client_id)| { + let client_id = client_id.load(Ordering::Relaxed); + match payload.first().copied() { + Some(S2C_CREATED_N) if payload.len() >= 5 => { + let nonce = u16::from_le_bytes([payload[1], payload[2]]); + let pty_id = u16::from_le_bytes([payload[3], payload[4]]); + blit_event!( + log, + EventType::PtyCreate, + events::payload_pty_create_stage( + client_id, + nonce, + blit_remote::events::PTY_CREATE_REPLY_WRITTEN, + STATUS_OK, + pty_id, + ) + ); + } + Some(S2C_CREATE_FAILED) if payload.len() >= 4 => { + let nonce = u16::from_le_bytes([payload[1], payload[2]]); + blit_event!( + log, + EventType::PtyCreate, + events::payload_pty_create_stage( + client_id, + nonce, + blit_remote::events::PTY_CREATE_REPLY_WRITTEN, + payload[3], + 0, + ) + ); + } + _ => {} + } + blit_event!( + log, + EventType::FrameWrite, + events::payload_frame(client_id, payload,) + ); + blit_event!(log, EventType::MessageWrite, { + let mut summary = events::payload_client(client_id); + summary.push(payload.first().copied().unwrap_or(0)); + summary.extend_from_slice(&(payload.len() as u32).to_le_bytes()); + summary + }); + blit_event!(log, EventType::OutboxQueue, { + let mut summary = events::payload_client(client_id); + summary.extend_from_slice(&(payload.len() as u32).to_le_bytes()); + summary + }); + match payload.first().copied() { + Some(blit_remote::S2C_SURFACE_FRAME) => blit_event!( + log, + EventType::SurfaceFrame, + events::payload_frame(client_id, payload) + ), + Some(blit_remote::S2C_AUDIO_FRAME) => blit_event!( + log, + EventType::AudioFrame, + events::payload_frame(client_id, payload) + ), + _ => {} + } + }); + } Ok(()) } @@ -8164,6 +8263,7 @@ impl Session { struct AppStateInner { config: Config, + events: Arc, #[cfg(any(unix, windows))] process_server: process::Server, /// Opaque identifier shared by every connection to this server process. @@ -8242,6 +8342,7 @@ fn new_boot_generation() -> u64 { } fn nudge_delivery(state: &AppState) { + blit_event!(state.events, EventType::TickNudge); state.delivery_notify.notify_one(); } @@ -9101,6 +9202,7 @@ fn resolve_term_cwd(osc7: Option<&str>, kernel: impl FnOnce() -> Option) /// no-op for them — a server must not send `S2C_CREATE_FAILED` to a client /// that did not ask for it (docs/protocol.md, "Common status registry"). fn refuse_create( + event_log: &events::EventLog, sess: &Session, client_id: u64, want_status: bool, @@ -9108,6 +9210,17 @@ fn refuse_create( status: u8, detail: &str, ) { + blit_event!( + event_log, + EventType::PtyCreate, + events::payload_pty_create_stage( + client_id, + nonce, + blit_remote::events::PTY_CREATE_REFUSED, + status, + 0, + ) + ); if !want_status { return; } @@ -9418,6 +9531,11 @@ async fn evict_exited(state: &AppState) { // Already exited by construction, so the fd and the child are gone; // this is only dropping the retained terminal state. drop(pty); + blit_event!( + state.events, + EventType::PtyRemove, + id.to_le_bytes().to_vec() + ); state.pty_fds.write().unwrap().remove(&id); for client in sess.clients.values_mut() { unsubscribe_client_from(client, id); @@ -9451,17 +9569,27 @@ fn stop_grace() -> Duration { async fn enforce_deadlines(state: &AppState) { let now = Instant::now(); let mut sess = state.session.lock().await; - for pty in sess.ptys.values_mut() { + for (&pty_id, pty) in &mut sess.ptys { if pty.exited { continue; } if pty.stop_deadline.is_some_and(|d| now >= d) { pty.stop_deadline = None; + blit_event!(state.events, EventType::Deadline, { + let mut payload = pty_id.to_le_bytes().to_vec(); + payload.push(2); + payload + }); pty::kill_pty(&pty.handle, SIGKILL, true); } else if pty.deadline.is_some_and(|d| now >= d) { pty.deadline = None; pty.exit_reason = blit_remote::EXIT_REASON_DEADLINE; pty.stop_deadline = Some(now + stop_grace()); + blit_event!(state.events, EventType::Deadline, { + let mut payload = pty_id.to_le_bytes().to_vec(); + payload.push(1); + payload + }); pty::kill_pty(&pty.handle, SIGTERM, true); } } @@ -9472,6 +9600,8 @@ async fn enforce_deadlines(state: &AppState) { /// Ordered reader EOF normally finalizes first. A descendant can keep the slave open, so the /// deadline forces one unpaced drain before cleanup instead of waiting forever. async fn supervise(state: &AppState) { + let pass_started = Instant::now(); + blit_event!(state.events, EventType::Supervisor, vec![1]); let now = Instant::now(); let fallback_due = { let mut sess = state.session.lock().await; @@ -9690,6 +9820,13 @@ async fn supervise(state: &AppState) { } } evict_exited(state).await; + blit_event!(state.events, EventType::Supervisor, { + let mut payload = vec![2]; + payload.extend_from_slice( + &(pass_started.elapsed().as_nanos().min(u64::MAX as u128) as u64).to_le_bytes(), + ); + payload + }); } /// Run a terminal's exit path. @@ -9714,6 +9851,12 @@ async fn cleanup_pty_internal(pty_id: u16, generation: Option, state: &AppS pty.stop_deadline = None; pty::close_pty(&pty.handle); pty.exit_status = pty::collect_exit_status(&pty.handle); + blit_event!(state.events, EventType::PtyExit, { + let mut payload = pty_id.to_le_bytes().to_vec(); + payload.extend_from_slice(&pty.exit_status.to_le_bytes()); + payload.push(pty.exit_reason); + payload + }); pty.mark_dirty(); // A command still running when the shell dies never gets its `D` // marker; closing it here is what stops a waiter hanging until its @@ -9830,6 +9973,12 @@ pub async fn run(config: Config) { // environment now, before any feature mask or service is constructed. let _ = ensure_deployment_settings(); kv::configure_server_name(&config.name); + let (event_log, startup_event_file) = events::EventLog::from_env(); + blit_event!(event_log, EventType::ServerStart, { + let mut payload = events::payload_name(env!("CARGO_PKG_VERSION")); + payload.extend_from_slice(&events::payload_name(config.name.as_str())); + payload + }); #[cfg(any(unix, windows))] let process_server = process::Server::new(config.verbose, config.processes); let boot_generation = new_boot_generation(); @@ -9852,6 +10001,7 @@ pub async fn run(config: Config) { extension::ExtensionService::from_env(config.allow_persistent_extensions, &config.name); let state: AppState = Arc::new(AppStateInner { config, + events: event_log.clone(), #[cfg(any(unix, windows))] process_server, boot_generation, @@ -9865,6 +10015,23 @@ pub async fn run(config: Config) { extension_jobs: extension_jobs::GlobalTracker::from_env(), extensions: extensions.clone(), }); + if let Some(file) = startup_event_file { + match event_log.start_file_stream(&file.path, file.flags).await { + Ok(stream_id) => blit_event!(event_log, EventType::StreamStart, { + let mut payload = stream_id.to_le_bytes().to_vec(); + payload.extend_from_slice(&events::payload_name(&file.path)); + payload + }), + Err(error) => { + eprintln!("blit-server: event file stream: {error}"); + blit_event!( + event_log, + EventType::Error, + events::payload_name(&error.to_string()) + ); + } + } + } extensions.restore(state.clone()).await; // Start the compositor eagerly so it is ready before any client @@ -9892,6 +10059,11 @@ pub async fn run(config: Config) { .filter(|&v| v > 0) .map(Duration::from_micros); tokio::spawn(async move { + blit_event!( + delivery_state.events, + EventType::TaskStart, + events::payload_name("delivery") + ); let mut next_deadline: Option = None; let mut last_tick: Option = None; loop { @@ -9917,6 +10089,11 @@ pub async fn run(config: Config) { let supervisor_state = state.clone(); tokio::spawn(async move { + blit_event!( + supervisor_state.events, + EventType::TaskStart, + events::payload_name("supervisor") + ); supervisor_loop(supervisor_state).await; }); @@ -9984,6 +10161,8 @@ pub async fn run(config: Config) { state.extensions.shutdown().await; state.process_server.shutdown().await; state.connections.wait_empty().await; + blit_event!(state.events, EventType::ServerStop); + state.events.shutdown_file_streams().await; return; } @@ -10020,6 +10199,8 @@ pub async fn run(config: Config) { state.process_server.shutdown().await; state.extensions.shutdown().await; state.connections.wait_empty().await; + blit_event!(state.events, EventType::ServerStop); + state.events.shutdown_file_streams().await; } /// Minimum interval between blanket RequestFrame rounds. Keeps video @@ -10050,7 +10231,17 @@ fn blanket_frame_interval(sess: &Session) -> Option { } async fn tick(state: &AppState) -> TickOutcome { + let tick_started = Instant::now(); + blit_event!(state.events, EventType::TickStart); + let lock_started = Instant::now(); let mut sess = state.session.lock().await; + blit_event!(state.events, EventType::SessionLock, { + let mut payload = events::payload_name("tick"); + payload.extend_from_slice( + &(lock_started.elapsed().as_nanos().min(u64::MAX as u128) as u64).to_le_bytes(), + ); + payload + }); sess.tick_fires += 1; let mut next_deadline: Option = None; let now = Instant::now(); @@ -10168,6 +10359,16 @@ async fn tick(state: &AppState) -> TickOutcome { width, height, } => { + blit_event!(state.events, EventType::CompositorEvent, { + let mut payload = vec![1]; + payload.extend_from_slice(&surface_id.to_le_bytes()); + payload.extend_from_slice(&parent_id.to_le_bytes()); + payload.extend_from_slice(&width.to_le_bytes()); + payload.extend_from_slice(&height.to_le_bytes()); + payload.extend_from_slice(&events::payload_name(&title)); + payload.extend_from_slice(&events::payload_name(&app_id)); + payload + }); broadcast.push(msg_surface_created( surface_id, parent_id, width, height, &title, &app_id, )); @@ -10196,6 +10397,11 @@ async fn tick(state: &AppState) -> TickOutcome { invalidate_client_encoders.push(surface_id); } CompositorEvent::SurfaceDestroyed { surface_id } => { + blit_event!(state.events, EventType::CompositorEvent, { + let mut payload = vec![2]; + payload.extend_from_slice(&surface_id.to_le_bytes()); + payload + }); #[cfg(target_os = "linux")] { let retired = retire_screencast_surface(cs, surface_id); @@ -10229,6 +10435,16 @@ async fn tick(state: &AppState) -> TickOutcome { timestamp_sub_us, encoder_skip, } => { + blit_event!(state.events, EventType::CompositorEvent, { + let mut payload = vec![3]; + payload.extend_from_slice(&surface_id.to_le_bytes()); + payload.extend_from_slice(&width.to_le_bytes()); + payload.extend_from_slice(&height.to_le_bytes()); + payload.extend_from_slice(×tamp_ms.to_le_bytes()); + payload.extend_from_slice(×tamp_sub_us.to_le_bytes()); + payload.push(encoder_skip as u8); + payload + }); surface_commit_count += 1; #[cfg(target_os = "linux")] let screencast_frame = { @@ -10301,6 +10517,16 @@ async fn tick(state: &AppState) -> TickOutcome { timestamp_ms, timestamp_sub_us, } => { + blit_event!(state.events, EventType::SurfaceEncode, { + let mut payload = frame.surface_id.to_le_bytes().to_vec(); + payload.extend_from_slice(&frame.client_id.to_le_bytes()); + payload.extend_from_slice(&frame.width.to_le_bytes()); + payload.extend_from_slice(&frame.height.to_le_bytes()); + payload.extend_from_slice(&(frame.data.len() as u32).to_le_bytes()); + payload.push(frame.codec_flag); + payload.push(frame.is_keyframe as u8); + payload + }); surface_commit_count += 1; cs.pixel_generation += 1; let new_is_keyframe = frame.is_keyframe; @@ -13441,21 +13667,58 @@ async fn tick(state: &AppState) -> TickOutcome { }; match input { PtyInput::Data(data) => { + blit_event!(state.events, EventType::PtyRead, { + let mut payload = id.to_le_bytes().to_vec(); + payload.extend_from_slice(&(data.len() as u32).to_le_bytes()); + payload.extend_from_slice(&data); + payload + }); charge_pty_parse_budgets(&mut budget, &mut session_budget, data.len()); + let parse_started = Instant::now(); if let Some(msg) = feed_pty_chunk(pty, id, &data) { cwd_msgs.push(msg); } + blit_event!(state.events, EventType::PtyParse, { + let mut payload = id.to_le_bytes().to_vec(); + payload.extend_from_slice(&(data.len() as u32).to_le_bytes()); + payload.extend_from_slice( + &(parse_started.elapsed().as_nanos().min(u64::MAX as u128) as u64) + .to_le_bytes(), + ); + payload + }); pty.mark_output_dirty(now, output_coalesce_cap); } PtyInput::SyncBoundary { before } => { + blit_event!(state.events, EventType::PtyRead, { + let mut payload = id.to_le_bytes().to_vec(); + payload.extend_from_slice(&(before.len() as u32).to_le_bytes()); + payload.extend_from_slice(&before); + payload + }); charge_pty_parse_budgets(&mut budget, &mut session_budget, before.len()); if !before.is_empty() { + let parse_started = Instant::now(); if let Some(msg) = feed_pty_chunk(pty, id, &before) { cwd_msgs.push(msg); } + blit_event!(state.events, EventType::PtyParse, { + let mut payload = id.to_le_bytes().to_vec(); + payload.extend_from_slice(&(before.len() as u32).to_le_bytes()); + payload.extend_from_slice( + &(parse_started.elapsed().as_nanos().min(u64::MAX as u128) as u64) + .to_le_bytes(), + ); + payload + }); pty.mark_output_dirty(now, output_coalesce_cap); } if !pty.driver.synced_output() { + blit_event!( + state.events, + EventType::PtySnapshot, + id.to_le_bytes().to_vec() + ); let frame = take_snapshot(pty); enqueue_ready_frame(&mut pty.ready_frames, frame); pty.clear_dirty(); @@ -13578,6 +13841,11 @@ async fn tick(state: &AppState) -> TickOutcome { // it, ordinary shell output is immediate and alternate-screen output // reaches here after either a short quiet period or the display-rate // liveness ceiling. + blit_event!( + state.events, + EventType::PtySnapshot, + id.to_le_bytes().to_vec() + ); snapshots.insert(id, take_snapshot(pty)); pty.clear_dirty(); sess.tick_snaps += 1; @@ -14106,6 +14374,14 @@ async fn tick(state: &AppState) -> TickOutcome { next_deadline = Some(next_deadline.map_or(blanket_deadline, |d| d.min(blanket_deadline))); } + blit_event!(state.events, EventType::TickStop, { + let mut payload = (tick_started.elapsed().as_nanos().min(u64::MAX as u128) as u64) + .to_le_bytes() + .to_vec(); + payload.extend_from_slice(&(sess.clients.len() as u32).to_le_bytes()); + payload.extend_from_slice(&(sess.ptys.len() as u32).to_le_bytes()); + payload + }); TickOutcome { next_deadline } } @@ -17600,6 +17876,350 @@ async fn deliver_terminal_notice( tokio::time::timeout(timeout, sent_rx).await.is_ok() } +fn protocol_event_type(opcode: u8) -> Option { + use blit_remote::events::EVENTS; + match opcode { + blit_remote::fs::C2S_FS_SYNC..=blit_remote::fs::C2S_FS_READ => Some(EventType::FsRequest), + blit_remote::lsp::C2S_LSP_OPEN..=blit_remote::lsp::C2S_LSP_BUFFER => { + Some(EventType::LspRequest) + } + blit_remote::kv::C2S_KV_OPEN..=blit_remote::env::C2S_ENV_GET => Some(EventType::KvRequest), + blit_remote::net::C2S_NET_OPEN..=blit_remote::net::C2S_NET_DGRAM => { + Some(EventType::NetRequest) + } + blit_remote::extension::EXT_RUN..=blit_remote::extension::EXT_COMMAND => { + Some(EventType::ExtensionRequest) + } + blit_remote::channel::CHANNEL => Some(EventType::ChannelRequest), + blit_remote::git::C2S_GIT_OPEN..=blit_remote::git::C2S_GIT_WORKTREES => { + Some(EventType::GitRequest) + } + blit_remote::process::C2S_PROCESS_SPAWN..=blit_remote::process::C2S_PROCESS_WATCH => { + Some(EventType::ProcessRequest) + } + C2S_CLIENT_LIST | C2S_CLIENT_WATCH | C2S_CLIENT_UNWATCH | C2S_KICK | EVENTS => { + Some(EventType::ClientControl) + } + C2S_SURFACE_INPUT..=blit_remote::media::C2S_MEDIA_DATA => { + Some(EventType::CompositorCommand) + } + _ => None, + } +} + +fn events_config_message(state: &AppState, nonce: u16) -> Vec { + let (stats, activations) = state.events.configuration(); + blit_remote::events::msg_events_config( + nonce, + blit_remote::events::EventConfig { + revision: stats.revision, + size: stats.capacity as u64, + used: stats.used as u64, + records: stats.records, + dropped: stats.dropped, + next_sequence: stats.next_sequence, + activations, + }, + ) +} + +async fn handle_events_message( + data: &[u8], + state: &AppState, + out: &TrackedOutboxSender, + client_streams: &mut HashMap>, + client_id: u64, +) { + use blit_remote::events::{ + EVENTS_STREAM_HISTORY, EVENTS_TARGET_CLIENT, EventStreamInfo, EventsRequest, + msg_events_dumped, msg_events_records, msg_events_result, msg_events_stream_gap, + msg_events_stream_started, msg_events_stream_stopped, msg_events_streams, + parse_events_request, + }; + let request = match parse_events_request(data) { + Ok(request) => request, + Err(error) => { + blit_event!( + state.events, + EventType::ProtocolError, + events::payload_name(&error.to_string()) + ); + if data.len() >= 5 { + let nonce = u16::from_le_bytes([data[3], data[4]]); + let _ = out.send(msg_events_result( + nonce, + STATUS_INVALID, + 0, + &error.to_string(), + )); + } + return; + } + }; + match request { + EventsRequest::ConfigGet { nonce } => { + let _ = out.send(events_config_message(state, nonce)); + } + EventsRequest::ConfigSet { + nonce, + expected_revision, + size, + activations, + } => { + let Ok(size) = usize::try_from(size) else { + let _ = out.send(msg_events_result( + nonce, + STATUS_TOO_LARGE, + 0, + "event ring size does not fit this host", + )); + return; + }; + let log = state.events.clone(); + let expected_revision = (expected_revision + != blit_remote::events::EVENTS_CONFIG_REVISION_ANY) + .then_some(expected_revision); + let result = tokio::task::spawn_blocking(move || { + log.configure(size, activations, expected_revision) + }) + .await; + match result { + Ok(Ok(stats)) => { + blit_event!(state.events, EventType::ConfigChange, { + let mut payload = events::payload_client(client_id); + payload.extend_from_slice(&(size as u64).to_le_bytes()); + payload.extend_from_slice(&activations.to_bytes()); + payload + }); + // Reply with the revision produced by this replacement, + // even if another update races with response delivery. + // A temporary capture can then safely use this exact + // revision as the condition for restoring its settings. + let _ = out.send(blit_remote::events::msg_events_config( + nonce, + blit_remote::events::EventConfig { + revision: stats.revision, + size: stats.capacity as u64, + used: stats.used as u64, + records: stats.records, + dropped: stats.dropped, + next_sequence: stats.next_sequence, + activations, + }, + )); + } + Ok(Err(error)) => { + let status = if error == events::ConfigureError::Conflict { + STATUS_CONFLICT + } else { + STATUS_INVALID + }; + let _ = out.send(msg_events_result(nonce, status, 0, &error.to_string())); + } + Err(_) => { + let _ = out.send(msg_events_result( + nonce, + STATUS_OTHER, + 0, + "event configuration task failed", + )); + } + } + } + EventsRequest::Dump { nonce } => { + let log = state.events.clone(); + let out = out.clone(); + tokio::spawn(async move { + blit_event!( + log, + EventType::TaskStart, + events::payload_name("events.dump") + ); + let dump_log = log.clone(); + if let Ok(dump) = tokio::task::spawn_blocking(move || dump_log.dump()).await { + let _ = out.send(msg_events_dumped(nonce, &dump)); + } + blit_event!( + log, + EventType::TaskStop, + events::payload_name("events.dump") + ); + }); + } + EventsRequest::StreamStart { + nonce, + target, + flags, + path: _, + } if target == EVENTS_TARGET_CLIENT => { + client_streams.retain(|_, task| !task.is_finished()); + let history = flags & EVENTS_STREAM_HISTORY != 0; + let (stream_id, dump, mut receiver) = state.events.client_stream(history); + let _ = out.send(msg_events_stream_started(nonce, STATUS_OK, stream_id, "")); + let out = out.clone(); + let log = state.events.clone(); + let task = tokio::spawn(async move { + blit_event!( + log, + EventType::TaskStart, + events::payload_name("events.client-stream") + ); + if let Some(dump) = dump + && out.send(msg_events_dumped(nonce, &dump)).is_err() + { + blit_event!( + log, + EventType::TaskStop, + events::payload_name("events.client-stream") + ); + return; + } + let mut batch = Vec::with_capacity(64); + let mut pending = None; + loop { + batch.clear(); + let first = if let Some(record) = pending.take() { + Ok(record) + } else { + receiver.recv().await + }; + match first { + Ok(record) => batch.push(record), + Err(tokio::sync::broadcast::error::RecvError::Lagged(lost)) => { + if out.send(msg_events_stream_gap(stream_id, lost)).is_err() { + break; + } + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + let mut bytes = batch[0].len(); + let mut lost = 0u64; + while batch.len() < u16::MAX as usize { + match receiver.try_recv() { + Ok(record) + if bytes.saturating_add(record.len()) + <= events::LIVE_BATCH_MAX_BYTES => + { + bytes += record.len(); + batch.push(record); + } + Ok(record) => { + pending = Some(record); + break; + } + Err(tokio::sync::broadcast::error::TryRecvError::Lagged(count)) => { + lost = lost.saturating_add(count); + break; + } + Err(tokio::sync::broadcast::error::TryRecvError::Empty) + | Err(tokio::sync::broadcast::error::TryRecvError::Closed) => break, + } + } + if out.send(msg_events_records(stream_id, &batch)).is_err() { + break; + } + if lost != 0 && out.send(msg_events_stream_gap(stream_id, lost)).is_err() { + break; + } + } + let _ = out.send(msg_events_stream_stopped(stream_id, STATUS_OK, "")); + blit_event!( + log, + EventType::TaskStop, + events::payload_name("events.client-stream") + ); + }); + client_streams.insert(stream_id, task); + blit_event!(state.events, EventType::StreamStart, { + let mut payload = events::payload_client(client_id); + payload.extend_from_slice(&stream_id.to_le_bytes()); + payload.push(EVENTS_TARGET_CLIENT); + payload + }); + } + EventsRequest::StreamStart { + nonce, + target: _, + flags, + path, + } => match state.events.start_file_stream(path, flags).await { + Ok(stream_id) => { + let _ = out.send(msg_events_stream_started(nonce, STATUS_OK, stream_id, "")); + blit_event!(state.events, EventType::StreamStart, { + let mut payload = events::payload_client(client_id); + payload.extend_from_slice(&stream_id.to_le_bytes()); + payload.push(blit_remote::events::EVENTS_TARGET_FILE); + payload.extend_from_slice(&events::payload_name(path)); + payload + }); + } + Err(error) => { + let _ = out.send(msg_events_stream_started( + nonce, + STATUS_OTHER, + 0, + &error.to_string(), + )); + } + }, + EventsRequest::StreamStop { nonce, stream_id } => { + let stopped_client = if let Some(task) = client_streams.remove(&stream_id) { + let was_running = !task.is_finished(); + task.abort(); + let _ = task.await; + if was_running { + blit_event!( + state.events, + EventType::TaskStop, + events::payload_name("events.client-stream") + ); + } + true + } else { + false + }; + let file_result = if stopped_client { + Ok(false) + } else { + state.events.stop_file_stream(stream_id).await + }; + let (stopped, status, detail) = match file_result { + Ok(stopped_file) if stopped_client || stopped_file => { + (true, STATUS_OK, String::new()) + } + Ok(_) => (false, STATUS_NOT_FOUND, "event stream not found".to_owned()), + Err(error) => (true, STATUS_OTHER, error), + }; + let _ = out.send(msg_events_result(nonce, status, stream_id, &detail)); + if stopped { + blit_event!(state.events, EventType::StreamStop, { + let mut payload = events::payload_client(client_id); + payload.extend_from_slice(&stream_id.to_le_bytes()); + payload + }); + } + } + EventsRequest::StreamList { nonce } => { + let streams = state.events.file_streams(); + let wire = streams + .iter() + .map(|stream| EventStreamInfo { + stream_id: stream.id, + state: stream.state, + flags: stream.flags, + records: stream.records, + bytes: stream.bytes, + lost: stream.lost, + path: &stream.path, + error: &stream.error, + }) + .collect::>(); + let _ = out.send(msg_events_streams(nonce, &wire)); + } + } +} + /// Connection-local mirror of this client's auxiliary subscription tables. /// /// `sync` runs after *every* filesystem, git, LSP, KV and network message, so @@ -17669,6 +18289,11 @@ pub(crate) fn spawn_network_client bool { let options = ConnectionOptions::network(); let Some(registration) = state.connections.register(options.cancellation.clone()) else { + blit_event!( + state.events, + EventType::ClientReject, + events::payload_name("shutdown") + ); return false; }; let max = state.config.max_connections; @@ -17680,11 +18305,32 @@ pub(crate) fn spawn_network_client> = HashMap::new(); // BLIT_NET=0 turns the relay off: the bit is unadvertised AND every // NET_OPEN is refused with PERMISSION, so a client that ignores // feature bits still gets its one reply. @@ -17840,7 +18488,15 @@ async fn handle_client_registered {} @@ -18124,7 +18780,16 @@ async fn handle_client_registered= 9 { let pid = u16::from_le_bytes([data[1], data[2]]); + blit_event!(state.events, EventType::PtyInput, { + let mut payload = events::payload_client(client_id); + payload.extend_from_slice(&pid.to_le_bytes()); + payload.extend_from_slice(&data[3..]); + payload + }); let type_ = data[3]; let button = data[4]; let col = u16::from_le_bytes([data[5], data[6]]); @@ -19245,6 +19947,12 @@ async fn handle_client_registered= 3 { let pid = u16::from_le_bytes([data[1], data[2]]); + blit_event!(state.events, EventType::PtyInput, { + let mut payload = events::payload_client(client_id); + payload.extend_from_slice(&pid.to_le_bytes()); + payload.extend_from_slice(&data[3..]); + payload + }); let mut need_nudge = false; { let mut sess = state.session.lock().await; @@ -19270,6 +19984,13 @@ async fn handle_client_registered= 7 => { + blit_event!(state.events, EventType::PtyResize, { + let mut payload = events::payload_client(client_id); + payload.extend_from_slice(&data[1..]); + payload + }); let entries = data[1..].chunks_exact(6); if !entries.remainder().is_empty() { continue; @@ -20093,6 +20852,11 @@ async fn handle_client_registered { if let Some(nonce) = err.nonce { refuse_create( + &state.events, &sess, client_id, err.want_status, @@ -20377,6 +21185,7 @@ async fn handle_client_registered = None; - if let Some(pty) = pty::spawn_pty( + blit_event!( + state.events, + EventType::PtyCreate, + events::payload_pty_create_stage( + client_id, + nonce, + blit_remote::events::PTY_CREATE_SPAWN_BEGIN, + STATUS_OK, + id, + ) + ); + let spawned = pty::spawn_pty( &config.shell, &config.shell_flags, rows, @@ -20464,7 +21298,23 @@ async fn handle_client_registered= 3 => { let pid = u16::from_le_bytes([data[1], data[2]]); if let Some(pty) = sess.ptys.remove(&pid) { + blit_event!(state.events, EventType::PtyRemove, { + let mut payload = pid.to_le_bytes().to_vec(); + payload.push(1); + payload + }); if !pty.exited { state.pty_fds.write().unwrap().remove(&pid); drop(pty.reader_handle); @@ -21972,6 +22844,22 @@ async fn handle_client_registered { + let features = u32::from_le_bytes(frame[3..7].try_into().unwrap()); + advertised = features & blit_remote::events::FEATURE_EVENTS != 0; + } + Some(S2C_READY) => break, + _ => {} + } + } + assert!(advertised); + + assert!( + write_frame( + &mut client, + &blit_remote::events::msg_events_stream_list(50), + ) + .await + ); + loop { + let frame = next_frame(&mut client).await; + if let Ok(blit_remote::events::EventsMessage::Streams { nonce: 50, streams }) = + blit_remote::events::parse_events_message(&frame) + { + assert!(streams.is_empty()); + break; + } + } + + assert!( + write_frame( + &mut client, + &blit_remote::events::msg_events_config_set( + 51, + blit_remote::events::EVENTS_CONFIG_REVISION_ANY, + events::MIN_RING_SIZE as u64, + blit_remote::events::ActivationSet::low_throughput(), + ), + ) + .await + ); + let revision = loop { + let frame = next_frame(&mut client).await; + if let Ok(blit_remote::events::EventsMessage::Config { + nonce: 51, + revision, + size, + activations, + .. + }) = blit_remote::events::parse_events_message(&frame) + { + assert_eq!(size, events::MIN_RING_SIZE as u64); + assert_eq!( + activations, + blit_remote::events::ActivationSet::low_throughput() + ); + break revision; + } + }; + + assert!( + write_frame( + &mut client, + &blit_remote::events::msg_events_config_set( + 52, + revision - 1, + events::MIN_RING_SIZE as u64, + blit_remote::events::ActivationSet::all(), + ), + ) + .await + ); + loop { + let frame = next_frame(&mut client).await; + if let Ok(blit_remote::events::EventsMessage::Result { + nonce: 52, status, .. + }) = blit_remote::events::parse_events_message(&frame) + { + assert_eq!(status, STATUS_CONFLICT); + break; + } + } + + assert!(write_frame(&mut client, &blit_remote::events::msg_events_dump(53),).await); + loop { + let frame = next_frame(&mut client).await; + if let Ok(blit_remote::events::EventsMessage::Dump { nonce: 53, bytes }) = + blit_remote::events::parse_events_message(&frame) + { + assert_eq!( + &bytes[..blit_remote::events::EVENT_DUMP_MAGIC.len()], + blit_remote::events::EVENT_DUMP_MAGIC + ); + break; + } + } + + let mut stream_ids = Vec::new(); + for nonce in 60..65 { + assert!( + write_frame( + &mut client, + &blit_remote::events::msg_events_stream_start( + nonce, + blit_remote::events::EVENTS_TARGET_CLIENT, + 0, + "", + ), + ) + .await + ); + loop { + let frame = next_frame(&mut client).await; + if let Ok(blit_remote::events::EventsMessage::StreamStarted { + nonce: reply_nonce, + status, + stream_id, + .. + }) = blit_remote::events::parse_events_message(&frame) + && reply_nonce == nonce + { + assert_eq!(status, STATUS_OK); + stream_ids.push(stream_id); + break; + } + } + } + assert_eq!(stream_ids.len(), 5); + + drop(client); + timeout(Duration::from_secs(5), connection) + .await + .expect("connection cleanup timed out") + .unwrap(); + process_server.shutdown().await; + } + + #[tokio::test] + async fn extension_endpoint_advertises_and_reads_events_family() { + let process_server = process::Server::new(false, false); + let state = test_state(process_server.clone()); + let init = blit_remote::extension::ExtensionInit { + extension_id: 7, + definition_revision: 2, + attempt: 3, + task_id: 4, + flags: blit_remote::extension::EXT_FLAG_ENABLED, + hash: [5; 32], + name: "event-inspector", + args: Vec::new(), + }; + let options = ConnectionOptions::extension(&init, ConnectionCancellation::default()) + .expect("valid extension init"); + let (mut client, server_stream) = tokio::io::duplex(4 * 1024 * 1024); + let connection = + tokio::spawn(handle_client_with_options(server_stream, state, options)); + + let mut advertised = false; + loop { + let frame = next_frame(&mut client).await; + match frame.first().copied() { + Some(S2C_HELLO) => { + let features = u32::from_le_bytes(frame[3..7].try_into().unwrap()); + advertised = features & blit_remote::events::FEATURE_EVENTS != 0; + } + Some(S2C_READY) => break, + _ => {} + } + } + assert!(advertised); + + assert!( + write_frame(&mut client, &blit_remote::events::msg_events_config_get(71),).await + ); + loop { + let frame = next_frame(&mut client).await; + if let Ok(blit_remote::events::EventsMessage::Config { nonce: 71, .. }) = + blit_remote::events::parse_events_message(&frame) + { + break; + } + } + + drop(client); + timeout(Duration::from_secs(5), connection) + .await + .expect("connection cleanup timed out") + .unwrap(); + process_server.shutdown().await; + } + #[tokio::test] async fn copy_range_for_a_missing_pty_returns_a_correlated_failure() { let process_server = process::Server::new(false, false); @@ -22181,6 +23282,14 @@ mod tests { async fn create2_subscribes_the_creator_by_default() { let process_server = process::Server::new(false, false); let state = test_state(process_server.clone()); + state + .events + .configure( + events::MIN_RING_SIZE, + blit_remote::events::ActivationSet::low_throughput(), + None, + ) + .unwrap(); let inspect = state.clone(); let (mut client, server_stream) = tokio::io::duplex(1024 * 1024); let connection = tokio::spawn(handle_client_with_options( @@ -22209,6 +23318,41 @@ mod tests { break pty_id; } }; + let dump = inspect.events.dump(); + let mut at = blit_remote::events::EVENT_DUMP_HEADER_LEN; + let mut stages = Vec::new(); + let mut correlation_id = None; + while dump.len().saturating_sub(at) >= blit_remote::events::EVENT_RECORD_HEADER_LEN { + let len = u32::from_le_bytes(dump[at..at + 4].try_into().unwrap()) as usize; + if len < blit_remote::events::EVENT_RECORD_HEADER_LEN + || dump.len().saturating_sub(at) < len + { + break; + } + let kind = u16::from_le_bytes(dump[at + 4..at + 6].try_into().unwrap()); + let payload = &dump[at + blit_remote::events::EVENT_RECORD_HEADER_LEN..at + len]; + if kind == EventType::PtyCreate.id() + && payload.len() >= 14 + && u16::from_le_bytes(payload[8..10].try_into().unwrap()) == 91 + { + let connection_id = u64::from_le_bytes(payload[..8].try_into().unwrap()); + assert_eq!(*correlation_id.get_or_insert(connection_id), connection_id); + stages.push(payload[10]); + } + at += len; + } + assert_ne!(correlation_id, Some(0)); + assert_eq!( + stages, + [ + blit_remote::events::PTY_CREATE_REQUEST_RECEIVED, + blit_remote::events::PTY_CREATE_SESSION_ACQUIRED, + blit_remote::events::PTY_CREATE_SPAWN_BEGIN, + blit_remote::events::PTY_CREATE_SPAWN_END, + blit_remote::events::PTY_CREATE_REGISTERED, + blit_remote::events::PTY_CREATE_REPLY_WRITTEN, + ] + ); let session = inspect.session.lock().await; let creator = session.clients.values().next().expect("creating client"); assert!(creator.subscriptions.contains(&pty_id)); @@ -22241,6 +23385,138 @@ mod tests { process_server.shutdown().await; } + #[tokio::test(flavor = "multi_thread")] + async fn legacy_create_opcodes_use_the_fixed_pty_create_event_payload() { + let process_server = process::Server::new(false, false); + let state = test_state(process_server.clone()); + state + .events + .configure( + events::MIN_RING_SIZE, + blit_remote::events::ActivationSet::low_throughput(), + None, + ) + .unwrap(); + let inspect = state.clone(); + let (mut client, server_stream) = tokio::io::duplex(1024 * 1024); + let connection = tokio::spawn(handle_client_with_options( + server_stream, + state, + ConnectionOptions::network(), + )); + + while next_frame(&mut client).await.first().copied() != Some(S2C_READY) {} + + assert!( + write_frame( + &mut client, + &blit_remote::msg_create_tagged_command(24, 80, "legacy", "sleep 30"), + ) + .await + ); + let legacy_id = loop { + let frame = next_frame(&mut client).await; + if let Some(ServerMsg::Created { + pty_id, + tag: "legacy", + }) = blit_remote::parse_server_msg(&frame) + { + break pty_id; + } + }; + + assert!( + write_frame( + &mut client, + &blit_remote::msg_create_n_command(57, 24, 80, "numbered", "exit 0"), + ) + .await + ); + let numbered_id = loop { + let frame = next_frame(&mut client).await; + if let Some(ServerMsg::CreatedN { + nonce: 57, + pty_id, + tag: "numbered", + }) = blit_remote::parse_server_msg(&frame) + { + break pty_id; + } + }; + + assert!( + write_frame( + &mut client, + &blit_remote::msg_create_at(24, 80, "at", legacy_id), + ) + .await + ); + let at_id = loop { + let frame = next_frame(&mut client).await; + if let Some(ServerMsg::Created { pty_id, tag: "at" }) = + blit_remote::parse_server_msg(&frame) + { + break pty_id; + } + }; + + let dump = inspect.events.dump(); + let mut offset = blit_remote::events::EVENT_DUMP_HEADER_LEN; + let mut registered = Vec::new(); + while dump.len().saturating_sub(offset) >= blit_remote::events::EVENT_RECORD_HEADER_LEN + { + let len = u32::from_le_bytes(dump[offset..offset + 4].try_into().unwrap()) as usize; + if len < blit_remote::events::EVENT_RECORD_HEADER_LEN + || dump.len().saturating_sub(offset) < len + { + break; + } + let kind = u16::from_le_bytes(dump[offset + 4..offset + 6].try_into().unwrap()); + let payload = + &dump[offset + blit_remote::events::EVENT_RECORD_HEADER_LEN..offset + len]; + if kind == EventType::PtyCreate.id() { + assert_eq!(payload.len(), 14, "pty.create must have one wire layout"); + let connection_id = u64::from_le_bytes(payload[..8].try_into().unwrap()); + let nonce = u16::from_le_bytes(payload[8..10].try_into().unwrap()); + let stage = payload[10]; + let status = payload[11]; + let pty_id = u16::from_le_bytes(payload[12..14].try_into().unwrap()); + assert_ne!(connection_id, 0); + if stage == blit_remote::events::PTY_CREATE_REGISTERED { + assert_eq!(status, STATUS_OK); + registered.push((connection_id, nonce, pty_id)); + } + } + offset += len; + } + + assert_eq!(registered.len(), 3); + let connection_id = registered[0].0; + assert!(registered.contains(&(connection_id, 0, legacy_id))); + assert!(registered.contains(&(connection_id, 57, numbered_id))); + assert!(registered.contains(&(connection_id, 0, at_id))); + + for pty_id in [legacy_id, numbered_id, at_id] { + assert!(write_frame(&mut client, &blit_remote::msg_close(pty_id)).await); + } + let mut closed = Vec::new(); + while closed.len() < 3 { + let frame = next_frame(&mut client).await; + if let Some(ServerMsg::Closed { pty_id }) = blit_remote::parse_server_msg(&frame) + && !closed.contains(&pty_id) + { + closed.push(pty_id); + } + } + + drop(client); + timeout(Duration::from_secs(5), connection) + .await + .expect("connection cleanup timed out") + .unwrap(); + process_server.shutdown().await; + } + #[tokio::test(flavor = "multi_thread")] async fn create2_can_skip_updates_but_still_receives_exit_lifecycle() { let process_server = process::Server::new(false, false); diff --git a/docs/design/events.md b/docs/design/events.md new file mode 100644 index 00000000..49af5ab7 --- /dev/null +++ b/docs/design/events.md @@ -0,0 +1,425 @@ +# Binary server events (`blit.events.v1`) + +- **Status:** Implemented (`FEATURE_EVENTS`, feature bit 31) +- **Date:** 2026-08-21 +- **Companion to:** [../protocol.md](../protocol.md), + [../server.md](../server.md) + +## Summary + +The server owns one process-wide binary event journal. It is intended for +post-mortem diagnosis and short, targeted captures of hot paths that are too +expensive or too noisy for ordinary stderr logging. + +The journal defaults to a 1 MiB contiguous byte ring and the low-throughput +lifecycle set. Event call sites use `blit_event!`; the macro checks one atomic +activation bit before it evaluates the payload expression. Disabled hot events +therefore do not allocate, format, lock the ring, or copy inspected bytes. + +The same retained history can be dumped on demand, followed live through a +client connection, or written by a persistent server-side task. The history +and live handoff is ordered: for a history-enabled stream, a concurrent record +is in exactly one of the initial dump or the live stream, never silently +between them. + +## Goals + +- Keep a bounded, always-available diagnostic history in every server process. +- Make the disabled cost of high-volume events one atomic bit test, with no + payload construction or shared lock acquisition. +- Activate individual event types, whole categories, the safe default set, or + the complete catalog without restarting the server. +- Preserve raw protocol and PTY bytes when explicitly enabled, while keeping + those expensive and sensitive events off by default. +- Support post-mortem dumps and live client/file capture through one stable, + self-describing binary representation. +- Keep event production independent of slow clients and files. Observability + must not become backpressure on the terminal, compositor, or network paths. +- Expose loss explicitly through counters, sequence gaps, and stream gap + records instead of presenting an incomplete capture as complete. + +## Non-goals + +- **Not a replacement for stderr, metrics, or tracing spans.** The journal is + a bounded forensic record, not the primary operator log or an aggregation + system. +- **Not lossless under arbitrary load.** The ring overwrites old records and a + slow live consumer can lag. Both cases are detectable. +- **Not durable unless a file stream is configured.** Ring contents disappear + with the process and configuration changes are not persisted by the + protocol. +- **Not a new authorization boundary.** Direct clients and extensions already + have server-side filesystem/process authority and may inspect events. A + transport that deliberately exposes a read-only subset still withholds the + family. +- **Not a universal payload schema.** The common record header is stable; + event-specific payloads remain compact binary structures owned by their + event types. + +## Architecture + +```mermaid +flowchart LR + Site["blit_event! call site"] -->|"atomic bit enabled"| Encode["Build binary payload"] + Encode --> Record["EventLog::record"] + Record --> Ring["Bounded byte ring"] + Record --> Broadcast["Bounded live broadcast"] + Ring --> Dump["Dedicated dump task"] + Ring --> Handoff["Atomic history/live handoff"] + Broadcast --> Handoff + Handoff --> Client["Connection-scoped client task"] + Handoff --> File["Process-scoped file task"] + Protocol["blit.events.v1 config"] --> Bits["4 atomic u64 words"] + Protocol --> Resize["Ring resize task"] + Bits --> Site + Resize --> Ring +``` + +`EventLog` is process-wide and stored in `AppState`. It owns four independent +activation words, the ring, a monotonic sequence allocator, one bounded Tokio +broadcast channel, and the registry of persistent file tasks. Detached file +recordings and connection-local client stream tasks have no arbitrary admission +cap. Client tasks remain owned by the connection handler, which aborts all of +them on disconnect so none can retain a dead outbox. + +### Storage model + +The ring is one fixed-size byte allocation rather than a queue of heap-owned +event objects. Every record begins with its complete length. Before an append, +whole oldest records are evicted until the new record fits; neither wrapping +nor shrinking can leave a partial retained record. Resizing builds a new ring +from oldest to newest, allowing its ordinary eviction rule to preserve the +newest records that fit. + +Each accepted event receives a sequence before retention is attempted. An +oversized event therefore advances `next_sequence` and `dropped`, even though +it cannot enter the ring. If live receivers exist, that event is still sent to +them: retention capacity does not impose a smaller live-record limit. + +Timestamps contain both process-monotonic nanoseconds and an approximate Unix +nanosecond value derived from the process-start wall clock. Sequence and +monotonic time define ordering; wall time exists for correlation with external +logs and may inherit ordinary wall-clock error. + +### Activation and hot-path cost + +An event id is also its bit index. `blit_event!` reads exactly one atomic word +and evaluates its payload expression only when the bit is set. This ordering is +intentional: full frame/PTY capture often allocates and copies more bytes than +the event header itself. `EventLog::record` then serializes the common header, +takes the ring mutex once, appends, and performs a non-awaiting broadcast send. + +Configuration stores and event reads are allowed to race. A concurrent event +may observe either activation set, which is preferable to placing a global +configuration lock on every call site. Ring resizing is serialized by the same +mutex used for append and dump. Configuration replacement has a process-wide +revision and may be conditional on that revision; the comparison and mutation +share this same mutex. + +The common 32-byte record header is encoded in a stack array and header plus +payload are copied directly into the wrapping ring. No heap-owned broadcast +record is built when the broadcast channel has no receivers. With receivers, +one `Arc<[u8]>` is allocated after retention and shared by all followers. + +### Snapshot and live-stream ordering + +Creating a stream locks the ring, subscribes to the broadcast channel, and +builds its history header before releasing the lock. Producers append and +broadcast while holding that lock; the broadcast operation itself never +awaits. Consequently, a concurrent record is either appended before the +snapshot and absent from the new receiver, or appended afterward and delivered +live. This is the central no-gap/no-duplicate handoff invariant. + +The channel is bounded by record count. A receiver that falls behind gets an +explicit lost-record count and resumes at the oldest still-available live +record. Client streams encode that as `STREAM_GAP`; file streams insert the +synthetic record type `65535`. Stream I/O happens only in dedicated tasks and +never while holding the ring mutex. + +Client tasks drain already-queued records into one `RECORD` message up to a +256 KiB soft byte limit (an individual larger record is sent alone). This +amortizes protocol envelopes and outbox entries without adding latency before +the first available record. + +Standalone dump construction and configuration resize run on blocking tasks +because they can copy the entire configured capacity. Starting a stream builds +its bounded initial snapshot under the ring lock before handing subsequent I/O +to the stream task. File tasks receive a one-shot stop, drain already-queued +live records (including a final gap marker if necessary), and flush before +shutdown returns. + +## Core invariants + +1. Retained bytes contain only complete, length-valid records. +2. `used` never exceeds `capacity`; capacity is bounded from 4 KiB to below the + maximum logical protocol message. +3. Event sequence numbers are process-wide, monotonic, and never reused. +4. A disabled macro call does not construct its payload or lock the ring. +5. A stream's history/live boundary neither drops nor duplicates a concurrent + event. +6. A slow stream cannot block an event producer; any resulting loss is + observable. +7. Live `EVENTS_RECORD` delivery is not itself logged, preventing recursive + record generation. Other event-control replies remain inspectable. +8. Stable event ids are never renumbered or reused with a different meaning. + +## Configuration + +`blit events config` reads the current configuration revision, capacity, +retained byte/record counts, overwrite count, next sequence, and complete +256-bit activation set. + +```bash +blit events config +blit events set --size 8388608 --events 'default,+frame.*,+pty.*' +blit events set --events 'all,-frame.write' +blit events set --if-revision 12 --events default --size 1048576 +``` + +Selectors are evaluated left-to-right. They are `default`, `all`, `none`, an +exact catalog name, or `category.*`; `+` and `-` enable and disable. A spec +whose first selector has a sign starts from the low-throughput default. + +Every successful replacement advances `revision`. `--if-revision X` applies a +replacement only if the revision is still `X`; otherwise it returns common +status `CONFLICT` and leaves both size and activations unchanged. A temporary +capture can therefore save configuration at revision `X`, enable/resize it, +then restore the saved values only if the capture's own returned revision is +still current. A successful set returns the exact revision created by that +replacement even if another update races with response delivery. The restore +therefore cannot erase a concurrent operator's change. + +All settings are also available at startup: + +| Variable | Default | Meaning | +| -------------------------- | --------- | ------------------------------------------------- | +| `BLIT_EVENTS_SIZE` | `1048576` | Ring bytes; 4 KiB through just under 64 MiB | +| `BLIT_EVENTS` | `default` | Activation selector expression | +| `BLIT_EVENTS_FILE` | unset | Start a persistent server-side binary file stream | +| `BLIT_EVENTS_FILE_HISTORY` | `1` | `0` starts the startup file at the next event | +| `BLIT_EVENTS_FILE_APPEND` | `0` | `1` appends instead of truncating | + +Invalid startup sizes fall back to 1 MiB. Invalid activation expressions are +reported and fall back to `default`. + +## Event catalog + +The stable event id is its activation-bit index. IDs 0–15 are the default +low-throughput set. + +| IDs | Names | Payload intent | +| ----- | ---------------------------------------------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------- | +| 0–15 | `server.*`, `task.*`, `client.*`, `config.change`, `stream.*`, `protocol.error`, `pty.create`, `pty.exit`, `pty.remove`, `pty.deadline`, `server.capacity` | Lifecycle and failures | +| 16–19 | `frame.read`, `frame.write`, `message.read`, `message.write` | Exact framed payloads and opcode/length summaries | +| 20–23 | `tick.*`, `session.lock` | Scheduler passes, wakeups, and lock wait time | +| 24–29 | `pty.read`, `pty.write`, `pty.parse`, `pty.snapshot`, `pty.resize`, `pty.input` | PTY byte flow and terminal processing | +| 30–34 | `compositor.*`, `surface.encode`, `surface.frame`, `audio.frame` | Compositor and media pipeline activity | +| 35–43 | `fs.request`, `git.request`, `lsp.request`, `kv.request`, `net.request`, `process.request`, `extension.request`, `channel.request`, `client.control` | Protocol-family dispatch | +| 44–47 | `outbox.queue`, `supervisor.event`, `connection.accept`, `server.error` | Delivery and server internals | + +`pty.create` uses one fixed payload for every create opcode: +`[connection_id:8][nonce:2][stage:1][status:1][pty_id:2]`. Stages are valid +request received (1), session mutex acquired (2), spawn begin (3), spawn end +(4), PTY registered (5), refused (6), and correlated reply physically written +(7). `connection_id` and `nonce` are present in every stage; `status` and +`pty_id` make success/refusal paths distinguishable. Legacy `CREATE` and +`CREATE_AT` have no nonce and emit zero; `CREATE_N` preserves its nonce. Legacy +opcodes currently emit only the PTY-registered stage. + +The full names and ids live in `crates/remote/src/events.rs`. Unknown activation +bits round-trip through the protocol, allowing a new server catalog to be +configured by a generic client. + +## Security and access + +`frame.read`, `frame.write`, `pty.read`, and `pty.write` contain the inspected +bytes, not a text rendering. They can include terminal contents, clipboard +data, paths, environment values, and other secrets. The event family has the +same authority model as the rest of a direct server connection; read-only +share forwarders do not allow opcode `0xD0`. In-process extension endpoints +already have filesystem and process authority, so they advertise and accept the +event family under the same stream machinery instead of pretending the recorder +is an additional sandbox boundary. + +## Dump format + +A dump is self-describing and uses little-endian fields: + +```text +[magic:"BLITEVT1":8] +[header_len:2 = 84][version:2 = 1] +[capacity:8][used:8][record_count:8][dropped:8][next_sequence:8] +[activations:4 * u64] +[records...] +``` + +Each retained record is complete even after wrapping: + +```text +[record_len:4][event_type:2][flags:2] +[sequence:8][monotonic_ns:8][unix_ns:8][type_payload:N] +``` + +`record_len` includes the 32-byte record header. `sequence` increments for +every attempted record, including a record too large for the configured ring. +`dropped` counts oversize records and records overwritten or discarded during +a shrink. `unix_ns` is the process-start wall clock plus the monotonic offset; +ordering should use `sequence` or `monotonic_ns`. + +A live file stream begins with a dump header. With history it contains the +retained records too; without history the header has zero used bytes and zero +records, followed by new records. Append mode starts another self-describing +header/history segment rather than appending unframed records to an old +segment. A lagged file task inserts synthetic type `65535` with `[lost:8]`. +Client streams report the same condition in an `EVENTS_STREAM_GAP` envelope. + +## Protocol and tasks + +Feature bit 31 advertises the direction-local `0xD0` envelope. Every request +starts `[0xD0][version:1 = 1][operation:1][nonce:2]`: + +| Operation | Body | +| ------------------ | ----------------------------------------------- | +| `CONFIG_GET` (1) | empty | +| `CONFIG_SET` (2) | `[expected_revision:8][size:8][activations:32]` | +| `DUMP` (3) | empty | +| `STREAM_START` (4) | `[target:1][flags:1][path_len:2][path:N]` | +| `STREAM_STOP` (5) | `[stream_id:4]` | +| `STREAM_LIST` (6) | empty | + +Targets are client `0` and server file `1`. Flags are history bit 0 and append +bit 1; append is invalid for a client target. Config changes are process-wide. +`expected_revision = u64::MAX` means unconditional replacement; any other value +is compared atomically and returns `CONFLICT` when stale. Stream ids are +process-wide and unique. Server-file streams persist after the requesting +client disconnects; client streams stop with their connection and can be +stopped only on that connection. + +Server messages start `[0xD0][version:1 = 1][kind:1]`: + +| Kind | Body | +| -------------------- | ---------------------------------------------------------------------------------------------- | +| `CONFIG` (1) | `[nonce:2][revision:8][size:8][used:8][records:8][dropped:8][next_sequence:8][activations:32]` | +| `RESULT` (2) | `[nonce:2][status:1][stream_id:4][detail:utf8]` | +| `DUMPED` (3) | `[nonce:2][dump:N]` | +| `STREAM_STARTED` (4) | `[nonce:2][status:1][stream_id:4][detail:utf8]` | +| `RECORD` (5) | `[stream_id:4][count:2][complete records...]` | +| `STREAM_STOPPED` (6) | `[stream_id:4][status:1][detail:utf8]` | +| `STREAM_GAP` (7) | `[stream_id:4][lost:8]` | +| `STREAMS` (8) | `[nonce:2][count:2][entries...]` | + +Each `STREAMS` entry is +`[stream_id:4][state:1][flags:1][records:8][bytes:8][lost:8]` +`[path_len:2][error_len:2][path:N][error:M]`. States are running `1`, stopped +`2`, and failed `3`. Counters cover successfully written records/bytes and live +records lost before the synthetic gap marker. It enumerates process-scoped +server recordings, including a task that failed after start but has not yet +been stopped and removed. + +Standalone `DUMP` requests run outside the connection reader; every live +client/file stream has its own dedicated Tokio task and broadcast receiver, +without a fixed per-connection or process-wide admission count. A slow stream +can lag and report loss, but never blocks the producer or the ring. + +## Failure behavior + +| Condition | Behavior visible to the operator | +| -------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | +| Ring lacks space | Complete oldest records are evicted and `dropped` increases. | +| One record exceeds ring capacity | It is not retained, `dropped` increases, and active live streams still receive it. | +| Live receiver falls behind | Client gets `STREAM_GAP`; file gets synthetic type `65535`, both with the lost count. | +| Resize shrinks below retained use | Newest complete records survive and every eviction increments `dropped`. | +| Conditional config revision is stale | The replacement returns `CONFLICT`; size and activations remain unchanged. | +| Invalid protocol request | A correlated common-status error is returned when a nonce can be recovered. | +| Invalid startup size | Capacity falls back to 1 MiB. | +| Invalid startup activation expression | The error is written to stderr and activation falls back to `default`. | +| Server-file open, header/history write, or initial flush fails | Protocol start returns an error and no recording id; startup configuration reports stderr and records `server.error` when enabled. | +| File write or final flush fails after start | `record list` reports `failed`, successful record/byte counters, live loss, and the error; `record stop` removes the task but returns the error. | +| Client disconnects | Its stream tasks are aborted and removed; process-scoped file streams continue. | +| Server shuts down | `server.stop` is recorded, file tasks drain queued records, flush, and join. | + +Loss has two independent measures. The dump header's `dropped` count describes +retention loss in the ring. A stream gap describes delivery loss for one live +receiver. A capture can have either without the other, so consumers must not +merge the counters. + +## Compatibility and evolution + +- Feature bit 31 lets a client avoid sending this family to an older server. + The v1 envelope is strictly decoded by a supporting server, without changing + any existing opcode layout. +- `0xD0` is direction-local and was unused in both tables when assigned. +- Event ids are append-only. Decoders must preserve unknown ids as binary + records, and activation bits outside the known catalog round-trip unchanged. +- `header_len` lets a future dump version add common metadata without moving + the record section for a decoder that honors the length. +- A record's `flags` and type payload are available for compatible per-event + growth. Changing the meaning of an existing id or common field requires a + new dump/protocol version. +- Append-mode files contain consecutive self-describing dump segments. + Readers should accept a new `BLITEVT1` header at a record boundary. + +## Alternatives considered + +**Text or JSON events.** Rejected for the hot path: formatting and escaping are +paid before retention, raw binary frames inflate substantially, and parsing +cost obscures the timings being diagnosed. + +**A `VecDeque>` of records.** Simpler wrapping semantics, but it adds an +allocation per enabled event and makes the configured byte bound include +allocator behavior rather than one exact reservation. + +**A fixed-record or lock-free ring.** Not adopted without measurements. The +variable-length byte ring preserves exact raw frames and compact richer +payloads under one byte budget. V1 first removes avoidable header and idle-live +allocations; a benchmark must demonstrate that another layout improves the +real hot path before trading those semantics away. + +**Synchronous writes to every stream.** Rejected because a slow filesystem or +client would add latency and failure modes to PTY, compositor, and protocol +processing. + +**An unbounded live channel.** Rejected because diagnostics must not turn a +slow consumer into unbounded server memory. The bounded broadcast plus explicit +gaps makes the tradeoff visible. + +**A memory-mapped persistent ring.** Rejected for v1: it adds crash-consistency, +permissions, cleanup, and cross-platform concerns to the always-on path. An +ordinary file stream supplies opt-in durability without changing ring +semantics. + +**One global activation lock.** Rejected because disabled events are the common +case. Four atomic `u64` words cover 256 stable ids on every supported target +without requiring wide atomic support. + +## Validation + +Tests cover ring wrapping, shrink preservation, disabled-event gating, +oversized live delivery, conditional configuration conflicts, client/file +streams beyond the former admission counts, file initialization and +delayed-write status, activation +selectors, strict batched-record codecs, extension-endpoint access, and +correlated `CREATE2` stages through physical reply write. Framed connections +negotiate the feature, change configuration, and retrieve dumps. The full +remote, server, and CLI unit suites and strict Clippy are part of the +implementation verification. + +## Operator workflow + +```bash +blit events dump > /tmp/blit.events +blit events dump --output /tmp/blit.events +blit events tail > /tmp/live.events +blit events tail --output /tmp/live.events --from-now +ID=$(blit events record start /var/log/blit.events) +blit events record list +blit events record stop "$ID" +``` + +`dump` and `tail` always deliver bytes to the invoking client and default to +stdout; `--output` is therefore always a local path. `record` exclusively +manages detached server-side file tasks, making both path locality and client +lifetime explicit in the grammar. `record start` does not print an id until the +file header and requested history are written and flushed; `record list` +exposes later failures and counters, and `record stop` reports a delayed write +or final-flush failure. diff --git a/docs/protocol.md b/docs/protocol.md index 5ebe5fd8..97182736 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -112,6 +112,7 @@ every pending operation as a connection error in that case. | `0x51` | `TERM_OUTPUT` | `[nonce:2][pty_id:2][index:8][max_bytes:4][flags:1]` — one command's output; `index = u64::MAX` is the newest ([design/term-journal.md](design/term-journal.md)) | | `0x52` | `TERM_SINCE` | `[nonce:2][pty_id:2][from_seq:8][from_col:2][max_bytes:4][flags:1]` — output since a sequence cursor; `flags` bit 0 (`SINCE_PROBE`) reports the cursor and returns no text ([design/term-journal.md](design/term-journal.md)) | | `0x53` | `TERM_JOURNAL_WAIT` | `[nonce:2][pty_id:2][index:8][timeout_ms:4]` — block until that command finishes; `index = u64::MAX` waits on whatever is running, or the next one to start ([design/term-journal.md](design/term-journal.md)) | +| `0xD0` | `EVENTS` | `[version:1][operation:1][nonce:2][body:N]` — `blit.events.v1` config, dump, and stream control ([design/events.md](design/events.md)) | **Notes:** @@ -371,6 +372,7 @@ shared sizing input without a `SURFACE_RESIZE` entry. | `0x50` | `TERM_JOURNAL` | `[nonce:2][pty_id:2][status:1][oldest_index:8][next_index:8][count:2][records…]` — answer to `C2S_TERM_JOURNAL` ([design/term-journal.md](design/term-journal.md)) | | `0x51` | `TERM_OUTPUT` | `[nonce:2][pty_id:2][status:1][flags:1][start_seq:8][start_col:2][next_seq:8][next_col:2][text_len:4][text:N]` — answers `TERM_OUTPUT` and `TERM_SINCE`; `next_*` is the cursor to resume from ([design/term-journal.md](design/term-journal.md)) | | `0x52` | `TERM_COMMAND` | `[nonce:2][pty_id:2][status:1][record]` — answer to `C2S_TERM_JOURNAL_WAIT`; on timeout the record is still flagged `RUNNING` ([design/term-journal.md](design/term-journal.md)) | +| `0xD0` | `EVENTS` | `[version:1][operation:1][body:N]` — correlated config/dump/start results or stream record/stop/gap ([design/events.md](design/events.md)) | **Notes:** @@ -409,6 +411,7 @@ shared sizing input without a `SURFACE_RESIZE` entry. | 28 | `TERM_JOURNAL` | Per-command journal and sequence-addressed output | | 29 | `CREATE_EXEC` | `CREATE2(HAS_ARGV)` and `CREATE2(HAS_ENV)`; Unix hosts only | | 30 | `CREATE_NO_SUBSCRIBE` | `CREATE2(NO_SUBSCRIBE)` skips creator frame subscription | +| 31 | `EVENTS` | Server supports the `blit.events.v1` binary event journal | Bit 26 is advertised with bit 12 and never alone; it is separate because a `WATCH` an older server does not know is dropped by the channel family's diff --git a/docs/server.md b/docs/server.md index fdc3903b..fe3a08ae 100644 --- a/docs/server.md +++ b/docs/server.md @@ -19,6 +19,11 @@ no RPC API beyond the binary protocol described in | `BLIT_TERM_JOURNAL_MAX` | `256` | Finished command records retained per PTY | | `BLIT_TERM_JOURNAL_CMD_MAX` | `4096` | Bytes of command-line text retained per record | | `BLIT_TERM_OUTPUT_MAX` | `1048576` (1 MiB) | Server ceiling on a `TERM_OUTPUT` / `TERM_SINCE` `max_bytes` | +| `BLIT_EVENTS_SIZE` | `1048576` (1 MiB) | Process-wide binary event-ring capacity | +| `BLIT_EVENTS` | `default` | Fine-grained event activation selectors | +| `BLIT_EVENTS_FILE` | unset | Start a persistent server-side binary event stream | +| `BLIT_EVENTS_FILE_HISTORY` | `1` | `0` excludes retained history from the startup file | +| `BLIT_EVENTS_FILE_APPEND` | `0` | `1` appends the startup file instead of truncating | | `BLIT_VAAPI_DEVICE` | `/dev/dri/renderD128` | VA-API render node for surface encoding and camera decoding | | `BLIT_CUDA_DEVICE` | `0` | CUDA device ordinal for NVENC and NVDEC | | `BLIT_FD_CHANNEL` | unset | fd-channel file descriptor | diff --git a/e2e/tests/mobile-keyboard.spec.ts b/e2e/tests/mobile-keyboard.spec.ts index ff2eaf2e..c2db0e45 100644 --- a/e2e/tests/mobile-keyboard.spec.ts +++ b/e2e/tests/mobile-keyboard.spec.ts @@ -38,7 +38,7 @@ async function authenticate(page: Page) { // Terminal creation goes over the mux; wait for it, not just the UI shell. await expect(page.getByRole("status")).toHaveAttribute( "aria-label", - "connected", + "Connected", { timeout: 15_000 }, ); await page.waitForTimeout(500);