diff --git a/Cargo.toml b/Cargo.toml index 3cfd081b..b0232301 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -44,8 +44,8 @@ composefs-ostree = { version = "0.7.0", path = "crates/composefs-ostree", defaul cap-std-ext = "5.1.2" ocidir = "0.8.0" -zlink = { version = "0.6", default-features = false, features = ["tokio", "introspection", "proxy", "server", "service", "tracing"] } -zlink-core = { version = "0.6", default-features = false, features = ["introspection", "std", "tracing"] } +zlink = { version = "0.7", default-features = false, features = ["tokio", "introspection", "proxy", "server", "service", "tracing"] } +zlink-core = { version = "0.7", default-features = false, features = ["introspection", "std", "tracing"] } [profile.dev.package.sha2] # this is *really* slow otherwise diff --git a/crates/composefs-ctl/src/lib.rs b/crates/composefs-ctl/src/lib.rs index 45687e02..57129cba 100644 --- a/crates/composefs-ctl/src/lib.rs +++ b/crates/composefs-ctl/src/lib.rs @@ -1373,8 +1373,8 @@ where /// Copy an OCI image (and all its layers) from one repository to another using varlink connections. #[cfg(feature = "oci")] pub async fn copy_image( - conn_src: &mut zlink::unix::Connection, - conn_dest: &mut zlink::unix::Connection, + conn_src: &mut zlink::tokio::unix::Connection, + conn_dest: &mut zlink::tokio::unix::Connection, handle_src: u64, handle_dest: u64, image: &OciReference, diff --git a/crates/composefs-ctl/src/varlink.rs b/crates/composefs-ctl/src/varlink.rs index e5135156..cfed573e 100644 --- a/crates/composefs-ctl/src/varlink.rs +++ b/crates/composefs-ctl/src/varlink.rs @@ -1888,11 +1888,11 @@ mod service_impl { #[derive(Debug)] pub(crate) struct ActivatedListener { /// The connection to yield on the first accept(), consumed after use. - conn: Option>, + conn: Option>, } impl zlink::Listener for ActivatedListener { - type Socket = zlink::unix::Stream; + type Socket = zlink::tokio::unix::Stream; async fn accept(&mut self) -> zlink::Result>> { match self.conn.take() { @@ -1909,7 +1909,7 @@ pub(crate) enum ActivatedSocket { Connected(ActivatedListener), /// A listening socket (systemd `.socket` with `Accept=no`, or the test /// harness): served with a normal accept loop. - Listening(zlink::unix::Listener), + Listening(zlink::tokio::unix::Listener), } /// Try to classify a socket-activation fd inherited from the service manager. @@ -1949,7 +1949,7 @@ pub(crate) fn try_activated_listener() -> Result> { if is_listening { // The fd is a bound, listening Unix socket. Hand it to zlink's // Listener adapter, which calls set_nonblocking and wraps it in tokio. - let listener = zlink::unix::Listener::try_from(owned) + let listener = zlink::tokio::unix::Listener::try_from(owned) .context("converting listening activation fd to zlink Listener")?; Ok(Some(ActivatedSocket::Listening(listener))) } else { @@ -1962,7 +1962,7 @@ pub(crate) fn try_activated_listener() -> Result> { let tokio_stream = tokio::net::UnixStream::from_std(std_stream) .context("converting systemd UnixStream to tokio")?; let zlink_stream = - zlink::unix::Stream::try_from(tokio_stream).map_err(|e| anyhow::anyhow!(e))?; + zlink::tokio::unix::Stream::try_from(tokio_stream).map_err(|e| anyhow::anyhow!(e))?; let conn = zlink::Connection::new(zlink_stream); Ok(Some(ActivatedSocket::Connected(ActivatedListener { conn: Some(conn), @@ -1981,7 +1981,7 @@ pub(crate) fn try_activated_listener() -> Result> { /// wrap exactly one `LocalSet`. pub(crate) async fn serve_activated(service: S, listener: ActivatedListener) -> Result<()> where - S: zlink::Service, + S: zlink::Service, { log::info!("Listening on systemd-activated socket"); let server = zlink::Server::new(listener, service); @@ -1991,14 +1991,17 @@ where .context("running varlink server (activated)") } -/// Serve `service` on a listening [`zlink::unix::Listener`] inside a +/// Serve `service` on a listening [`zlink::tokio::unix::Listener`] inside a /// [`tokio::task::LocalSet`]. /// /// Used for both the socket-activated listening fd path and the normal /// `bind`-a-fresh-socket path (see [`serve`]). -pub(crate) async fn serve_on_listener(service: S, listener: zlink::unix::Listener) -> Result<()> +pub(crate) async fn serve_on_listener( + service: S, + listener: zlink::tokio::unix::Listener, +) -> Result<()> where - S: zlink::Service, + S: zlink::Service, { let server = zlink::Server::new(listener, service); tokio::task::LocalSet::new() @@ -2016,7 +2019,7 @@ where /// 2. A freshly bound socket at `address` (which must be `Some`). pub(crate) async fn serve(service: S, address: Option<&Path>) -> Result<()> where - S: zlink::Service, + S: zlink::Service, { match try_activated_listener()? { Some(ActivatedSocket::Connected(l)) => return serve_activated(service, l).await, @@ -2027,7 +2030,7 @@ where None => {} } let address = address.context("no --address given and not socket-activated")?; - let listener = zlink::unix::bind(address) + let listener = zlink::tokio::unix::bind(address) .with_context(|| format!("binding varlink socket at {}", address.display()))?; log::info!("Listening on {}", address.display()); serve_on_listener(service, listener).await @@ -2937,7 +2940,7 @@ pub(crate) use oci::*; /// Spawn a `CfsctlService` in-process over a Unix socket pair for testing. /// -/// Returns a connected client [`zlink::unix::Connection`] and a +/// Returns a connected client [`zlink::tokio::unix::Connection`] and a /// [`std::thread::JoinHandle`] for the server thread. /// /// Mirrors the pattern in `composefs-storage`'s `spawn_in_process`: the zlink @@ -2948,14 +2951,14 @@ pub(crate) use oci::*; #[cfg(feature = "oci")] pub(crate) fn spawn_in_process( service: CfsctlService, -) -> std::io::Result<(zlink::unix::Connection, std::thread::JoinHandle<()>)> { +) -> std::io::Result<(zlink::tokio::unix::Connection, std::thread::JoinHandle<()>)> { let (client_std, server_std) = std::os::unix::net::UnixStream::pair()?; client_std.set_nonblocking(true)?; server_std.set_nonblocking(true)?; let client_stream = tokio::net::UnixStream::from_std(client_std)?; let client_zlink = - zlink::unix::Stream::try_from(client_stream).map_err(std::io::Error::other)?; + zlink::tokio::unix::Stream::try_from(client_stream).map_err(std::io::Error::other)?; let client_conn = zlink::Connection::new(client_zlink); let handle = std::thread::Builder::new() @@ -2980,7 +2983,7 @@ pub(crate) fn spawn_in_process( return; } }; - let server_zlink = match zlink::unix::Stream::try_from(server_stream) { + let server_zlink = match zlink::tokio::unix::Stream::try_from(server_stream) { Ok(s) => s, Err(e) => { log::error!("CfsctlService server zlink stream conversion failed: {e:#?}"); diff --git a/crates/composefs-fuse/Cargo.toml b/crates/composefs-fuse/Cargo.toml index 599abd9d..7544328e 100644 --- a/crates/composefs-fuse/Cargo.toml +++ b/crates/composefs-fuse/Cargo.toml @@ -17,7 +17,7 @@ default = ['pre-6.16'] [dependencies] anyhow = { version = "1.0.98", default-features = false } composefs = { workspace = true } -fuser = { version = "0.17.0", default-features = false } +fuser = { version = "0.18.0", default-features = false } log = { version = "0.4.8", default-features = false } memmap2 = { version = "0.9", default-features = false } rustix = { version = "1.0.0", default-features = false, features = ["fs", "mount"] } diff --git a/crates/composefs-integration-tests/src/tests/copy_image.rs b/crates/composefs-integration-tests/src/tests/copy_image.rs index f2e90cba..edb2e962 100644 --- a/crates/composefs-integration-tests/src/tests/copy_image.rs +++ b/crates/composefs-integration-tests/src/tests/copy_image.rs @@ -278,7 +278,7 @@ impl VarlinkProc { /// Open `repo_path` over an existing connection and return the handle. /// /// The handle is reused for all subsequent calls on the same connection. -async fn open_repo(conn: &mut zlink::unix::Connection, repo_path: &Path) -> Result { +async fn open_repo(conn: &mut zlink::tokio::unix::Connection, repo_path: &Path) -> Result { let path_str = repo_path.to_str().context("repo path is not valid UTF-8")?; let reply = conn .open_repository(Some(path_str), None, None) @@ -325,10 +325,10 @@ fn test_copy_image_via_varlink() -> Result<()> { let image: composefs_ctl::OciReference = "copyme:v1".parse()?; let finalize_reply = rt.block_on(async { - let mut conn_a = zlink::unix::connect(svc_a.socket()) + let mut conn_a = zlink::tokio::unix::connect(svc_a.socket()) .await .context("connecting to server A")?; - let mut conn_b = zlink::unix::connect(svc_b.socket()) + let mut conn_b = zlink::tokio::unix::connect(svc_b.socket()) .await .context("connecting to server B")?; @@ -432,8 +432,8 @@ fn test_copy_image_cross_algorithm() -> Result<()> { let image: composefs_ctl::OciReference = "copyme:v1".parse()?; let finalize_reply = rt.block_on(async { - let mut conn_a = zlink::unix::connect(svc_a.socket()).await?; - let mut conn_b = zlink::unix::connect(svc_b.socket()).await?; + let mut conn_a = zlink::tokio::unix::connect(svc_a.socket()).await?; + let mut conn_b = zlink::tokio::unix::connect(svc_b.socket()).await?; let handle_a = open_repo(&mut conn_a, &repo_a_path).await?; let handle_b = open_repo(&mut conn_b, &repo_b_path).await?; diff --git a/crates/composefs-integration-tests/src/tests/varlink.rs b/crates/composefs-integration-tests/src/tests/varlink.rs index 5847dd79..b1d38431 100644 --- a/crates/composefs-integration-tests/src/tests/varlink.rs +++ b/crates/composefs-integration-tests/src/tests/varlink.rs @@ -216,8 +216,8 @@ impl VarlinkService { // bare `is_err()` (that also catches transport failures) could not. /// Connect a fresh zlink client to the service socket. - async fn connect(&self) -> zlink::Result { - zlink::unix::connect(&self.socket).await + async fn connect(&self) -> zlink::Result { + zlink::tokio::unix::connect(&self.socket).await } /// `org.composefs.Oci.Inspect` via the typed proxy, using the cached handle. diff --git a/crates/composefs-oci/Cargo.toml b/crates/composefs-oci/Cargo.toml index 084f398f..287e4605 100644 --- a/crates/composefs-oci/Cargo.toml +++ b/crates/composefs-oci/Cargo.toml @@ -21,7 +21,7 @@ varlink = ['dep:zlink', 'dep:zlink-core', 'composefs/varlink'] anyhow = { version = "1.0.87", default-features = false } fn-error-context = "0.2" async-compression = { version = "0.4.0", default-features = false, features = ["tokio", "zstd", "gzip"] } -base64 = { version = "0.22", default-features = false, features = ["std"], optional = true } +base64 = { version = "0.23", default-features = false, features = ["std"], optional = true } composefs-splitdirfdstream = { path = "../composefs-splitdirfdstream", version = "0.7.0", features = ["tokio"] } bytes = { version = "1", default-features = false } composefs = { workspace = true } diff --git a/crates/composefs-oci/src/cstor.rs b/crates/composefs-oci/src/cstor.rs index 84731e9f..336f5974 100644 --- a/crates/composefs-oci/src/cstor.rs +++ b/crates/composefs-oci/src/cstor.rs @@ -366,7 +366,7 @@ async fn import_from_containers_storage_direct( #[allow(clippy::too_many_arguments)] async fn import_layer_via_transfer( repo: &Arc>, - client: &mut zlink::unix::Connection, + client: &mut zlink::tokio::unix::Connection, storage_path: &str, storage_layer_id: &str, diff_id: &OciDigest, diff --git a/crates/composefs-ostree/Cargo.toml b/crates/composefs-ostree/Cargo.toml index e7a583be..8694b0f4 100644 --- a/crates/composefs-ostree/Cargo.toml +++ b/crates/composefs-ostree/Cargo.toml @@ -12,7 +12,7 @@ version.workspace = true [dependencies] anyhow = { version = "1.0.87", default-features = false } -base64 = { version = "0.22", default-features = false, features = ["std"] } +base64 = { version = "0.23", default-features = false, features = ["std"] } bsdiff = { version = "0.2.1", default-features = false } chrono = { version = "0.4", default-features = false, features = ["alloc"] } cap-std = { version = "4.0.0", default-features = false } diff --git a/crates/composefs-storage/Cargo.toml b/crates/composefs-storage/Cargo.toml index bf649c46..5b9ac8bc 100644 --- a/crates/composefs-storage/Cargo.toml +++ b/crates/composefs-storage/Cargo.toml @@ -12,7 +12,7 @@ version.workspace = true [dependencies] anyhow = { version = "1.0", default-features = false, features = ["std"] } -base64 = { version = "0.22", default-features = false, features = ["std"] } +base64 = { version = "0.23", default-features = false, features = ["std"] } cap-std = { version = "4.0", default-features = false } cap-std-ext = { version = "5.0", default-features = false } composefs-splitdirfdstream = { path = "../composefs-splitdirfdstream", version = "0.7.0", optional = true, features = ["tokio"] } diff --git a/crates/composefs-storage/src/cstor_service.rs b/crates/composefs-storage/src/cstor_service.rs index 39ad76ab..1b053109 100644 --- a/crates/composefs-storage/src/cstor_service.rs +++ b/crates/composefs-storage/src/cstor_service.rs @@ -504,18 +504,18 @@ impl Drop for InProcessServer { /// Spawn a [`CstorLayerService`] in-process over a Unix socket pair. /// -/// Returns a connected `zlink::unix::Connection` and an [`InProcessServer`] +/// Returns a connected `zlink::tokio::unix::Connection` and an [`InProcessServer`] /// handle for shutdown. pub fn spawn_in_process( service: CstorLayerService, -) -> std::io::Result<(zlink::unix::Connection, InProcessServer)> { +) -> std::io::Result<(zlink::tokio::unix::Connection, InProcessServer)> { let (client_std, server_std) = std::os::unix::net::UnixStream::pair()?; client_std.set_nonblocking(true)?; server_std.set_nonblocking(true)?; let client_stream = tokio::net::UnixStream::from_std(client_std)?; let client_zlink = - zlink::unix::Stream::try_from(client_stream).map_err(std::io::Error::other)?; + zlink::tokio::unix::Stream::try_from(client_stream).map_err(std::io::Error::other)?; let client_conn = zlink::Connection::new(client_zlink); let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel::<()>(); @@ -544,7 +544,7 @@ pub fn spawn_in_process( return; } }; - let server_zlink = match zlink::unix::Stream::try_from(server_stream) { + let server_zlink = match zlink::tokio::unix::Stream::try_from(server_stream) { Ok(s) => s, Err(e) => { tracing::error!( @@ -596,7 +596,7 @@ pub fn serve_on_socket_blocking(socket: std::os::unix::net::UnixStream) -> std:: local.block_on(&rt, async move { // `from_std` must be called inside the runtime context. let stream = tokio::net::UnixStream::from_std(socket)?; - let zs = zlink::unix::Stream::try_from(stream).map_err(std::io::Error::other)?; + let zs = zlink::tokio::unix::Stream::try_from(stream).map_err(std::io::Error::other)?; let listener = zlink::ReadyListener::new(zs); let server = zlink::Server::new(listener, CstorLayerService); server @@ -1021,7 +1021,7 @@ mod tests { /// Collect all frames from a streaming get_layer call. async fn collect_get_layer( - client: &mut zlink::unix::Connection, + client: &mut zlink::tokio::unix::Connection, storage_path: &str, layer_id: &str, ) -> (GetLayerReply, Vec, usize) { diff --git a/crates/composefs-storage/src/userns_helper.rs b/crates/composefs-storage/src/userns_helper.rs index 81782678..c0dc42f5 100644 --- a/crates/composefs-storage/src/userns_helper.rs +++ b/crates/composefs-storage/src/userns_helper.rs @@ -152,7 +152,7 @@ pub fn init_if_helper() { /// /// When the caller cannot bypass file permissions, `StorageProxy::spawn()` /// starts a helper process inside a user namespace using `podman unshare` and -/// returns a connected [`zlink::unix::Connection`] the caller can use with the +/// returns a connected [`zlink::tokio::unix::Connection`] the caller can use with the /// [`composefs_oci::varlink_types::OciProxy`] trait to stream layers. /// /// # Dependency on `podman` @@ -174,7 +174,7 @@ pub struct StorageProxy { /// `shutdown` can take ownership and call `wait()` without fighting the /// `Drop` impl. child: Option, - conn: zlink::unix::Connection, + conn: zlink::tokio::unix::Connection, } impl std::fmt::Debug for StorageProxy { @@ -238,7 +238,7 @@ impl StorageProxy { parent_sock.set_nonblocking(true)?; let tok = tokio::net::UnixStream::from_std(parent_sock) .map_err(|e| HelperError::Ipc(format!("failed to convert socket: {e}")))?; - let zs = zlink::unix::Stream::try_from(tok).map_err(std::io::Error::other)?; + let zs = zlink::tokio::unix::Stream::try_from(tok).map_err(std::io::Error::other)?; let conn = zlink::Connection::new(zs); Ok(Self { @@ -252,7 +252,7 @@ impl StorageProxy { /// Callers use this with the `OciProxy` trait (from /// `composefs_oci::varlink_types`) to drive `GetLayer` calls over the /// helper connection. - pub fn connection(&mut self) -> &mut zlink::unix::Connection { + pub fn connection(&mut self) -> &mut zlink::tokio::unix::Connection { &mut self.conn }