niri_ipc::Socket; niri msg version; version checking on IPC (#278)

* Implement version checking in IPC

implement version checking; streamed IPC

streamed IPC will allow multiple requests per connection

add nonsense request

change inline struct to json macro

only check version if request actually fails

fix usage of inspect_err (MSRV 1.72.0; stabilized 1.76.0)

"nonsense request" -> "return error"

oneshot connections

* Change some things around

* Unqualify niri_ipc::Transform

---------

Co-authored-by: Ivan Molodetskikh <yalterz@gmail.com>

Cargo.lock

@@ -2202,6 +2202,7 @@ version = "0.1.4"
 dependencies = [
  "clap",
  "serde",
+ "serde_json",
 ]
 
 [[package]]

Cargo.toml

@@ -14,6 +14,7 @@ anyhow = "1.0.81"
 bitflags = "2.5.0"
 clap = { version = "~4.4.18", features = ["derive"] }
 serde = { version = "1.0.197", features = ["derive"] }
+serde_json = "1.0.115"
 tracing = { version = "0.1.40", features = ["max_level_trace", "release_max_level_debug"] }
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 tracy-client = { version = "0.17.0", default-features = false }
@@ -67,7 +68,7 @@ portable-atomic = { version = "1.6.0", default-features = false, features = ["fl
 profiling = "1.0.15"
 sd-notify = "0.4.1"
 serde.workspace = true
-serde_json = "1.0.115"
+serde_json.workspace = true
 smithay-drm-extras.workspace = true
 tracing-subscriber.workspace = true
 tracing.workspace = true

niri-ipc/Cargo.toml

@@ -10,6 +10,7 @@ repository.workspace = true
 [dependencies]
 clap = { workspace = true, optional = true }
 serde.workspace = true
+serde_json.workspace = true
 
 [features]
 clap = ["dep:clap"]

niri-ipc/src/lib.rs

@@ -6,18 +6,22 @@ use std::str::FromStr;
 
 use serde::{Deserialize, Serialize};
 
-/// Name of the environment variable containing the niri IPC socket path.
+mod socket;
-pub const SOCKET_PATH_ENV: &str = "NIRI_SOCKET";
+pub use socket::{Socket, SOCKET_PATH_ENV};
 
 /// Request from client to niri.
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub enum Request {
+    /// Request the version string for the running niri instance.
+    Version,
     /// Request information about connected outputs.
     Outputs,
     /// Request information about the focused window.
     FocusedWindow,
     /// Perform an action.
     Action(Action),
+    /// Respond with an error (for testing error handling).
+    ReturnError,
 }
 
 /// Reply from niri to client.
@@ -35,6 +39,8 @@ pub type Reply = Result<Response, String>;
 pub enum Response {
     /// A request that does not need a response was handled successfully.
     Handled,
+    /// The version string for the running niri instance.
+    Version(String),
     /// Information about connected outputs.
     ///
     /// Map from connector name to output info.

niri-ipc/src/socket.rs

@@ -0,0 +1,63 @@
+//! Helper for blocking communication over the niri socket.
+
+use std::env;
+use std::io::{self, Read, Write};
+use std::net::Shutdown;
+use std::os::unix::net::UnixStream;
+use std::path::Path;
+
+use crate::{Reply, Request};
+
+/// Name of the environment variable containing the niri IPC socket path.
+pub const SOCKET_PATH_ENV: &str = "NIRI_SOCKET";
+
+/// Helper for blocking communication over the niri socket.
+///
+/// This struct is used to communicate with the niri IPC server. It handles the socket connection
+/// and serialization/deserialization of messages.
+pub struct Socket {
+    stream: UnixStream,
+}
+
+impl Socket {
+    /// Connects to the default niri IPC socket.
+    ///
+    /// This is equivalent to calling [`Self::connect_to`] with the path taken from the
+    /// [`SOCKET_PATH_ENV`] environment variable.
+    pub fn connect() -> io::Result<Self> {
+        let socket_path = env::var_os(SOCKET_PATH_ENV).ok_or_else(|| {
+            io::Error::new(
+                io::ErrorKind::NotFound,
+                format!("{SOCKET_PATH_ENV} is not set, are you running this within niri?"),
+            )
+        })?;
+        Self::connect_to(socket_path)
+    }
+
+    /// Connects to the niri IPC socket at the given path.
+    pub fn connect_to(path: impl AsRef<Path>) -> io::Result<Self> {
+        let stream = UnixStream::connect(path.as_ref())?;
+        Ok(Self { stream })
+    }
+
+    /// Sends a request to niri and returns the response.
+    ///
+    /// Return values:
+    ///
+    /// * `Ok(Ok(response))`: successful [`Response`](crate::Response) from niri
+    /// * `Ok(Err(message))`: error message from niri
+    /// * `Err(error)`: error communicating with niri
+    pub fn send(self, request: Request) -> io::Result<Reply> {
+        let Self { mut stream } = self;
+
+        let mut buf = serde_json::to_vec(&request).unwrap();
+        stream.write_all(&buf)?;
+        stream.shutdown(Shutdown::Write)?;
+
+        buf.clear();
+        stream.read_to_end(&mut buf)?;
+
+        let reply = serde_json::from_slice(&buf)?;
+        Ok(reply)
+    }
+}

src/cli.rs

@@ -52,6 +52,8 @@ pub enum Sub {
 
 #[derive(Subcommand)]
 pub enum Msg {
+    /// Print the version of the running niri instance.
+    Version,
     /// List connected outputs.
     Outputs,
     /// Print information about the focused window.
@@ -61,4 +63,6 @@ pub enum Msg {
         #[command(subcommand)]
         action: Action,
     },
+    /// Request an error from the running niri instance.
+    RequestError,
 }

src/ipc/client.rs

@@ -1,54 +1,99 @@
-use std::env;
-use std::io::{Read, Write};
-use std::net::Shutdown;
-use std::os::unix::net::UnixStream;
-
 use anyhow::{anyhow, bail, Context};
-use niri_ipc::{LogicalOutput, Mode, Output, Reply, Request, Response};
+use niri_ipc::{LogicalOutput, Mode, Output, Request, Response, Socket, Transform};
+use serde_json::json;
 
 use crate::cli::Msg;
+use crate::utils::version;
 
 pub fn handle_msg(msg: Msg, json: bool) -> anyhow::Result<()> {
-    let socket_path = env::var_os(niri_ipc::SOCKET_PATH_ENV).with_context(|| {
-        format!(
-            "{} is not set, are you running this within niri?",
-            niri_ipc::SOCKET_PATH_ENV
-        )
-    })?;
-
-    let mut stream =
-        UnixStream::connect(socket_path).context("error connecting to {socket_path}")?;
-
     let request = match &msg {
+        Msg::Version => Request::Version,
         Msg::Outputs => Request::Outputs,
         Msg::FocusedWindow => Request::FocusedWindow,
         Msg::Action { action } => Request::Action(action.clone()),
+        Msg::RequestError => Request::ReturnError,
     };
-    let mut buf = serde_json::to_vec(&request).unwrap();
-    stream
-        .write_all(&buf)
-        .context("error writing IPC request")?;
-    stream
-        .shutdown(Shutdown::Write)
-        .context("error closing IPC stream for writing")?;
 
-    buf.clear();
+    let socket = Socket::connect().context("error connecting to the niri socket")?;
-    stream
-        .read_to_end(&mut buf)
-        .context("error reading IPC response")?;
 
-    let reply: Reply = serde_json::from_slice(&buf).context("error parsing IPC reply")?;
+    let reply = socket
+        .send(request)
+        .context("error communicating with niri")?;
 
-    let response = reply
+    let compositor_version = match reply {
-        .map_err(|msg| anyhow!(msg))
+        Err(_) if !matches!(msg, Msg::Version) => {
-        .context("niri could not handle the request")?;
+            // If we got an error, it might be that the CLI is a different version from the running
+            // niri instance. Request the running instance version to compare and print a message.
+            Socket::connect()
+                .and_then(|socket| socket.send(Request::Version))
+                .ok()
+        }
+        _ => None,
+    };
 
     // Default SIGPIPE so that our prints don't panic on stdout closing.
     unsafe {
         libc::signal(libc::SIGPIPE, libc::SIG_DFL);
     }
 
+    let response = reply.map_err(|err_msg| {
+        // Check for CLI-server version mismatch to add helpful context.
+        match compositor_version {
+            Some(Ok(Response::Version(compositor_version))) => {
+                let cli_version = version();
+                if cli_version != compositor_version {
+                    eprintln!("Running niri compositor has a different version from the niri CLI:");
+                    eprintln!("Compositor version: {compositor_version}");
+                    eprintln!("CLI version:        {cli_version}");
+                    eprintln!("Did you forget to restart niri after an update?");
+                    eprintln!();
+                }
+            }
+            Some(_) => {
+                eprintln!("Unable to get the running niri compositor version.");
+                eprintln!("Did you forget to restart niri after an update?");
+                eprintln!();
+            }
+            None => {
+                // Communication error, or the original request was already a version request.
+                // Don't add irrelevant context.
+            }
+        }
+
+        anyhow!(err_msg).context("niri returned an error")
+    })?;
+
     match msg {
+        Msg::RequestError => {
+            bail!("unexpected response: expected an error, got {response:?}");
+        }
+        Msg::Version => {
+            let Response::Version(compositor_version) = response else {
+                bail!("unexpected response: expected Version, got {response:?}");
+            };
+
+            let cli_version = version();
+
+            if json {
+                println!(
+                    "{}",
+                    json!({
+                        "compositor": compositor_version,
+                        "cli": cli_version,
+                    })
+                );
+                return Ok(());
+            }
+
+            if cli_version != compositor_version {
+                eprintln!("Running niri compositor has a different version from the niri CLI.");
+                eprintln!("Did you forget to restart niri after an update?");
+                eprintln!();
+            }
+
+            println!("Compositor version: {compositor_version}");
+            println!("CLI version:        {cli_version}");
+        }
         Msg::Outputs => {
             let Response::Outputs(outputs) = response else {
                 bail!("unexpected response: expected Outputs, got {response:?}");
@@ -123,18 +168,14 @@ pub fn handle_msg(msg: Msg, json: bool) -> anyhow::Result<()> {
                     println!("  Scale: {scale}");
 
                     let transform = match transform {
-                        niri_ipc::Transform::Normal => "normal",
+                        Transform::Normal => "normal",
-                        niri_ipc::Transform::_90 => "90° counter-clockwise",
+                        Transform::_90 => "90° counter-clockwise",
-                        niri_ipc::Transform::_180 => "180°",
+                        Transform::_180 => "180°",
-                        niri_ipc::Transform::_270 => "270° counter-clockwise",
+                        Transform::_270 => "270° counter-clockwise",
-                        niri_ipc::Transform::Flipped => "flipped horizontally",
+                        Transform::Flipped => "flipped horizontally",
-                        niri_ipc::Transform::Flipped90 => {
+                        Transform::Flipped90 => "90° counter-clockwise, flipped horizontally",
-                            "90° counter-clockwise, flipped horizontally"
+                        Transform::Flipped180 => "flipped vertically",
-                        }
+                        Transform::Flipped270 => "270° counter-clockwise, flipped horizontally",
-                        niri_ipc::Transform::Flipped180 => "flipped vertically",
-                        niri_ipc::Transform::Flipped270 => {
-                            "270° counter-clockwise, flipped horizontally"
-                        }
                     };
                     println!("  Transform: {transform}");
                 }

src/ipc/server.rs

@@ -8,7 +8,7 @@ use calloop::io::Async;
 use directories::BaseDirs;
 use futures_util::io::{AsyncReadExt, BufReader};
 use futures_util::{AsyncBufReadExt, AsyncWriteExt};
-use niri_ipc::{Request, Response};
+use niri_ipc::{Reply, Request, Response};
 use smithay::desktop::Window;
 use smithay::reexports::calloop::generic::Generic;
 use smithay::reexports::calloop::{Interest, LoopHandle, Mode, PostAction};
@@ -18,6 +18,7 @@ use smithay::wayland::shell::xdg::XdgToplevelSurfaceData;
 
 use crate::backend::IpcOutputMap;
 use crate::niri::State;
+use crate::utils::version;
 
 pub struct IpcServer {
     pub socket_path: PathBuf,
@@ -114,10 +115,18 @@ async fn handle_client(ctx: ClientCtx, stream: Async<'_, UnixStream>) -> anyhow:
         .await
         .context("error reading request")?;
 
-    let reply = process(&ctx, &buf).map_err(|err| {
+    let request = serde_json::from_str(&buf)
+        .context("error parsing request")
+        .map_err(|err| err.to_string());
+    let requested_error = matches!(request, Ok(Request::ReturnError));
+
+    let reply = request.and_then(|request| process(&ctx, request));
+
+    if let Err(err) = &reply {
+        if !requested_error {
             warn!("error processing IPC request: {err:?}");
-        err.to_string()
+        }
-    });
+    }
 
     let buf = serde_json::to_vec(&reply).context("error formatting reply")?;
     write.write_all(&buf).await.context("error writing reply")?;
@@ -125,10 +134,10 @@ async fn handle_client(ctx: ClientCtx, stream: Async<'_, UnixStream>) -> anyhow:
     Ok(())
 }
 
-fn process(ctx: &ClientCtx, buf: &str) -> anyhow::Result<Response> {
+fn process(ctx: &ClientCtx, request: Request) -> Reply {
-    let request: Request = serde_json::from_str(buf).context("error parsing request")?;
-
     let response = match request {
+        Request::ReturnError => return Err(String::from("example compositor error")),
+        Request::Version => Response::Version(version()),
         Request::Outputs => {
             let ipc_outputs = ctx.ipc_outputs.lock().unwrap().clone();
             Response::Outputs(ipc_outputs)