diff --git a/README.md b/README.md index a4cfb6e..9def218 100644 --- a/README.md +++ b/README.md @@ -191,6 +191,12 @@ breakd coop join 'wss://breaks.example.net/ws#breakd=' breakd coop status ``` +When using Tailscale, use the host's MagicDNS name in the relay URL rather than +its numeric `100.x.y.z` address. A machine shared between tailnets can appear +under different Tailscale IPv4 addresses to its owner and guest; MagicDNS maps +the same hostname correctly for both. The invite is consumed by `breakd coop +join` and will not join a room when opened in a browser. + The host remains authoritative: a guest's skip, postpone, pause, resume, reset, or manual-break command is sent to the host, and the resulting host snapshot is mirrored back. Both systems should have normal network time synchronization diff --git a/crates/daemon/src/lib.rs b/crates/daemon/src/lib.rs index e806704..49927b8 100644 --- a/crates/daemon/src/lib.rs +++ b/crates/daemon/src/lib.rs @@ -1,5 +1,8 @@ use std::{ collections::VecDeque, + ffi::OsString, + fs, + os::unix::fs::FileTypeExt, path::{Path, PathBuf}, process::Stdio, }; @@ -248,6 +251,7 @@ pub async fn run() -> Result<()> { if coop.holds_local_schedule() { continue; } + tracing::info!(?event, "logind state changed"); let now = clock.sample()?; let scheduler_event = match event { PowerEvent::PreparingForSleep => SchedulerEvent::SuspendStarted, @@ -655,6 +659,7 @@ fn tray_state(status: SchedulerStatus) -> TrayState { remaining_seconds: status .remaining_ms .map(|milliseconds| milliseconds.saturating_add(999) / 1_000), + awaiting_resume: status.awaiting_resume, can_skip: status.can_skip, can_postpone: status.can_postpone, } @@ -849,15 +854,18 @@ impl OverlaySupervisor { self.stop_any().await; let executable = std::env::current_exe()?; let serialized = serde_json::to_string(&spec)?; - let child = TokioCommand::new(executable) + let wayland_display = overlay_wayland_display()?; + let mut command = TokioCommand::new(executable); + command .arg("overlay") .env("BREAKD_OVERLAY_SPEC", serialized) + .env("WAYLAND_DISPLAY", &wayland_display) + .env("XDG_SESSION_TYPE", "wayland") .stdin(Stdio::null()) .stdout(Stdio::null()) .stderr(Stdio::inherit()) - .kill_on_drop(true) - .spawn() - .context("spawn overlay child")?; + .kill_on_drop(true); + let child = command.spawn().context("spawn overlay child")?; self.active = Some((spec.session_id, child)); Ok(()) } @@ -892,6 +900,58 @@ impl OverlaySupervisor { } } +fn overlay_wayland_display() -> Result { + if let Some(display) = std::env::var_os("WAYLAND_DISPLAY").filter(|value| !value.is_empty()) { + return Ok(display); + } + + let runtime_dir = std::env::var_os("XDG_RUNTIME_DIR") + .map(PathBuf::from) + .context("WAYLAND_DISPLAY is unset and XDG_RUNTIME_DIR is unavailable")?; + discover_wayland_display(&runtime_dir) +} + +fn discover_wayland_display(runtime_dir: &Path) -> Result { + let mut candidates = fs::read_dir(runtime_dir) + .with_context(|| { + format!( + "inspect Wayland runtime directory {}", + runtime_dir.display() + ) + })? + .filter_map(|entry| entry.ok()) + .filter(|entry| { + entry + .file_name() + .to_str() + .is_some_and(|name| name.starts_with("wayland-")) + && entry.file_type().is_ok_and(|kind| kind.is_socket()) + }) + .map(|entry| entry.file_name()) + .collect::>(); + candidates.sort(); + + match candidates.as_slice() { + [display_name] => { + tracing::info!(display = %display_name.to_string_lossy(), "discovered Wayland display for overlay"); + Ok(display_name.clone()) + } + [] => anyhow::bail!( + "WAYLAND_DISPLAY is unset and no Wayland socket exists in {}", + runtime_dir.display() + ), + displays => anyhow::bail!( + "WAYLAND_DISPLAY is unset and multiple Wayland sockets exist in {}: {}", + runtime_dir.display(), + displays + .iter() + .map(|display| display.to_string_lossy()) + .collect::>() + .join(", ") + ), + } +} + fn command_message(command: &Command) -> &'static str { match command { Command::Pause { .. } => "schedule paused", @@ -921,6 +981,8 @@ pub fn socket_exists(path: &Path) -> bool { #[cfg(test)] mod tests { + use std::os::unix::net::UnixListener; + use super::*; #[test] @@ -930,4 +992,21 @@ mod tests { Some(PathBuf::from("/nix/store/hash-breakd/share/breakd")) ); } + + #[test] + fn discovers_the_only_live_wayland_socket() { + let runtime_dir = + std::env::temp_dir().join(format!("breakd-wayland-discovery-{}", uuid::Uuid::new_v4())); + fs::create_dir(&runtime_dir).unwrap(); + fs::write(runtime_dir.join("wayland-1.lock"), "ignored").unwrap(); + let listener = UnixListener::bind(runtime_dir.join("wayland-1")).unwrap(); + + assert_eq!( + discover_wayland_display(&runtime_dir).unwrap(), + OsString::from("wayland-1") + ); + + drop(listener); + fs::remove_dir_all(runtime_dir).unwrap(); + } } diff --git a/crates/platform-linux/src/logind.rs b/crates/platform-linux/src/logind.rs index 17a11a5..3a14f65 100644 --- a/crates/platform-linux/src/logind.rs +++ b/crates/platform-linux/src/logind.rs @@ -29,11 +29,8 @@ trait LoginManager { interface = "org.freedesktop.login1.Session" )] trait LoginSession { - #[zbus(signal)] - fn lock(&self) -> zbus::Result<()>; - - #[zbus(signal)] - fn unlock(&self) -> zbus::Result<()>; + #[zbus(property)] + fn locked_hint(&self) -> zbus::Result; } pub async fn spawn_logind_monitor(sender: mpsc::Sender) -> zbus::Result<()> { @@ -48,8 +45,7 @@ pub async fn spawn_logind_monitor(sender: mpsc::Sender) -> zbus::Res .build() .await?; let mut sleep_events = manager.receive_prepare_for_sleep().await?; - let mut lock_events = session.receive_lock().await?; - let mut unlock_events = session.receive_unlock().await?; + let mut locked_events = session.receive_locked_hint_changed().await; tokio::spawn(async move { loop { @@ -64,8 +60,16 @@ pub async fn spawn_logind_monitor(sender: mpsc::Sender) -> zbus::Res } } } - Some(_) = lock_events.next() => Some(PowerEvent::Locked), - Some(_) = unlock_events.next() => Some(PowerEvent::Unlocked), + Some(change) = locked_events.next() => { + match change.get().await { + Ok(true) => Some(PowerEvent::Locked), + Ok(false) => Some(PowerEvent::Unlocked), + Err(error) => { + tracing::warn!(%error, "invalid logind lock state"); + None + } + } + }, else => break, }; if let Some(event) = event diff --git a/crates/tray/src/lib.rs b/crates/tray/src/lib.rs index e7752e2..cc96ddb 100644 --- a/crates/tray/src/lib.rs +++ b/crates/tray/src/lib.rs @@ -13,6 +13,7 @@ pub struct TrayState { pub paused: bool, pub active_kind: Option, pub remaining_seconds: Option, + pub awaiting_resume: bool, pub can_skip: bool, pub can_postpone: bool, } @@ -22,6 +23,14 @@ impl TrayState { if self.paused { return "Schedule paused".into(); } + if self.awaiting_resume { + return match self.active_kind { + Some(BreakKind::Mini) => "Mini break complete — resume when ready".into(), + Some(BreakKind::Long) => "Long break complete — resume when ready".into(), + Some(BreakKind::Rest) => "Rest break complete — resume when ready".into(), + None => "Break complete — resume when ready".into(), + }; + } let remaining = self .remaining_seconds .map(format_duration) @@ -189,6 +198,11 @@ impl ksni::Tray for BreakdTray { self.command_item("Start rest break", Command::Rest, !has_active_break), self.command_item("Skip break", Command::Skip, self.state.can_skip), self.command_item("Postpone break", Command::Postpone, self.state.can_postpone), + self.command_item( + "Resume work", + Command::ResumeBreak, + self.state.awaiting_resume, + ), ksni::MenuItem::Separator, self.command_item( "Reset schedule", @@ -229,6 +243,7 @@ mod tests { paused: false, active_kind: None, remaining_seconds: Some(65), + awaiting_resume: false, can_skip: false, can_postpone: false, }; @@ -253,6 +268,19 @@ mod tests { ..running }; assert_eq!(paused.status_text(), "Schedule paused"); + + let waiting = TrayState { + paused: false, + active_kind: Some(BreakKind::Long), + remaining_seconds: Some(0), + awaiting_resume: true, + can_skip: false, + can_postpone: false, + }; + assert_eq!( + waiting.status_text(), + "Long break complete — resume when ready" + ); } #[tokio::test] @@ -263,6 +291,7 @@ mod tests { paused: false, active_kind: None, remaining_seconds: Some(60), + awaiting_resume: false, can_skip: false, can_postpone: false, }, @@ -288,6 +317,7 @@ mod tests { paused: false, active_kind: None, remaining_seconds: Some(60), + awaiting_resume: false, can_skip: false, can_postpone: false, }, @@ -301,4 +331,31 @@ mod tests { (item.activate)(&mut tray); assert_eq!(receiver.recv().await, Some(TrayAction::OpenSettings)); } + + #[tokio::test] + async fn completed_break_menu_item_sends_resume_break() { + let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel(); + let mut tray = BreakdTray { + state: TrayState { + paused: false, + active_kind: Some(BreakKind::Long), + remaining_seconds: Some(0), + awaiting_resume: true, + can_skip: false, + can_postpone: false, + }, + sender, + name: "breakd-dev".into(), + }; + let menu = ksni::Tray::menu(&tray); + let ksni::MenuItem::Standard(item) = menu.into_iter().nth(8).unwrap() else { + panic!("expected resume work menu item"); + }; + assert!(item.enabled); + (item.activate)(&mut tray); + assert_eq!( + receiver.recv().await, + Some(TrayAction::Command(Command::ResumeBreak)) + ); + } } diff --git a/docs/coop.md b/docs/coop.md index a0ae211..cf62762 100644 --- a/docs/coop.md +++ b/docs/coop.md @@ -21,6 +21,22 @@ cargo run -p breakd-relay -- --listen 127.0.0.1:8787 breakd coop host --relay ws://127.0.0.1:8787/ws ``` +For a private Tailscale room, listen on the host's Tailscale interface (or on +`0.0.0.0` with a firewall rule restricted to `tailscale0`) and use the host's +MagicDNS name in the relay URL: + +```bash +breakd-relay --listen 0.0.0.0:8787 +breakd coop host --relay ws://my-host.my-tailnet.ts.net:8787/ws +``` + +Prefer the MagicDNS name over a copied `100.x.y.z` address when a machine is +shared between tailnets. Tailscale can present that same machine under different +IPv4 addresses to its owner and to a shared user, while the MagicDNS name maps to +the correct address for each person. The invite is a `breakd` CLI value, not a +web page: the guest must run `breakd coop join ''` rather than +opening it in a browser. + Plain `ws://` exposes the room token to the network. Use it only on localhost or another trusted, encrypted network. For internet use, keep the process bound to localhost and put a TLS reverse proxy in front of it. For example, a Caddy site @@ -71,6 +87,12 @@ Run `breakd coop host` again to make a new token and invalidate the previous roo from that host. `breakd coop leave` clears the relay URL and token and resets a fresh local schedule. +`breakd coop host` waits until the host is connected before printing its final +status. `breakd coop join` waits for the first authoritative host snapshot, so a +successful return means `connected`, `host_present`, and `following_host` are all +true. A timeout only stops the CLI wait; the daemon keeps reconnecting in the +background and `breakd coop status` shows its latest state and error. + ## Synchronization and failure behavior The host publishes at most one regular snapshot per second and immediately after diff --git a/src/main.rs b/src/main.rs index a8713ca..554ae8c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,8 +1,10 @@ use std::process::ExitCode; use anyhow::{Context, Result, bail}; -use breakd_core::{Command, DurationMs, OverlaySpec}; +use breakd_core::{Command, DurationMs, OverlaySpec, Response}; use clap::{Parser, Subcommand}; +use serde_json::Value; +use tokio::time::{Duration, Instant, sleep}; use tracing_subscriber::EnvFilter; #[derive(Debug, Parser)] @@ -126,10 +128,8 @@ async fn execute(arguments: Arguments) -> Result<()> { CliCommand::Outputs { json } => send(Command::Outputs, json).await, CliCommand::Doctor { json } => send(Command::Doctor, json).await, CliCommand::Coop { command } => match command { - CoopCommand::Host { relay } => { - send(Command::CoopHost { relay_url: relay }, false).await - } - CoopCommand::Join { invite } => send(Command::CoopJoin { invite }, false).await, + CoopCommand::Host { relay } => host_coop_room(relay).await, + CoopCommand::Join { invite } => join_coop_room(invite).await, CoopCommand::Leave => send(Command::CoopLeave, false).await, CoopCommand::Status { json } => send(Command::CoopStatus, json).await, }, @@ -138,6 +138,11 @@ async fn execute(arguments: Arguments) -> Result<()> { } async fn send(command: Command, json: bool) -> Result<()> { + let response = request(command).await?; + print_response(response, json) +} + +async fn request(command: Command) -> Result { let response = tokio::time::timeout( std::time::Duration::from_secs(2), breakd_ipc::request(breakd_config::socket_path(), command), @@ -147,6 +152,10 @@ async fn send(command: Command, json: bool) -> Result<()> { if !response.ok { bail!(response.message); } + Ok(response) +} + +fn print_response(response: Response, json: bool) -> Result<()> { if json { println!( "{}", @@ -161,6 +170,83 @@ async fn send(command: Command, json: bool) -> Result<()> { Ok(()) } +async fn host_coop_room(relay_url: String) -> Result<()> { + let response = request(Command::CoopHost { relay_url }).await?; + let invite = response + .data + .as_ref() + .and_then(|data| data.get("invite")) + .and_then(Value::as_str) + .context("daemon did not return a co-op invite")? + .to_owned(); + let status = wait_for_coop(CoopReadiness::Host).await?; + println!("co-op room created"); + println!( + "{}", + serde_json::to_string_pretty(&serde_json::json!({ + "invite": invite, + "status": status, + }))? + ); + Ok(()) +} + +async fn join_coop_room(invite: String) -> Result<()> { + request(Command::CoopJoin { invite }).await?; + let status = wait_for_coop(CoopReadiness::Guest).await?; + println!("joined co-op room"); + println!("{}", serde_json::to_string_pretty(&status)?); + Ok(()) +} + +#[derive(Debug, Clone, Copy)] +enum CoopReadiness { + Host, + Guest, +} + +async fn wait_for_coop(expected: CoopReadiness) -> Result { + let deadline = Instant::now() + Duration::from_secs(5); + let failed_status = loop { + sleep(Duration::from_millis(100)).await; + let response = request(Command::CoopStatus).await?; + let status = response.data.unwrap_or(Value::Null); + if coop_is_ready(&status, expected) { + return Ok(status); + } + if Instant::now() >= deadline { + break status; + } + }; + let detail = failed_status + .get("last_error") + .and_then(Value::as_str) + .map(|error| format!(": {error}")) + .unwrap_or_default(); + bail!( + "co-op connection did not become ready within 5 seconds{detail}; the daemon will keep retrying" + ) +} + +fn coop_is_ready(status: &Value, expected: CoopReadiness) -> bool { + let connected = status + .get("connected") + .and_then(Value::as_bool) + .unwrap_or(false); + let host_present = status + .get("host_present") + .and_then(Value::as_bool) + .unwrap_or(false); + let role_ready = match expected { + CoopReadiness::Host => true, + CoopReadiness::Guest => status + .get("following_host") + .and_then(Value::as_bool) + .unwrap_or(false), + }; + connected && host_present && role_ready +} + fn run_overlay() -> Result<()> { let encoded = std::env::var("BREAKD_OVERLAY_SPEC").context("BREAKD_OVERLAY_SPEC is unavailable")?; @@ -191,3 +277,26 @@ fn init_logging() { builder.compact().init(); } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn coop_join_waits_for_the_first_host_snapshot() { + let connecting = serde_json::json!({ + "connected": true, + "host_present": true, + "following_host": false, + }); + let following = serde_json::json!({ + "connected": true, + "host_present": true, + "following_host": true, + }); + + assert!(!coop_is_ready(&connecting, CoopReadiness::Guest)); + assert!(coop_is_ready(&following, CoopReadiness::Guest)); + assert!(coop_is_ready(&connecting, CoopReadiness::Host)); + } +}