diff --git a/compositor/src/handlers/screenshot.rs b/compositor/src/handlers/screenshot.rs index 84f0e4d..e1d023b 100644 --- a/compositor/src/handlers/screenshot.rs +++ b/compositor/src/handlers/screenshot.rs @@ -345,6 +345,7 @@ enum RequestState { } struct Request { + owner: u64, response_id: u64, reply: ReplySender, state: RequestState, @@ -374,6 +375,8 @@ pub(crate) const REQUEST_TIMEOUT: Duration = Duration::from_secs(30); pub(crate) struct Coordinator { requests: std::collections::HashMap, next_id: u64, + deadline_changed: Option)>>, + notified_deadline: Option, } impl Coordinator { @@ -381,11 +384,29 @@ impl Coordinator { Self::default() } + pub(crate) fn set_deadline_observer(&mut self, observer: impl Fn(Option) + 'static) { + self.notified_deadline = self.next_deadline(); + observer(self.notified_deadline); + self.deadline_changed = Some(Box::new(observer)); + } + pub(crate) fn admit( &mut self, + owner: u64, response_id: u64, reply: ReplySender, specs: Vec, + ) -> Result { + self.admit_at(owner, response_id, reply, specs, Instant::now()) + } + + fn admit_at( + &mut self, + owner: u64, + response_id: u64, + reply: ReplySender, + specs: Vec, + now: Instant, ) -> Result { if specs.is_empty() { return Err("No outputs are enabled".into()); @@ -416,14 +437,16 @@ impl Coordinator { self.requests.insert( id, Request { + owner, response_id, reply, state: RequestState::Collecting, parts: specs.into_iter().map(|spec| (spec, PartState::Pending)).collect(), received: 0, - deadline: Instant::now() + REQUEST_TIMEOUT, + deadline: now + REQUEST_TIMEOUT, }, ); + self.notify_deadline_change(); Ok(id) } @@ -455,36 +478,22 @@ impl Coordinator { } let frame = match result { - Ok(ref buffer) => { - let mut pixels = Vec::new(); - if pixels.try_reserve_exact(buffer.pixels.len()).is_err() { - None - } else { - pixels.extend_from_slice(&buffer.pixels); - Some(OutputFrame { - preserve_alpha: spec.preserve_alpha, - width: buffer.width, - height: buffer.height, - stride: buffer.stride, - pixels, - transform: spec.transform, - location: spec.location, - scale: spec.scale, - logical_width: spec.logical_width, - logical_height: spec.logical_height, - }) - } + Ok(buffer) => OutputFrame { + preserve_alpha: spec.preserve_alpha, + width: buffer.width, + height: buffer.height, + stride: buffer.stride, + pixels: buffer.pixels, + transform: spec.transform, + location: spec.location, + scale: spec.scale, + logical_width: spec.logical_width, + logical_height: spec.logical_height, + }, + Err(error) => { + self.reject_inner(request, &error); + return Action::None; } - Err(_) => None, - }; - - let Some(frame) = frame else { - let error = match result { - Err(error) => error, - Ok(_) => "Screenshot capture could not be allocated".to_string(), - }; - self.reject_inner(request, &error); - return Action::None; }; let Some(entry) = self.requests.get_mut(&request) else { @@ -516,7 +525,7 @@ impl Coordinator { Err(_) => Action::None, }; } - let entry = self.requests.remove(&request).expect("checked above"); + let entry = self.remove_request(request).expect("checked above"); match result { Ok(path) => { reply(&entry, Ok(&path)); @@ -536,8 +545,7 @@ impl Coordinator { // Answers every request that outlived its deadline. The returned ids let the // caller drop readbacks that would otherwise publish into a request that no // longer exists. - pub(crate) fn expire(&mut self) -> Vec { - let now = Instant::now(); + pub(crate) fn expire_at(&mut self, now: Instant) -> Vec { let expired: Vec = self .requests .iter() @@ -550,13 +558,22 @@ impl Coordinator { expired } - // The remaining time before the earliest deadline, so the caller can sleep - // exactly that long instead of polling on a fixed tick. - pub(crate) fn next_deadline(&self) -> Option { - self.requests - .values() - .map(|entry| entry.deadline.saturating_duration_since(Instant::now())) - .min() + // Absolute deadlines do not drift when unrelated work wakes the loop. + pub(crate) fn next_deadline(&self) -> Option { + self.requests.values().map(|entry| entry.deadline).min() + } + + pub(crate) fn terminate_owner(&mut self, owner: u64) -> Vec { + let ids: Vec = self + .requests + .iter() + .filter(|(_, entry)| entry.owner == owner) + .map(|(id, _)| *id) + .collect(); + for id in &ids { + self.reject_inner(*id, "Screenshot cancelled: the client disconnected"); + } + ids } pub(crate) fn terminate_all(&mut self) -> Vec { @@ -572,10 +589,28 @@ impl Coordinator { } fn reject_inner(&mut self, request: u64, error: &str) { - if let Some(entry) = self.requests.remove(&request) { + if let Some(entry) = self.remove_request(request) { reply(&entry, Err(error)); } } + + // All terminal paths pass through here, including readback/encode failure, + // owner closure, cancellation and timeout, so none can leave a stale timer. + fn remove_request(&mut self, request: u64) -> Option { + let entry = self.requests.remove(&request)?; + self.notify_deadline_change(); + Some(entry) + } + + fn notify_deadline_change(&mut self) { + let deadline = self.next_deadline(); + if deadline != self.notified_deadline { + self.notified_deadline = deadline; + if let Some(observer) = &self.deadline_changed { + observer(deadline); + } + } + } } fn reply(entry: &Request, result: Result<&Path, &str>) { @@ -918,7 +953,7 @@ mod completion { let (reply, received) = sync_channel(8); let mut coordinator = Coordinator::new(); coordinator - .admit(42, reply.clone(), (0..parts).map(|_| spec(2, 2)).collect()) + .admit(1, 42, reply.clone(), (0..parts).map(|_| spec(2, 2)).collect()) .expect("admitted"); Self { coordinator, received } } @@ -928,6 +963,23 @@ mod completion { } } + #[test] + fn completed_readback_moves_its_pixels_to_the_encoder() { + let mut harness = Harness::new(1); + let id = harness.coordinator.next_id; + let mut captured = buffer(2, 2); + captured.pixels.fill(137); + let allocation = captured.pixels.as_ptr(); + + let Action::Encode { frames, .. } = harness.coordinator.on_part(id, 0, Ok(captured)) else { + panic!("completed readback was not queued for encoding"); + }; + + assert_eq!(frames.len(), 1); + assert_eq!(frames[0].pixels.as_ptr(), allocation, "readback pixels were copied"); + assert_eq!(frames[0].pixels, vec![137; 16]); + } + #[test] fn part_failure_terminates_the_request_with_an_error() { let mut harness = Harness::new(2); @@ -1105,7 +1157,7 @@ mod completion { let id = harness.coordinator.next_id; harness.coordinator.requests.get_mut(&id).expect("tracked").deadline = Instant::now() - Duration::from_secs(1); - assert_eq!(harness.coordinator.expire(), vec![id]); + assert_eq!(harness.coordinator.expire_at(Instant::now()), vec![id]); let response = harness.response().expect("the caller is answered"); assert_eq!(response.error.as_ref().unwrap().code, "screenshot_failed"); assert!( @@ -1126,10 +1178,14 @@ mod completion { fn a_live_request_is_not_expired_early() { let mut harness = Harness::new(1); let id = harness.coordinator.next_id; - assert!(harness.coordinator.expire().is_empty(), "not yet due"); + assert!(harness.coordinator.expire_at(Instant::now()).is_empty(), "not yet due"); assert!(harness.coordinator.is_live(id)); assert!(harness.response().is_none(), "still unanswered, not failed"); - let remaining = harness.coordinator.next_deadline().expect("still armed"); + let remaining = harness + .coordinator + .next_deadline() + .expect("still armed") + .saturating_duration_since(Instant::now()); assert!( remaining <= REQUEST_TIMEOUT && remaining > REQUEST_TIMEOUT - Duration::from_secs(1), "the timer re-arms to roughly the full timeout, got {remaining:?}" @@ -1142,17 +1198,17 @@ mod completion { let mut coordinator = Coordinator::new(); for _ in 0..MAX_OUTSTANDING { coordinator - .admit(42, reply.clone(), vec![spec(2, 2)]) + .admit(1, 42, reply.clone(), vec![spec(2, 2)]) .expect("within the request cap"); } for entry in coordinator.requests.values_mut() { entry.deadline = Instant::now() - Duration::from_secs(1); } - assert_eq!(coordinator.expire().len(), MAX_OUTSTANDING); + assert_eq!(coordinator.expire_at(Instant::now()).len(), MAX_OUTSTANDING); assert_eq!(coordinator.next_deadline(), None); // The cap is usable again rather than permanently degraded. coordinator - .admit(42, reply, vec![spec(2, 2)]) + .admit(1, 42, reply, vec![spec(2, 2)]) .expect("a slot was released"); } @@ -1162,10 +1218,10 @@ mod completion { let mut coordinator = Coordinator::new(); for _ in 0..MAX_OUTSTANDING { coordinator - .admit(42, reply.clone(), vec![spec(2, 2)]) + .admit(1, 42, reply.clone(), vec![spec(2, 2)]) .expect("within the request cap"); } - let overflow = coordinator.admit(42, reply.clone(), vec![spec(2, 2)]); + let overflow = coordinator.admit(1, 42, reply.clone(), vec![spec(2, 2)]); assert!(overflow.is_err(), "outstanding requests are bounded"); let huge = PartSpec { @@ -1176,11 +1232,172 @@ mod completion { }; let mut fresh = Coordinator::new(); assert!( - fresh.admit(42, reply.clone(), vec![huge]).is_err(), + fresh.admit(1, 42, reply.clone(), vec![huge]).is_err(), "the byte budget is checked before any readback allocates" ); } + fn observe_deadlines(coordinator: &mut Coordinator) -> std::rc::Rc>>> { + let changes = std::rc::Rc::new(std::cell::RefCell::new(Vec::new())); + let observed = changes.clone(); + coordinator.set_deadline_observer(move |deadline| observed.borrow_mut().push(deadline)); + changes + } + + #[test] + fn deadline_admission_after_long_idle_and_earliest_removal_rearm_exactly() { + let now = Instant::now(); + let (reply, _received) = sync_channel(8); + let mut coordinator = Coordinator::new(); + let changes = observe_deadlines(&mut coordinator); + assert_eq!(&*changes.borrow(), &[None]); + + // An idle period does not establish a polling epoch. Every request gets + // exactly the timeout measured from its own admission. + let late = now + Duration::from_secs(3600); + let first = coordinator + .admit_at(1, 42, reply.clone(), vec![spec(2, 2)], late) + .unwrap(); + let second = coordinator + .admit_at(2, 43, reply.clone(), vec![spec(2, 2)], late + Duration::from_secs(5)) + .unwrap(); + assert_eq!( + changes.borrow().len(), + 2, + "a later request does not rearm the earliest timer" + ); + assert!(coordinator.admit_at(1, 44, reply, Vec::new(), late).is_err()); + assert_eq!(changes.borrow().len(), 2, "rejected admission does not touch the timer"); + assert_eq!(coordinator.terminate_owner(1), vec![first]); + assert!(coordinator.is_live(second)); + assert_eq!(coordinator.terminate_owner(99), Vec::::new()); + coordinator.reject(second, "encoding queue is full"); + assert_eq!( + &*changes.borrow(), + &[ + None, + Some(late + REQUEST_TIMEOUT), + Some(late + Duration::from_secs(5) + REQUEST_TIMEOUT), + None, + ] + ); + } + + #[test] + fn deadline_survives_readback_and_encoding_then_disarms_on_completion() { + for result in [Ok(PathBuf::from("/tmp/shot.png")), Err("encoder failed".into())] { + let mut harness = Harness::new(2); + let changes = observe_deadlines(&mut harness.coordinator); + let id = harness.coordinator.next_id; + let deadline = harness.coordinator.next_deadline(); + harness.coordinator.on_part(id, 0, Ok(buffer(2, 2))); + assert!(matches!( + harness.coordinator.on_part(id, 1, Ok(buffer(2, 2))), + Action::Encode { .. } + )); + assert_eq!(&*changes.borrow(), &[deadline], "encoding keeps the original deadline"); + harness.coordinator.on_encoded(id, result); + assert_eq!(&*changes.borrow(), &[deadline, None]); + harness.coordinator.on_part(id, 0, Err("late readback".into())); + harness.coordinator.on_encoded(id, Err("late encode".into())); + assert_eq!( + changes.borrow().len(), + 2, + "late results cannot rearm an empty coordinator" + ); + } + } + + #[test] + fn deadline_cancellations_disarm_collecting_and_encoding_requests() { + for encoding in [false, true] { + for cancellation in [ + "output was removed", + "session locked", + "privacy changed", + "client closed", + ] { + // Topology failures arrive as parts while collecting. Once + // encoding begins its snapshotted frames remain valid. + if encoding && cancellation == "output was removed" { + continue; + } + let mut harness = Harness::new(1); + let changes = observe_deadlines(&mut harness.coordinator); + let id = harness.coordinator.next_id; + if encoding { + harness.coordinator.on_part(id, 0, Ok(buffer(2, 2))); + } + match cancellation { + "output was removed" => { + harness.coordinator.on_part(id, 0, Err(cancellation.into())); + } + "session locked" => { + harness.coordinator.terminate_all(); + } + "privacy changed" => { + harness.coordinator.terminate_all_with_reason(cancellation); + } + "client closed" => { + harness.coordinator.terminate_owner(1); + } + _ => unreachable!(), + } + assert_eq!(changes.borrow().last(), Some(&None), "{cancellation} must disarm"); + assert_eq!(changes.borrow().len(), 2); + assert!(harness.response().unwrap().error.is_some()); + assert!(!harness.coordinator.is_live(id)); + } + } + } + + #[test] + fn deadline_expiry_moves_to_next_request_and_frees_encoding_slots() { + let now = Instant::now(); + let (reply, received) = sync_channel(8); + let mut coordinator = Coordinator::new(); + let changes = observe_deadlines(&mut coordinator); + let first = coordinator + .admit_at(1, 42, reply.clone(), vec![spec(2, 2)], now) + .unwrap(); + let second = coordinator + .admit_at(2, 43, reply.clone(), vec![spec(2, 2)], now + Duration::from_secs(5)) + .unwrap(); + coordinator.on_part(first, 0, Ok(buffer(2, 2))); + assert!( + coordinator + .expire_at(now + REQUEST_TIMEOUT - Duration::from_nanos(1)) + .is_empty() + ); + assert_eq!(coordinator.expire_at(now + REQUEST_TIMEOUT), vec![first]); + assert!(coordinator.is_live(second)); + assert_eq!( + coordinator.expire_at(now + REQUEST_TIMEOUT + Duration::from_secs(5)), + vec![second] + ); + assert_eq!( + &*changes.borrow(), + &[ + None, + Some(now + REQUEST_TIMEOUT), + Some(now + REQUEST_TIMEOUT + Duration::from_secs(5)), + None, + ] + ); + assert_eq!(received.try_iter().count(), 2, "both timeouts answer once"); + assert!(matches!( + coordinator.on_encoded(first, Ok(PathBuf::from("/tmp/late.png"))), + Action::DiscardFile(_) + )); + coordinator + .admit_at(3, 44, reply, vec![spec(2, 2)], now + Duration::from_secs(3600)) + .unwrap(); + assert_eq!( + changes.borrow().last(), + Some(&Some(now + Duration::from_secs(3600) + REQUEST_TIMEOUT)) + ); + } + #[test] fn a_sink_publishes_once_and_then_rejects_duplicates() { let (sender, receiver) = loop_channel(); diff --git a/compositor/src/handlers/screenshot_worker.rs b/compositor/src/handlers/screenshot_worker.rs index 7a76072..09cddba 100644 --- a/compositor/src/handlers/screenshot_worker.rs +++ b/compositor/src/handlers/screenshot_worker.rs @@ -115,7 +115,7 @@ fn encode_png(canvas: &Canvas) -> Result, String> { if width == 0 || height == 0 { return Err("Screenshot is empty".into()); } - let image = image::RgbaImage::from_raw(width, height, canvas.pixels.clone()) + let image = image::ImageBuffer::, _>::from_raw(width, height, canvas.pixels.as_slice()) .ok_or("Screenshot buffer does not match its dimensions")?; let mut out = Vec::new(); let encoder = image::codecs::png::PngEncoder::new_with_quality( @@ -210,7 +210,11 @@ mod tests { request: 1, frames: vec![frame(4, 3)], }; - let canvas = compose(&job.frames).unwrap(); + let mut canvas = compose(&job.frames).unwrap(); + for (index, pixel) in canvas.pixels.chunks_exact_mut(4).enumerate() { + pixel[..3].copy_from_slice(&[index as u8, 137, 241]); + } + let png = encode_png(&canvas).expect("png"); assert_eq!(&png[..8], b"\x89PNG\r\n\x1a\n", "png signature"); @@ -218,6 +222,7 @@ mod tests { .expect("decodes back") .to_rgba8(); assert_eq!(decoded.dimensions(), (4, 3)); + assert_eq!(decoded.as_raw(), &canvas.pixels, "encoding changed the pixel data"); // The opaque alpha we force is preserved through the round trip. assert_eq!(decoded.get_pixel(0, 0)[3], 255); } diff --git a/compositor/src/ipc.rs b/compositor/src/ipc.rs index 0dfb8f5..27ed1bd 100644 --- a/compositor/src/ipc.rs +++ b/compositor/src/ipc.rs @@ -1,11 +1,13 @@ +use std::cell::RefCell; use std::os::fd::AsRawFd; use std::os::unix::fs::{FileTypeExt, MetadataExt, PermissionsExt}; use std::os::unix::net::{UnixListener, UnixStream}; use std::path::{Path, PathBuf}; +use std::rc::Rc; use std::sync::Arc; use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; use std::sync::mpsc::{SyncSender, sync_channel}; -use std::time::Duration; +use std::time::{Duration, Instant}; use std::{env, fs, io, thread}; use ferese_core::LayoutMode; @@ -13,7 +15,7 @@ use ferese_ipc::{Request, Response, VERSION, read_frame, write_frame}; use ferese_layout::Direction; use serde_json::{Value, json}; use smithay::output::Output; -use smithay::reexports::calloop::{EventLoop, LoopSignal, channel, timer}; +use smithay::reexports::calloop::{EventLoop, LoopHandle, LoopSignal, RegistrationToken, channel, timer}; use smithay::utils::Transform; use crate::Ferese; @@ -34,10 +36,6 @@ const _: () = assert!( "the result queue must outsize the job queue plus the in-flight job" ); const SWEEP_INTERVAL: Duration = crate::handlers::screenshot_worker::SWEEP_INTERVAL; -// Only consulted when no request is outstanding, so it bounds how late a newly -// admitted request can be noticed. With a request pending the timer re-arms to -// the exact remaining time instead. -const DEADLINE_IDLE: Duration = Duration::from_secs(1); #[derive(Debug)] struct IpcCall { @@ -110,6 +108,11 @@ pub(crate) fn init(event_loop: &mut EventLoop<'static, Ferese>) -> Result call, IpcEvent::Closed(owner) => { + for request in state.screenshot.terminate_owner(owner) { + state + .pending_screencopies + .retain(|capture| capture.request_id() != Some(request)); + } let captured = state.input_capture.active(); state.input_capture.remove_owner(owner); if captured && !state.input_capture.active() { @@ -191,11 +194,11 @@ pub(crate) fn init(event_loop: &mut EventLoop<'static, Ferese>) -> Result) -> Result) -> Result) -> Result, pub(crate) _guard: IpcSocketGuard, } +// The timer callback clears its registration before expiring requests, allowing +// the coordinator observer to arm the next deadline during the same dispatch. +// Keeping absolute Instants also avoids pushing a timeout later on each rearm. +pub(crate) struct DeadlineTimer { + handle: LoopHandle<'static, Data>, + armed: Rc>>, + expire: fn(&mut Data, Instant), +} + +impl DeadlineTimer { + fn new(handle: LoopHandle<'static, Data>, expire: fn(&mut Data, Instant)) -> Self { + Self { + handle, + armed: Rc::new(RefCell::new(None)), + expire, + } + } + + pub(crate) fn update(&self, deadline: Option) { + if self.armed.borrow().as_ref().map(|(deadline, _)| *deadline) == deadline { + return; + } + if let Some((_, token)) = self.armed.borrow_mut().take() { + self.handle.remove(token); + } + let Some(deadline) = deadline else { + return; + }; + let armed = self.armed.clone(); + let expire = self.expire; + // calloop's Timer registers only in its in-memory wheel and always + // returns Ok; a failure here would break that infrastructure invariant. + let token = self + .handle + .insert_source(timer::Timer::from_deadline(deadline), move |_, _, state| { + armed.borrow_mut().take(); + expire(state, Instant::now()); + timer::TimeoutAction::Drop + }) + .expect("could not register screenshot deadline timer"); + *self.armed.borrow_mut() = Some((deadline, token)); + } +} + +impl Drop for DeadlineTimer { + fn drop(&mut self) { + if let Some((_, token)) = self.armed.borrow_mut().take() { + self.handle.remove(token); + } + } +} + fn accept_connections(listener: UnixListener, sender: channel::SyncSender, signal: LoopSignal) { let active_connections = Arc::new(AtomicUsize::new(0)); let connection_ids = AtomicU64::new(1); @@ -777,7 +826,7 @@ impl Ferese { // Screenshot replies are deferred: this either answers the caller now, or // hands the reply to the coordinator, which answers exactly once. - pub(crate) fn start_screenshot(&mut self, request: Request, response: SyncSender) { + pub(crate) fn start_screenshot(&mut self, owner: u64, request: Request, response: SyncSender) { // A macro rather than a closure: the message may be a borrowed str or an // owned String, and Response::error already accepts either. macro_rules! reject { @@ -849,7 +898,7 @@ impl Ferese { // admit stores the reply only on success, so a clone survives the // rejection path and every caller is answered exactly once. let fallback = response.clone(); - let id = match self.screenshot.admit(request.id, response, specs) { + let id = match self.screenshot.admit(owner, request.id, response, specs) { Ok(id) => id, Err(message) => { let _ = fallback.try_send(Response::error(request.id, "screenshot_rejected", message)); @@ -878,7 +927,7 @@ impl Ferese { crate::backends::direct::render_on(self, &outputs); } - fn start_window_screenshot(&mut self, request: Request, response: SyncSender) { + fn start_window_screenshot(&mut self, owner: u64, request: Request, response: SyncSender) { let reject = |message: String| { let _ = response.try_send(Response::error(request.id, "window_capture_failed", message)); }; @@ -936,7 +985,7 @@ impl Ferese { buffer_height: size.h, }; let fallback = response.clone(); - let capture = match self.screenshot.admit(request.id, response, vec![spec]) { + let capture = match self.screenshot.admit(owner, request.id, response, vec![spec]) { Ok(capture) => capture, Err(error) => { let _ = fallback.try_send(Response::error(request.id, "screenshot_rejected", error)); @@ -1392,6 +1441,115 @@ mod tests { use super::*; + #[derive(Default)] + struct DeadlineState { + fired: usize, + next: Option, + timer: Option>, + } + + fn deadline_loop() -> (EventLoop<'static, DeadlineState>, DeadlineState) { + let event_loop = EventLoop::try_new().unwrap(); + let timer = DeadlineTimer::new(event_loop.handle(), |state: &mut DeadlineState, _| { + state.fired += 1; + state.timer.as_ref().unwrap().update(state.next.take()); + }); + ( + event_loop, + DeadlineState { + timer: Some(timer), + ..Default::default() + }, + ) + } + + #[test] + fn screenshot_deadline_idle_has_no_timer_and_admission_arms_immediately() { + let (mut event_loop, mut state) = deadline_loop(); + let timer = state.timer.as_ref().unwrap(); + timer.update(None); + assert!(timer.armed.borrow().is_none(), "idle registers no timeout source"); + event_loop.dispatch(Duration::ZERO, &mut state).unwrap(); + assert_eq!(state.fired, 0); + + let deadline = Instant::now() + Duration::from_secs(3600); + let timer = state.timer.as_ref().unwrap(); + timer.update(Some(deadline)); + assert_eq!( + timer.armed.borrow().as_ref().map(|(deadline, _)| *deadline), + Some(deadline) + ); + timer.update(None); + assert!(timer.armed.borrow().is_none()); + event_loop.dispatch(Duration::ZERO, &mut state).unwrap(); + assert_eq!(state.fired, 0); + } + + #[test] + fn screenshot_deadline_replacement_and_cancellation_remove_old_sources() { + let (mut event_loop, mut state) = deadline_loop(); + let due = Instant::now() - Duration::from_secs(1); + let future = Instant::now() + Duration::from_secs(3600); + let timer = state.timer.as_ref().unwrap(); + timer.update(Some(due)); + let original = timer.armed.borrow().as_ref().unwrap().1; + timer.update(Some(due)); + assert_eq!( + timer.armed.borrow().as_ref().unwrap().1, + original, + "unchanged deadline keeps its source" + ); + timer.update(Some(future)); + event_loop.dispatch(Duration::ZERO, &mut state).unwrap(); + assert_eq!(state.fired, 0, "the replaced due timer must never fire"); + state.timer.as_ref().unwrap().update(Some(due)); + state.timer.as_ref().unwrap().update(None); + event_loop.dispatch(Duration::ZERO, &mut state).unwrap(); + assert_eq!(state.fired, 0, "cancelling the due timer removes its wakeup"); + assert!(state.timer.as_ref().unwrap().armed.borrow().is_none()); + } + + #[test] + fn screenshot_deadline_drop_removes_a_registered_due_source() { + let (mut event_loop, mut state) = deadline_loop(); + state + .timer + .as_ref() + .unwrap() + .update(Some(Instant::now() - Duration::from_secs(1))); + drop(state.timer.take()); + event_loop.dispatch(Duration::ZERO, &mut state).unwrap(); + assert_eq!(state.fired, 0, "dropping the observer removes its timer callback"); + } + + #[test] + fn screenshot_deadline_expiry_can_rearm_during_dispatch_then_returns_to_idle() { + let (mut event_loop, mut state) = deadline_loop(); + let due = Instant::now() - Duration::from_secs(1); + let future = Instant::now() + Duration::from_secs(3600); + state.next = Some(future); + state.timer.as_ref().unwrap().update(Some(due)); + event_loop.dispatch(Duration::ZERO, &mut state).unwrap(); + assert_eq!(state.fired, 1); + assert_eq!( + state + .timer + .as_ref() + .unwrap() + .armed + .borrow() + .as_ref() + .map(|(deadline, _)| *deadline), + Some(future) + ); + state.timer.as_ref().unwrap().update(Some(due)); + event_loop.dispatch(Duration::ZERO, &mut state).unwrap(); + assert_eq!(state.fired, 2); + assert!(state.timer.as_ref().unwrap().armed.borrow().is_none()); + event_loop.dispatch(Duration::ZERO, &mut state).unwrap(); + assert_eq!(state.fired, 2, "an empty queue has no repeat wakeup"); + } + fn request(version: u32, kind: &str) -> Request { Request { version, diff --git a/compositor/src/state.rs b/compositor/src/state.rs index e2407f1..931a34a 100644 --- a/compositor/src/state.rs +++ b/compositor/src/state.rs @@ -767,6 +767,9 @@ impl Ferese { }; state.config_worker = Some(crate::reload::Worker::new(event_loop)?); let screenshot = crate::ipc::init(event_loop)?; + state + .screenshot + .set_deadline_observer(move |deadline| screenshot.deadline_timer.update(deadline)); state.screenshot_parts = Some(screenshot.parts); state.screenshot_worker = Some(screenshot.worker); state._ipc_socket = Some(screenshot._guard); diff --git a/shell/ferese-shell/src/clock.rs b/shell/ferese-shell/src/clock.rs new file mode 100644 index 0000000..89c1129 --- /dev/null +++ b/shell/ferese-shell/src/clock.rs @@ -0,0 +1,749 @@ +//! Visible clock labels, driven by realtime boundaries and timezone changes. +use std::ffi::{CString, OsString}; +use std::io; +use std::os::fd::{AsRawFd, FromRawFd, OwnedFd}; +use std::os::unix::ffi::OsStrExt; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::sync::Arc; + +use cosmic::iced::{Subscription, futures}; +use ferese_config::desktop::Clock; +use jiff::{Timestamp, Zoned, tz::TimeZone}; + +const SECOND: i128 = 1_000_000_000; +const MINUTE: i128 = 60 * SECOND; +const UNNAMED_ZONE: &str = "ferese/unnamed-system-zone"; +// Fractional formats keep the previous sampling rate; nanosecond formats must +// not turn into a nanosecond timer that continuously rebuilds the shell. +const FRACTION_SAMPLE: i128 = SECOND / 2; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Labels { + pub bar: String, + pub desktop: (String, String), +} + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct Format { + time: String, + date: Option, + zone: Option, + lowercase: bool, +} + +#[derive(Clone, Debug, Hash, PartialEq, Eq)] +struct Settings { + bar: bool, + desktop: Option>, +} + +#[derive(Default)] +pub struct Service { + desktop: Option>, +} + +impl Service { + pub fn new(clock: &Clock) -> Self { + let mut service = Self::default(); + service.configure(clock); + service + } + + pub fn configure(&mut self, clock: &Clock) { + if !clock.enabled { + self.desktop = None; + return; + } + + let date = clock.show_date.then_some(clock.date_format.as_str()); + let zone = clock.time_zone.as_deref().filter(|zone| !zone.is_empty()); + if self.desktop.as_ref().is_some_and(|format| { + format.time == clock.time_format + && format.date.as_deref() == date + && format.zone.as_deref() == zone + && format.lowercase == clock.lowercase + }) { + return; + } + + self.desktop = Some(Arc::new(Format { + time: clock.time_format.clone(), + date: date.map(str::to_owned), + zone: zone.map(str::to_owned), + lowercase: clock.lowercase, + })); + } + + pub fn subscription(&self, bar: bool, desktop_visible: bool) -> Subscription { + self.settings(bar, desktop_visible) + .map_or_else(Subscription::none, |settings| Subscription::run_with(settings, stream)) + } + + fn settings(&self, bar: bool, desktop_visible: bool) -> Option { + let desktop = if desktop_visible { self.desktop.clone() } else { None }; + if !bar && desktop.is_none() { + return None; + } + + Some(Settings { bar, desktop }) + } +} + +fn stream(settings: &Settings) -> impl futures::Stream + use<> { + let settings = settings.clone(); + let (send, receive) = tokio::sync::mpsc::channel(1); + let stop = match UnixStream::pair() { + Ok((stop, cancel)) => { + match std::thread::Builder::new().name("ferese-clock".into()).spawn(move || { + if let Err(error) = watch(&settings, send, cancel) { + eprintln!("ferese-shell: clock subscription: {error}"); + } + }) { + Ok(_) => Some(stop), + Err(error) => { + eprintln!("ferese-shell: clock worker: {error}"); + None + } + } + } + Err(error) => { + eprintln!("ferese-shell: clock cancellation socket: {error}"); + None + } + }; + // Dropping the stream closes the socket and wakes the worker's poll. + futures::stream::unfold((receive, stop), |(mut receive, stop)| async move { + Some((receive.recv().await?, (receive, stop))) + }) +} + +fn watch(settings: &Settings, send: tokio::sync::mpsc::Sender, cancel: UnixStream) -> io::Result<()> { + let timer = + owned_fd(unsafe { libc::timerfd_create(libc::CLOCK_REALTIME, libc::TFD_CLOEXEC | libc::TFD_NONBLOCK) })?; + let (mut system, mut desktop, mut paths) = zones(settings); + let mut timezone = ZoneWatch::new(&paths)?; + let mut previous = None; + loop { + let now = Timestamp::now(); + let labels = labels(settings, now, &system, &desktop); + if previous.as_ref() != Some(&labels) { + previous = Some(labels.clone()); + if send.blocking_send(labels).is_err() { + return Ok(()); + } + } + if arm(&timer, deadline(settings, now, &system, &desktop))? { + continue; + } + let mut fds = [ + libc::pollfd { + fd: timer.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }, + libc::pollfd { + fd: timezone.fd.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }, + libc::pollfd { + fd: cancel.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }, + ]; + loop { + if unsafe { libc::poll(fds.as_mut_ptr(), fds.len() as _, -1) } >= 0 { + break; + } + let error = io::Error::last_os_error(); + if error.kind() != io::ErrorKind::Interrupted { + return Err(error); + } + } + if fds[2].revents != 0 { + return Ok(()); + } + if fds[0].revents != 0 { + read_timer(&timer)?; + } + if fds[1].revents != 0 && timezone.changed()? { + jiff::tz::db().reset(); + (system, desktop, paths) = zones(settings); + timezone = ZoneWatch::new(&paths)?; + } + } +} + +fn owned_fd(fd: libc::c_int) -> io::Result { + if fd < 0 { + Err(io::Error::last_os_error()) + } else { + Ok(unsafe { OwnedFd::from_raw_fd(fd) }) + } +} + +fn arm(timer: &OwnedFd, deadline: i128) -> io::Result { + let spec = libc::itimerspec { + it_interval: libc::timespec { tv_sec: 0, tv_nsec: 0 }, + it_value: libc::timespec { + tv_sec: deadline.div_euclid(SECOND) as _, + tv_nsec: deadline.rem_euclid(SECOND) as _, + }, + }; + let flags = libc::TFD_TIMER_ABSTIME | libc::TFD_TIMER_CANCEL_ON_SET; + if unsafe { libc::timerfd_settime(timer.as_raw_fd(), flags, &spec, std::ptr::null_mut()) } < 0 { + let error = io::Error::last_os_error(); + if error.raw_os_error() == Some(libc::ECANCELED) { + // settime can apply the new deadline yet report an unread clock + // cancellation. Drain it and resample before waiting again. + read_timer(timer)?; + return Ok(true); + } + return Err(error); + } + Ok(false) +} + +fn read_timer(timer: &OwnedFd) -> io::Result<()> { + let mut expirations = 0u64; + let result = unsafe { libc::read(timer.as_raw_fd(), (&mut expirations as *mut u64).cast(), 8) }; + if result < 0 { + let error = io::Error::last_os_error(); + // A wall-clock adjustment cancels the timer. Recompute labels/deadline + // just as after expiration; CLOCK_REALTIME also advances during suspend. + if !matches!(error.raw_os_error(), Some(libc::ECANCELED | libc::EAGAIN | libc::EINTR)) { + return Err(error); + } + } + Ok(()) +} + +fn zones(settings: &Settings) -> (TimeZone, TimeZone, Vec) { + let mut paths = vec![PathBuf::from("/etc/localtime"), PathBuf::from("/etc/timezone")]; + // Load the file directly: Jiff's system timezone cache lasts five minutes. + let system = match std::env::var("TZ") { + Ok(zone) if zone.is_empty() => TimeZone::UTC, + Ok(zone) => named_zone(zone.trim_start_matches(':'), &mut paths).unwrap_or_else(TimeZone::system), + Err(_) => file_zone(Path::new("/etc/localtime"), &mut paths).unwrap_or_else(TimeZone::system), + }; + let desktop = settings + .desktop + .as_ref() + .and_then(|format| format.zone.as_ref()) + .and_then(|zone| named_zone(zone, &mut paths)) + .unwrap_or_else(|| system.clone()); + (system, desktop, paths) +} + +fn named_zone(name: &str, paths: &mut Vec) -> Option { + // Jiff treats UTC case-insensitively and gives it the Etc/UTC identity. + if name.eq_ignore_ascii_case("UTC") { + return Some(TimeZone::UTC); + } + for root in ["/usr/share/zoneinfo", "/usr/share/lib/zoneinfo", "/etc/zoneinfo"] { + let path = Path::new(root).join(name); + if let Some(zone) = file_zone(&path, paths) { + return Some(zone); + } + } + file_zone(Path::new(name), paths) + .or_else(|| TimeZone::posix(name).ok()) + .or_else(|| TimeZone::get(name).ok()) +} + +fn zone_identifier(path: &Path) -> Option<&str> { + path.to_str() + .and_then(|path| path.split_once("zoneinfo/").map(|(_, name)| name)) +} + +fn file_zone(path: &Path, paths: &mut Vec) -> Option { + let data = std::fs::read(path).ok()?; + let resolved = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_owned()); + let name = zone_identifier(path) + .or_else(|| zone_identifier(&resolved)) + .unwrap_or(UNNAMED_ZONE); + let zone = TimeZone::tzif(name, &data).ok()?; + paths.push(path.to_owned()); + paths.push(resolved); + Some(zone) +} + +fn labels(settings: &Settings, now: Timestamp, system: &TimeZone, desktop: &TimeZone) -> Labels { + let bar = if settings.bar { + ferese_theme::calendar::format_bar_time(&now.to_zoned(system.clone())) + } else { + String::new() + }; + let desktop = settings + .desktop + .as_ref() + .map(|format| { + let now = now.to_zoned(desktop.clone()); + let mut broken = jiff::fmt::strtime::BrokenDownTime::from(&now); + if desktop.iana_name() == Some(UNNAMED_ZONE) { + // Copied localtime files have no IANA name. Preserve Jiff's + // %Q offset fallback while retaining their DST/abbreviation. + broken.set_iana_time_zone(None); + } + let time = broken.to_string(&format.time).unwrap_or_default(); + let date = format + .date + .as_ref() + .map(|date| broken.to_string(date).unwrap_or_default()) + .unwrap_or_default(); + if format.lowercase { + (time.to_lowercase(), date.to_lowercase()) + } else { + (time, date) + } + }) + .unwrap_or_default(); + Labels { bar, desktop } +} + +fn deadline(settings: &Settings, now: Timestamp, system: &TimeZone, desktop: &TimeZone) -> i128 { + let mut next = i128::MAX; + if settings.bar { + next = next.min(zone_deadline(&now.to_zoned(system.clone()), MINUTE)); + } + if let Some(format) = &settings.desktop { + let precision = + format_precision(&format.time).min(format.date.as_deref().map(format_precision).unwrap_or(MINUTE)); + next = next.min(zone_deadline(&now.to_zoned(desktop.clone()), precision)); + } + next +} + +fn zone_deadline(now: &Zoned, precision: i128) -> i128 { + let stamp = now.timestamp().as_nanosecond(); + let next = if precision == MINUTE { + stamp + MINUTE - i128::from(now.second()) * SECOND - i128::from(now.subsec_nanosecond()) + } else { + (stamp.div_euclid(precision) + 1) * precision + }; + now.time_zone() + .following(now.timestamp()) + .next() + .map_or(next, |transition| next.min(transition.timestamp().as_nanosecond())) +} + +fn format_precision(format: &str) -> i128 { + let mut precision = MINUTE; + let mut chars = format.chars(); + while let Some(character) = chars.next() { + if character != '%' { + continue; + } + let Some(mut directive) = chars.next() else { + break; + }; + if directive == '%' { + continue; + } + while matches!(directive, '_' | '-' | '0' | '^' | '#' | ':' | '.') || directive.is_ascii_digit() { + let Some(next) = chars.next() else { + return precision; + }; + directive = next; + } + precision = precision.min(match directive { + 'f' | 'N' => FRACTION_SAMPLE, + 'S' | 's' | 'T' | 'c' | 'r' | 'X' => SECOND, + _ => MINUTE, + }); + } + precision +} + +struct ZoneWatch { + fd: OwnedFd, + names: Vec<(libc::c_int, OsString)>, +} + +impl ZoneWatch { + fn new(paths: &[PathBuf]) -> io::Result { + let fd = owned_fd(unsafe { libc::inotify_init1(libc::IN_CLOEXEC | libc::IN_NONBLOCK) })?; + let mut names = Vec::new(); + for path in paths { + let (Some(parent), Some(name)) = (path.parent(), path.file_name()) else { + continue; + }; + let Ok(parent) = CString::new(parent.as_os_str().as_bytes()) else { + continue; + }; + let mask = libc::IN_CLOSE_WRITE + | libc::IN_MOVED_TO + | libc::IN_CREATE + | libc::IN_DELETE + | libc::IN_ATTRIB + | libc::IN_MOVE_SELF + | libc::IN_DELETE_SELF; + let wd = unsafe { libc::inotify_add_watch(fd.as_raw_fd(), parent.as_ptr(), mask) }; + if wd >= 0 { + names.push((wd, name.to_owned())); + } + } + Ok(Self { fd, names }) + } + + fn changed(&self) -> io::Result { + let mut buffer = [0u8; 8192]; + let mut changed = false; + loop { + let size = unsafe { libc::read(self.fd.as_raw_fd(), buffer.as_mut_ptr().cast(), buffer.len()) }; + if size < 0 { + let error = io::Error::last_os_error(); + if error.kind() == io::ErrorKind::WouldBlock { + return Ok(changed); + } + if error.kind() == io::ErrorKind::Interrupted { + continue; + } + return Err(error); + } + if size == 0 { + return Ok(changed); + } + let mut offset = 0; + while offset + std::mem::size_of::() <= size as usize { + let event = unsafe { (buffer.as_ptr().add(offset) as *const libc::inotify_event).read_unaligned() }; + offset += std::mem::size_of::(); + let end = offset + event.len as usize; + if end > size as usize { + break; + } + let name = &buffer[offset..end]; + let name = &name[..name.iter().position(|byte| *byte == 0).unwrap_or(name.len())]; + changed |= event.mask + & (libc::IN_Q_OVERFLOW | libc::IN_IGNORED | libc::IN_MOVE_SELF | libc::IN_DELETE_SELF) + != 0 + || self + .names + .iter() + .any(|(wd, watched)| *wd == event.wd && watched.as_bytes() == name); + offset = end; + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + + fn settings(time: Option<&str>, date: Option<&str>) -> Settings { + Settings { + bar: true, + desktop: time.map(|time| { + Arc::new(Format { + time: time.into(), + date: date.map(str::to_owned), + zone: None, + lowercase: false, + }) + }), + } + } + + #[test] + fn subscription_rebuilds_and_style_changes_keep_the_cached_format() { + let mut clock = Clock { + enabled: true, + time_zone: Some("Asia/Kathmandu".into()), + ..Clock::default() + }; + let mut service = Service::new(&clock); + let first = service.settings(true, true).unwrap(); + let format = first.desktop.as_ref().unwrap(); + for _ in 0..100 { + let current = service.settings(true, true).unwrap(); + assert!(Arc::ptr_eq(format, current.desktop.as_ref().unwrap())); + assert_eq!(first, current); + } + + clock.time_size += 8.; + clock.margin_x += 10; + service.configure(&clock); + assert!(Arc::ptr_eq(format, service.desktop.as_ref().unwrap())); + assert!(service.settings(true, false).unwrap().desktop.is_none()); + assert!(service.settings(false, false).is_none()); + assert!(Arc::ptr_eq( + format, + service.settings(false, true).unwrap().desktop.as_ref().unwrap() + )); + } + + #[test] + fn format_changes_replace_subscription_identity_without_mutating_the_worker_settings() { + let mut clock = Clock { + enabled: true, + ..Clock::default() + }; + let mut service = Service::new(&clock); + let initial = service.settings(true, true).unwrap(); + clock.time_format = "%H:%M:%S".into(); + service.configure(&clock); + let seconds = service.settings(true, true).unwrap(); + assert_ne!(initial, seconds); + assert_eq!(initial.desktop.as_ref().unwrap().time, "%-I:%M %p"); + + clock.show_date = false; + service.configure(&clock); + let no_date = service.settings(true, true).unwrap(); + assert_ne!(seconds, no_date); + assert!(no_date.desktop.as_ref().unwrap().date.is_none()); + clock.date_format = "%Y".into(); + service.configure(&clock); + assert!(Arc::ptr_eq( + no_date.desktop.as_ref().unwrap(), + service.desktop.as_ref().unwrap() + )); + + clock.time_zone = Some("UTC".into()); + service.configure(&clock); + assert_ne!(no_date, service.settings(true, true).unwrap()); + clock.lowercase = !clock.lowercase; + let previous = service.settings(true, true).unwrap(); + service.configure(&clock); + assert_ne!(previous, service.settings(true, true).unwrap()); + clock.enabled = false; + service.configure(&clock); + assert!(service.settings(false, true).is_none()); + assert!(service.settings(true, true).unwrap().desktop.is_none()); + clock.enabled = true; + service.configure(&clock); + assert!(service.settings(false, true).unwrap().desktop.is_some()); + } + + #[test] + fn production_subscription_uses_visible_outputs_formats_and_hidden_date() { + let mut clock = Clock { + enabled: true, + time_format: "%H:%M".into(), + date_format: "%S%.f".into(), + ..Clock::default() + }; + assert_eq!(Service::new(&clock).subscription(false, false).units(), 0); + assert_eq!(Service::new(&clock).subscription(true, false).units(), 1); + let hidden = Service::new(&clock).settings(true, false).unwrap(); + assert!(hidden.desktop.is_none()); + let now: Timestamp = "2026-10-04T12:34:56.123Z".parse().unwrap(); + assert_eq!( + deadline(&hidden, now, &TimeZone::UTC, &TimeZone::UTC), + "2026-10-04T12:35:00Z".parse::().unwrap().as_nanosecond() + ); + let visible = Service::new(&clock).settings(true, true).unwrap(); + assert_eq!( + deadline(&visible, now, &TimeZone::UTC, &TimeZone::UTC), + "2026-10-04T12:34:56.5Z".parse::().unwrap().as_nanosecond() + ); + clock.show_date = false; + let hidden_date = Service::new(&clock).settings(true, true).unwrap(); + assert_ne!(hidden_date, visible); + assert_eq!( + deadline(&hidden_date, now, &TimeZone::UTC, &TimeZone::UTC), + "2026-10-04T12:35:00Z".parse::().unwrap().as_nanosecond() + ); + clock.time_format = "%S".into(); + assert_ne!(Service::new(&clock).settings(true, true).unwrap(), hidden_date); + clock.enabled = false; + assert_eq!(Service::new(&clock).subscription(false, true).units(), 0); + } + + #[test] + fn named_zone_labels_match_configured_clock_including_aliases_and_case() { + let now: Timestamp = "2026-10-04T12:34:56.123Z".parse().unwrap(); + for zone in ["America/New_York", "Etc/UTC", "Asia/Kathmandu", "UTC", "utc"] { + let clock = Clock { + enabled: true, + time_format: "%H:%M:%S%.3f %Q %Z".into(), + date_format: "%a %F".into(), + time_zone: Some(zone.into()), + ..Clock::default() + }; + let settings = Service::new(&clock).settings(true, true).unwrap(); + let desktop = named_zone(zone, &mut Vec::new()).unwrap_or_else(|| panic!("zone {zone} unavailable")); + let actual = labels(&settings, now, &TimeZone::UTC, &desktop).desktop; + assert_eq!(actual, clock.labels(&now.to_zoned(TimeZone::UTC)).unwrap(), "{zone}"); + } + } + + #[test] + fn localtime_replacement_loads_new_zone_without_system_cache() { + let mut paths = Vec::new(); + named_zone("America/New_York", &mut paths).unwrap(); + let eastern = std::fs::read(paths.first().unwrap()).unwrap(); + paths.clear(); + named_zone("Asia/Kathmandu", &mut paths).unwrap(); + let nepal = std::fs::read(paths.first().unwrap()).unwrap(); + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("localtime"); + std::fs::write(&path, eastern).unwrap(); + let watcher = ZoneWatch::new(std::slice::from_ref(&path)).unwrap(); + let now: Timestamp = "2026-10-04T12:34:56Z".parse().unwrap(); + let first = file_zone(&path, &mut Vec::new()).unwrap(); + let replacement = dir.path().join("replacement"); + std::fs::write(&replacement, nepal).unwrap(); + std::fs::rename(replacement, &path).unwrap(); + assert!(watcher.changed().unwrap()); + let second = file_zone(&path, &mut Vec::new()).unwrap(); + let bar_settings = settings(None, None); + assert_eq!(labels(&bar_settings, now, &first, &first).bar, "4 oct, 8:34 am"); + assert_eq!(labels(&bar_settings, now, &second, &second).bar, "4 oct, 6:19 pm"); + let settings = settings(Some("%Q %:Q %Z %z"), None); + assert_eq!( + labels(&settings, now, &first, &first).desktop.0, + "-0400 -04:00 EDT -0400" + ); + assert_eq!( + labels(&settings, now, &second, &second).desktop.0, + "+0545 +05:45 +0545 +0545" + ); + } + + #[test] + fn visible_format_precision_handles_seconds_escapes_and_sampled_fractions() { + for format in ["%-I:%M %p", "%a, %b %-d", "%%S %%f", "%:z %:::z", "fixed text"] { + assert_eq!(format_precision(format), MINUTE, "{format}"); + } + for format in ["%S", "%_3S", "%s", "%T", "%c", "%r", "%X"] { + assert_eq!(format_precision(format), SECOND, "{format}"); + } + for format in ["%f", "%N", "%.f", "%.3f", "%9f", "%1f"] { + assert_eq!(format_precision(format), FRACTION_SAMPLE, "{format}"); + } + let now: Timestamp = "2026-10-04T12:34:56.123Z".parse().unwrap(); + let minute = "2026-10-04T12:35:00Z".parse::().unwrap().as_nanosecond(); + assert_eq!( + deadline(&settings(None, None), now, &TimeZone::UTC, &TimeZone::UTC), + minute + ); + assert_eq!( + deadline( + &settings(Some("%H:%M"), Some("%S")), + now, + &TimeZone::UTC, + &TimeZone::UTC + ), + "2026-10-04T12:34:57Z".parse::().unwrap().as_nanosecond() + ); + assert_eq!( + deadline( + &settings(Some("%H:%M:%S%.3f"), None), + now, + &TimeZone::UTC, + &TimeZone::UTC + ), + "2026-10-04T12:34:56.5Z".parse::().unwrap().as_nanosecond() + ); + } + + #[test] + fn boundaries_handle_midnight_dst_and_offsets_with_seconds() { + for (stamp, expected) in [ + ("2026-10-04T23:59:59.9Z", "2026-10-05T00:00:00Z"), + ("2026-10-04T12:34:00Z", "2026-10-04T12:35:00Z"), + ] { + let now: Timestamp = stamp.parse().unwrap(); + assert_eq!( + zone_deadline(&now.to_zoned(TimeZone::UTC), MINUTE), + expected.parse::().unwrap().as_nanosecond() + ); + } + let zone = TimeZone::get("America/New_York").unwrap(); + for (stamp, expected, label) in [ + ("2026-03-08T06:59:59Z", "2026-03-08T07:00:00Z", "8 mar, 3:00 am"), + ("2026-11-01T05:59:59Z", "2026-11-01T06:00:00Z", "1 nov, 1:00 am"), + ] { + let now: Timestamp = stamp.parse().unwrap(); + let expected: Timestamp = expected.parse().unwrap(); + assert_eq!( + zone_deadline(&now.to_zoned(zone.clone()), MINUTE), + expected.as_nanosecond() + ); + assert_eq!(labels(&settings(None, None), expected, &zone, &zone).bar, label); + } + let zone = TimeZone::fixed(jiff::tz::Offset::from_seconds(30).unwrap()); + let now: Timestamp = "2026-10-04T12:34:20Z".parse().unwrap(); + assert_eq!( + zone_deadline(&now.to_zoned(zone), MINUTE), + "2026-10-04T12:34:30Z".parse::().unwrap().as_nanosecond() + ); + } + + #[test] + fn realtime_jumps_recompute_boundaries_and_timezone_labels() { + let settings = settings(Some("%H:%M %Z"), None); + let initial: Timestamp = "2026-10-04T12:34:56Z".parse().unwrap(); + let earlier: Timestamp = "2026-10-04T11:20:01Z".parse().unwrap(); + let later: Timestamp = "2026-10-04T15:40:01Z".parse().unwrap(); + let first = labels(&settings, initial, &TimeZone::UTC, &TimeZone::UTC); + for (now, expected) in [(earlier, "2026-10-04T11:21:00Z"), (later, "2026-10-04T15:41:00Z")] { + assert_ne!(labels(&settings, now, &TimeZone::UTC, &TimeZone::UTC), first); + assert_eq!( + deadline(&settings, now, &TimeZone::UTC, &TimeZone::UTC), + expected.parse::().unwrap().as_nanosecond() + ); + } + let zone = TimeZone::get("Asia/Kathmandu").unwrap(); + assert_ne!(labels(&settings, initial, &zone, &zone), first); + assert_eq!(labels(&settings, initial, &zone, &zone).desktop.0, "18:19 +0545"); + assert_eq!( + labels(&settings, initial, &TimeZone::UTC, &TimeZone::UTC), + labels( + &settings, + initial + jiff::SignedDuration::from_secs(1), + &TimeZone::UTC, + &TimeZone::UTC + ) + ); + } + + #[test] + fn timezone_watch_filters_unrelated_files_and_survives_atomic_replacement() { + let dir = tempfile::tempdir().unwrap(); + let path = dir.path().join("localtime"); + std::fs::write(&path, b"old").unwrap(); + let watcher = ZoneWatch::new(std::slice::from_ref(&path)).unwrap(); + std::fs::write(dir.path().join("unrelated"), b"data").unwrap(); + assert!(!watcher.changed().unwrap()); + let replacement = dir.path().join("replacement"); + std::fs::write(&replacement, b"new").unwrap(); + std::fs::rename(replacement, &path).unwrap(); + assert!(watcher.changed().unwrap()); + assert!(!watcher.changed().unwrap()); + std::fs::write(&path, b"updated").unwrap(); + assert!(watcher.changed().unwrap()); + } + + #[test] + fn absolute_timer_and_subscription_drop_wake_without_recurring_poll() { + let timer = + owned_fd(unsafe { libc::timerfd_create(libc::CLOCK_REALTIME, libc::TFD_CLOEXEC | libc::TFD_NONBLOCK) }) + .unwrap(); + arm(&timer, Timestamp::now().as_nanosecond() - SECOND).unwrap(); + let mut poll = libc::pollfd { + fd: timer.as_raw_fd(), + events: libc::POLLIN, + revents: 0, + }; + assert_eq!(unsafe { libc::poll(&mut poll, 1, 1000) }, 1); + read_timer(&timer).unwrap(); + // Reading an already drained timer is harmless (EAGAIN). + read_timer(&timer).unwrap(); + + let (stop, cancel) = UnixStream::pair().unwrap(); + let (send, mut receive) = tokio::sync::mpsc::channel(1); + let (done_send, done_receive) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + done_send.send(watch(&settings(None, None), send, cancel)).unwrap(); + }); + assert!(receive.blocking_recv().is_some()); + drop(stop); + done_receive.recv_timeout(Duration::from_secs(2)).unwrap().unwrap(); + assert!(receive.blocking_recv().is_none()); + } +} diff --git a/shell/ferese-shell/src/main.rs b/shell/ferese-shell/src/main.rs index 1152e13..772dd22 100644 --- a/shell/ferese-shell/src/main.rs +++ b/shell/ferese-shell/src/main.rs @@ -6,6 +6,7 @@ use bar::*; use desktop_widgets::*; use surfaces::*; +mod clock; mod compositor_ipc; mod config; mod control; @@ -169,6 +170,7 @@ struct FereseShell { snapshot: ShellSnapshot, overview_active: bool, clock: String, + clock_service: clock::Service, desktop_clock: (String, String), outputs: Vec, notifications: notifications::Center, @@ -239,7 +241,9 @@ enum Message { WallpaperLoaded(Result), Event(Event, window::Id), NativeSurface(window::Id, Result<(Connection, wl_surface::WlSurface), String>), - Tick, + ClockChanged(clock::Labels), + RecorderEvent(recording::Event), + RecorderElapsed, ControlReady, ActivateWorkspace(u64), ToggleOverview, @@ -327,6 +331,7 @@ impl cosmic::Application for FereseShell { note_inflight: Vec::new(), note_saving: false, note_error: None, + clock_service: clock::Service::new(&config.desktop_widgets.clock), config, wallpaper: None, control: ShellControl::connect() @@ -378,7 +383,16 @@ impl cosmic::Application for FereseShell { )) => Some(Message::Event(event, id)), _ => None, }), - cosmic::iced::time::every(Duration::from_millis(500)).map(|_| Message::Tick), + self.clock_service + .subscription( + self.outputs.iter().any(|output| !output.hidden), + self.outputs.iter().any(|output| { + output.clock.is_some() && self.config.desktop_widgets.clock.on_output(output.name.as_deref()) + }), + ) + .map(Message::ClockChanged), + self.recorder.subscription().map(Message::RecorderEvent), + self.recorder.elapsed_subscription().map(|_| Message::RecorderElapsed), self.status_service.subscription().map(Message::StatusUpdated), if self.config.desktop_widgets.clock.enabled || self @@ -729,21 +743,16 @@ impl cosmic::Application for FereseShell { Task::none() } Message::Event(event, id) => self.handle_event(event, id), - Message::Tick => { - self.recorder.poll(); - self.clock = current_time(); - - if self.config.desktop_widgets.clock.enabled { - self.desktop_clock = self - .config - .desktop_widgets - .clock - .labels(&Zoned::now()) - .unwrap_or_default(); - } - + Message::ClockChanged(labels) => { + self.clock = labels.bar; + self.desktop_clock = labels.desktop; Task::none() } + Message::RecorderEvent(event) => { + self.recorder.handle(event); + Task::none() + } + Message::RecorderElapsed => Task::none(), Message::ControlReady => { let mut reload_task = Task::none(); if let Some(control) = &self.control { @@ -895,6 +904,7 @@ impl FereseShell { || old.bar_margin_horizontal != theme.bar_margin_horizontal || old.bar_window_gap != theme.bar_window_gap; self.notifications.configure(config.notifications.clone()); + self.clock_service.configure(&config.desktop_widgets.clock); self.config = config; self.desktop_clock = self .config diff --git a/shell/ferese-shell/src/recording.rs b/shell/ferese-shell/src/recording.rs index d64b827..b197752 100644 --- a/shell/ferese-shell/src/recording.rs +++ b/shell/ferese-shell/src/recording.rs @@ -2,10 +2,9 @@ //! Encoding and portal negotiation run in the separate ferese-record process. use std::io::{BufRead, BufReader, Read}; use std::path::PathBuf; -use std::process::{ChildStdin, Command, Stdio}; -use std::sync::mpsc; -use std::sync::mpsc::Receiver; -use std::time::Instant; +use std::process::{Child, ChildStdin, Command, Stdio}; +use std::sync::Arc; +use std::time::{Duration, Instant}; use serde::Deserialize; @@ -19,21 +18,40 @@ pub enum State { Saved(PathBuf), Error(String), } -#[derive(Deserialize)] -struct Update { +#[derive(Clone, Debug, Deserialize)] +pub struct Update { state: String, message: Option, path: Option, } -enum Event { +#[derive(Clone, Debug)] +pub enum Event { Update(Update), Exited, } +#[derive(Clone)] +struct Events(Arc>>); + +impl std::hash::Hash for Events { + fn hash(&self, state: &mut H) { + std::ptr::hash(Arc::as_ptr(&self.0), state); + } +} + +impl Events { + fn stream(&self) -> impl cosmic::iced::futures::Stream + use<> { + cosmic::iced::futures::stream::unfold(self.0.clone(), |receiver| async move { + let event = receiver.lock().await.recv().await?; + Some((event, receiver)) + }) + } +} + #[derive(Default)] pub struct Recorder { pub state: State, control: Option, - updates: Option>, + updates: Option, } impl Recorder { pub fn label(&self) -> &'static str { @@ -76,32 +94,11 @@ impl Recorder { { Ok(mut child) => { self.control = child.stdin.take(); - let output = child.stdout.take().unwrap(); - let (send, receive) = mpsc::sync_channel(8); - self.updates = Some(receive); + let (send, receive) = tokio::sync::mpsc::channel(8); + self.updates = Some(Events(Arc::new(tokio::sync::Mutex::new(receive)))); self.state = State::Selecting; - std::thread::spawn(move || { - let mut output = BufReader::new(output); - - loop { - let mut line = String::new(); - match output.by_ref().take(65537).read_line(&mut line) { - Ok(0) | Err(_) => break, - Ok(_) if line.len() > 65536 => break, - Ok(_) => { - if let Ok(update) = serde_json::from_str::(&line) - && send.send(Event::Update(update)).is_err() - { - break; - } - } - } - } - - let _ = child.wait(); - let _ = send.send(Event::Exited); - }); + std::thread::spawn(move || read_events(child, send)); } Err(error) => { let message = format!("Cannot start recorder: {error}"); @@ -119,47 +116,101 @@ impl Recorder { } } - pub fn poll(&mut self) { - let mut done = false; - if let Some(updates) = &self.updates { - while let Ok(event) = updates.try_recv() { - match event { - Event::Update(update) => match update.state.as_str() { - "selecting" if self.control.is_some() => self.state = State::Selecting, - "recording" if self.control.is_some() => self.state = State::Recording(Instant::now()), - "saving" => self.state = State::Saving, - "saved" => { - self.state = update - .path - .map(State::Saved) - .unwrap_or_else(|| State::Error("Recorder returned no saved file".into())) - } - "cancelled" => self.state = State::Idle, - "error" => { - self.state = State::Error(update.message.unwrap_or_else(|| "Recording failed".into())) - } - _ => (), - }, - Event::Exited => { - if matches!(self.state, State::Selecting | State::Recording(_) | State::Saving) { - self.state = State::Error("The recorder stopped unexpectedly".into()); - } - done = true; - } - } - } + pub fn subscription(&self) -> cosmic::iced::Subscription { + self.updates + .as_ref() + .map_or_else(cosmic::iced::Subscription::none, |events| { + cosmic::iced::Subscription::run_with(events.clone(), Events::stream) + }) + } + + pub fn elapsed_subscription(&self) -> cosmic::iced::Subscription<()> { + fn stream(deadline: &Instant) -> impl cosmic::iced::futures::Stream + use<> { + let deadline = *deadline; + cosmic::iced::futures::stream::once(async move { + tokio::time::sleep_until(deadline.into()).await; + }) } - if done { + self.elapsed_deadline(Instant::now()) + .map_or_else(cosmic::iced::Subscription::none, |deadline| { + cosmic::iced::Subscription::run_with(deadline, stream) + }) + } + + fn elapsed_deadline(&self, now: Instant) -> Option { + let State::Recording(start) = self.state else { + return None; + }; + Some(start + Duration::from_secs(now.saturating_duration_since(start).as_secs() + 1)) + } + + pub fn handle(&mut self, event: Event) { + if self.apply(event) { match &self.state { State::Saved(path) => notify("Recording saved", path.display().to_string()), State::Error(message) => notify("Recording failed", message.clone()), _ => (), } - self.control.take(); - self.updates.take(); } } + + fn apply(&mut self, event: Event) -> bool { + match event { + Event::Update(update) => match update.state.as_str() { + "selecting" if self.control.is_some() => self.state = State::Selecting, + "recording" if self.control.is_some() => { + if !matches!(self.state, State::Recording(_)) { + self.state = State::Recording(Instant::now()); + } + } + "saving" => self.state = State::Saving, + "saved" => { + self.state = update + .path + .map(State::Saved) + .unwrap_or_else(|| State::Error("Recorder returned no saved file".into())) + } + "cancelled" => self.state = State::Idle, + "error" => self.state = State::Error(update.message.unwrap_or_else(|| "Recording failed".into())), + _ => (), + }, + Event::Exited => { + if matches!(self.state, State::Selecting | State::Recording(_) | State::Saving) { + self.state = State::Error("The recorder stopped unexpectedly".into()); + } + self.control.take(); + self.updates.take(); + return true; + } + } + false + } +} + +fn read_events(mut child: Child, send: tokio::sync::mpsc::Sender) { + let mut output = BufReader::new(child.stdout.take().expect("piped recorder output")); + + loop { + let mut line = String::new(); + match output.by_ref().take(65537).read_line(&mut line) { + Ok(0) | Err(_) => break, + Ok(_) if line.len() > 65536 => break, + Ok(_) => { + if let Ok(update) = serde_json::from_str::(&line) + && send.blocking_send(Event::Update(update)).is_err() + { + break; + } + } + } + } + + // A rejected status line may leave the child blocked writing into stdout. + // Close our reader before waiting so that writer can observe the disconnect. + drop(output); + let _ = child.wait(); + let _ = send.blocking_send(Event::Exited); } // Use the normal notification center for results, without an extra recording menu. @@ -195,3 +246,211 @@ fn notify(title: &'static str, body: String) { } }); } + +#[cfg(test)] +mod tests { + use super::*; + use cosmic::iced::futures::{FutureExt, StreamExt}; + + #[test] + fn oversized_status_closes_stdout_before_waiting_for_exit() { + let child = Command::new("head") + .args(["-c", "16777216", "/dev/zero"]) + .stdout(Stdio::piped()) + .stderr(Stdio::null()) + .spawn() + .unwrap(); + let pid = child.id(); + let (send, mut receive) = tokio::sync::mpsc::channel(8); + let (done_send, done_receive) = std::sync::mpsc::channel(); + let worker = std::thread::spawn(move || { + read_events(child, send); + done_send.send(()).unwrap(); + }); + let done = done_receive.recv_timeout(Duration::from_secs(2)); + + if done.is_err() { + // Reap a stuck test child before reporting a failure. The reader + // still owns Child, so this PID cannot have been reused. + unsafe { libc::kill(pid as libc::pid_t, libc::SIGKILL) }; + } + + worker.join().unwrap(); + assert!( + done.is_ok(), + "reader kept the output pipe open while waiting for a blocked writer" + ); + assert!(matches!(receive.try_recv(), Ok(Event::Exited))); + } + + fn update(state: &str, path: Option) -> Event { + Event::Update(Update { + state: state.into(), + message: None, + path, + }) + } + + fn session() -> (Recorder, std::process::Child, tokio::sync::mpsc::Sender) { + // A real control pipe lets stop() exercise EOF without a recorder, + // portal, encoder, or D-Bus service. + let mut child = Command::new("cat") + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .spawn() + .unwrap(); + let (send, receive) = tokio::sync::mpsc::channel(8); + let recorder = Recorder { + state: State::Selecting, + control: child.stdin.take(), + updates: Some(Events(Arc::new(tokio::sync::Mutex::new(receive)))), + }; + (recorder, child, send) + } + + #[test] + fn active_recording_events_stop_by_eof_and_keep_terminal_result() { + let (mut recorder, mut child, send) = session(); + assert_eq!(recorder.subscription().units(), 1); + assert_eq!(recorder.elapsed_subscription().units(), 0); + recorder.apply(update("selecting", None)); + assert!(matches!(recorder.state, State::Selecting)); + recorder.apply(update("recording", None)); + let State::Recording(start) = recorder.state else { + panic!("recording event was lost") + }; + assert_eq!(recorder.elapsed_subscription().units(), 1); + recorder.apply(update("recording", None)); + assert!(matches!(recorder.state, State::Recording(same) if same == start)); + recorder.stop(); + assert!(matches!(recorder.state, State::Saving)); + assert!(recorder.control.is_none()); + assert!(recorder.busy()); + assert_eq!(recorder.elapsed_subscription().units(), 0); + assert!(child.wait().unwrap().success()); + recorder.apply(update("selecting", None)); + recorder.apply(update("recording", None)); + assert!(matches!(recorder.state, State::Saving)); + recorder.apply(update("saving", None)); + recorder.apply(update("saved", Some("display.webm".into()))); + assert!(recorder.apply(Event::Exited)); + assert!(matches!(recorder.state, State::Saved(ref path) if path == &PathBuf::from("display.webm"))); + assert!(!recorder.busy()); + assert_eq!(recorder.subscription().units(), 0); + assert!(send.try_send(Event::Exited).is_err()); + } + + #[test] + fn cancelling_selection_uses_eof_and_leaves_no_elapsed_or_event_timer() { + let (mut recorder, mut child, send) = session(); + recorder.stop(); + assert!(matches!(recorder.state, State::Saving)); + assert!(child.wait().unwrap().success()); + recorder.apply(update("cancelled", None)); + assert!(recorder.apply(Event::Exited)); + assert!(matches!(recorder.state, State::Idle)); + assert_eq!(recorder.subscription().units(), 0); + assert_eq!(recorder.elapsed_subscription().units(), 0); + assert!(send.try_send(Event::Exited).is_err()); + } + + #[test] + fn elapsed_deadlines_only_run_while_recording_and_align_to_start() { + let start = Instant::now(); + let mut recorder = Recorder::default(); + for state in [ + State::Idle, + State::Selecting, + State::Saving, + State::Saved("video.webm".into()), + State::Error("failed".into()), + ] { + recorder.state = state; + assert_eq!(recorder.elapsed_deadline(start), None); + } + recorder.state = State::Recording(start); + assert_eq!( + recorder.elapsed_deadline(start + Duration::from_millis(999)), + Some(start + Duration::from_secs(1)) + ); + assert_eq!( + recorder.elapsed_deadline(start + Duration::from_secs(1)), + Some(start + Duration::from_secs(2)) + ); + assert_eq!( + recorder.elapsed_deadline(start + Duration::from_millis(5200)), + Some(start + Duration::from_secs(6)) + ); + } + + #[test] + fn final_recording_results_survive_exit_and_release_session() { + for terminal in ["saved", "cancelled", "error"] { + let mut recorder = Recorder { + state: State::Saving, + ..Recorder::default() + }; + let (send, receive) = tokio::sync::mpsc::channel(8); + recorder.updates = Some(Events(Arc::new(tokio::sync::Mutex::new(receive)))); + recorder.apply(update(terminal, Some("video.webm".into()))); + assert!(recorder.busy()); + assert!(recorder.apply(Event::Exited)); + assert!(!recorder.busy()); + match terminal { + "saved" => { + assert!(matches!(recorder.state, State::Saved(ref path) if path == &PathBuf::from("video.webm"))) + } + "cancelled" => assert!(matches!(recorder.state, State::Idle)), + _ => assert!(matches!(recorder.state, State::Error(ref message) if message == "Recording failed")), + } + assert!(send.try_send(Event::Exited).is_err()); + } + } + + #[test] + fn missing_path_and_unexpected_exit_remain_errors() { + let mut recorder = Recorder::default(); + recorder.apply(update("saved", None)); + assert!(matches!(recorder.state, State::Error(ref message) if message == "Recorder returned no saved file")); + for state in [State::Selecting, State::Recording(Instant::now()), State::Saving] { + recorder.state = state; + recorder.apply(Event::Exited); + assert!( + matches!(recorder.state, State::Error(ref message) if message == "The recorder stopped unexpectedly") + ); + } + } + + #[test] + fn late_start_events_cannot_restart_stopped_recorder() { + let mut recorder = Recorder { + state: State::Saving, + ..Recorder::default() + }; + recorder.apply(update("selecting", None)); + recorder.apply(update("recording", None)); + assert!(matches!(recorder.state, State::Saving)); + } + + #[test] + fn event_stream_wakes_without_polling_and_keeps_bounded_order() { + let (send, receive) = tokio::sync::mpsc::channel(8); + let events = Events(Arc::new(tokio::sync::Mutex::new(receive))); + let stream = events.stream(); + cosmic::iced::futures::pin_mut!(stream); + assert!(stream.next().now_or_never().is_none()); + for _ in 0..8 { + send.try_send(update("saving", None)).unwrap(); + } + assert!(send.try_send(Event::Exited).is_err()); + for _ in 0..8 { + assert!( + matches!(stream.next().now_or_never(), Some(Some(Event::Update(Update { state, .. }))) if state == "saving") + ); + } + send.try_send(Event::Exited).unwrap(); + assert!(matches!(stream.next().now_or_never(), Some(Some(Event::Exited)))); + drop(send); + assert!(matches!(stream.next().now_or_never(), Some(None))); + } +}