fxfs_platform_testing/fuchsia/
remote_volume.rs

1// Copyright 2026 The Fuchsia Authors. All rights reserved.
2// Use of this source code is governed by a BSD-style license that can be
3// found in the LICENSE file.
4
5use crate::fuchsia::errors::map_to_status;
6use crate::fuchsia::file::FxFile;
7use crate::fuchsia::volume::FxVolume;
8use anyhow::Error;
9use fidl_fuchsia_fxfs as ffxfs;
10use fidl_fuchsia_io as fio;
11use fuchsia_sync::Mutex;
12use futures::StreamExt;
13use fxfs::log::error;
14use fxfs::object_store::ObjectStore;
15use std::sync::Arc;
16
17pub type UnmountFn =
18    Box<dyn FnOnce() -> futures::future::BoxFuture<'static, Result<(), Error>> + Send>;
19
20/// `FxRemoteVolume` manages a remotely served Fxfs volume for in-process co-location (e.g.
21/// Starnix). It handles the `fuchsia.fxfs.RemoteVolume` FIDL interface.
22pub struct FxRemoteVolume {
23    store: Arc<ObjectStore>,
24    volume: Option<Arc<FxVolume>>,
25    unmount_fn: Mutex<Option<UnmountFn>>,
26    scope: fuchsia_async::Scope,
27}
28
29#[fxfs_trace::trace]
30impl FxRemoteVolume {
31    pub fn new(
32        store: Arc<ObjectStore>,
33        volume: Option<Arc<FxVolume>>,
34        unmount_fn: Option<UnmountFn>,
35    ) -> Arc<Self> {
36        Arc::new(Self {
37            store,
38            volume,
39            unmount_fn: Mutex::new(unmount_fn),
40            scope: fuchsia_async::Scope::new_with_name("fxfs-remote-vol"),
41        })
42    }
43
44    pub fn store(&self) -> &Arc<ObjectStore> {
45        &self.store
46    }
47
48    pub fn serve(self: &Arc<Self>, stream: ffxfs::RemoteVolumeRequestStream) {
49        let this = self.clone();
50        self.scope.spawn(async move {
51            this.handle_requests(stream).await;
52        });
53    }
54
55    pub fn shutdown(&self) {
56        let _ = self.scope.to_handle().cancel();
57    }
58
59    pub async fn wait(&self) {
60        self.scope.on_no_tasks().await;
61    }
62
63    /// Returns the persistent layers from the ObjectStore's LSM tree as read-only VMOs.
64    pub async fn get_layer_set(&self) -> Result<ffxfs::LayerSet, Error> {
65        let layer_set = self.store.tree().immutable_layer_set();
66        let mut layer_files = Vec::new();
67        let rights =
68            zx::Rights::BASIC | zx::Rights::READ | zx::Rights::MAP | zx::Rights::GET_PROPERTY;
69        for layer in layer_set.layers.iter() {
70            if let Some(handle) = layer.handle() {
71                let object_id = handle.object_id();
72                let vmo = if let Some(vmo) = handle.get_vmo() {
73                    vmo.duplicate_handle(rights)?
74                } else {
75                    let size = handle.get_size();
76                    let vmo = zx::Vmo::create(size)?;
77                    vmo.set_content_size(&size)?;
78                    if size > 0 {
79                        let aligned_size = handle.block_size().align_up(size).unwrap() as usize;
80                        let mut buf = handle.allocate_buffer(aligned_size).await;
81                        let read_len = handle.read_aligned(0, buf.as_mut()).await?;
82                        vmo.write(&buf.subslice(..read_len).to_vec(), 0)?;
83                    }
84                    vmo.replace_handle(rights)?
85                };
86                layer_files.push(ffxfs::LayerFile { object_id, vmo });
87            }
88        }
89        Ok(ffxfs::LayerSet { layers: layer_files })
90    }
91
92    async fn unmount(&self) -> Result<(), Error> {
93        let unmount_fn = self.unmount_fn.lock().take();
94        if let Some(unmount_fn) = unmount_fn {
95            unmount_fn().await?;
96        } else if let Some(volume) = &self.volume {
97            volume.terminate().await;
98        }
99        Ok(())
100    }
101
102    /// Dispatches FIDL requests for `fuchsia.fxfs.RemoteVolume`.
103    pub async fn handle_requests(self: Arc<Self>, mut stream: ffxfs::RemoteVolumeRequestStream) {
104        let mut unmounted = false;
105        while let Some(request) = stream.next().await {
106            match request {
107                Ok(ffxfs::RemoteVolumeRequest::OpenFile { object_id, file, responder }) => {
108                    let res: Result<fio::FileInfo, zx::Status> = async {
109                        let volume = self.volume.as_ref().ok_or(zx::Status::NOT_SUPPORTED)?;
110                        // TODO: Support unnamed temporary files (which are in the graveyard).
111                        let opened_node =
112                            volume.open_file_by_id(object_id).await.map_err(map_to_status)?;
113                        let stream = zx::Stream::create(
114                            zx::StreamOptions::MODE_READ | zx::StreamOptions::MODE_WRITE,
115                            opened_node.vmo(),
116                            0,
117                        )?;
118                        let client_stream = stream.duplicate_handle(zx::Rights::SAME_RIGHTS)?;
119                        let file_info =
120                            fio::FileInfo { stream: Some(client_stream), ..Default::default() };
121                        let scope = volume.scope().clone();
122                        let flags =
123                            fio::Flags::PROTOCOL_FILE | fio::PERM_READABLE | fio::PERM_WRITABLE;
124                        let mut request = vfs::ObjectRequest::new(
125                            flags,
126                            &fio::Options::default(),
127                            file.into_channel(),
128                        );
129                        FxFile::create_connection_with_stream_async(
130                            opened_node,
131                            scope,
132                            flags,
133                            (&mut request).into(),
134                            stream,
135                        )
136                        .await?;
137                        Ok(file_info)
138                    }
139                    .await;
140                    let _ = responder.send(res.map_err(|s| s.into_raw()));
141                }
142
143                Ok(ffxfs::RemoteVolumeRequest::Close { responder }) => {
144                    let res = self.unmount().await;
145                    let _ = responder.send(res.map_err(|e| map_to_status(e).into_raw()));
146                    // NOTE: After calling this, this task might be dropped at the next await
147                    // point.
148                    self.shutdown();
149                    unmounted = true;
150                    break;
151                }
152
153                Err(e) => {
154                    error!(e:?; "RemoteVolume request stream error");
155                    break;
156                }
157            }
158        }
159        if !unmounted {
160            let _ = self.unmount().await;
161            self.shutdown();
162        }
163    }
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169    use fidl::endpoints::create_proxy;
170    use fxfs::filesystem::FxFilesystemBuilder;
171    use fxfs::object_handle::{ObjectHandle, WriteObjectHandle};
172    use fxfs::object_store::transaction::Options;
173    use fxfs::object_store::volume::root_volume;
174    use fxfs_insecure_crypto::new_insecure_crypt;
175    use storage_device::DeviceHolder;
176    use storage_device::fake_device::FakeDevice;
177
178    fn new_test_fs_builder() -> FxFilesystemBuilder {
179        FxFilesystemBuilder::new()
180    }
181
182    #[fuchsia::test]
183    async fn test_remote_volume_layer_set() {
184        let device = DeviceHolder::new(FakeDevice::new(16384, 512));
185        let fs = new_test_fs_builder().format(true).open(device).await.unwrap();
186        let root_vol = root_volume(fs.clone()).await.unwrap();
187        let store = root_vol.new_volume("test_remote_vol", Default::default()).await.unwrap();
188
189        let mut transaction = store
190            .new_transaction(
191                fxfs::object_store::transaction::LockKeys::default(),
192                Options::default(),
193            )
194            .await
195            .unwrap();
196        let _file = ObjectStore::create_object(
197            &store,
198            &mut transaction,
199            fxfs::object_store::HandleOptions::default(),
200            None,
201        )
202        .await
203        .unwrap();
204        transaction.commit().await.unwrap();
205        store.flush().await.expect("flush failed");
206
207        let remote_vol = FxRemoteVolume::new(store.clone(), None, None);
208        let layer_set = remote_vol.get_layer_set().await.expect("get_layer_set failed");
209        assert!(!layer_set.layers.is_empty());
210
211        fs.close().await.expect("close failed");
212    }
213
214    #[fuchsia::test]
215    async fn test_remote_volume_close_fidl() {
216        use std::sync::atomic::{AtomicBool, Ordering};
217
218        let device = DeviceHolder::new(FakeDevice::new(16384, 512));
219        let fs = new_test_fs_builder().format(true).open(device).await.unwrap();
220        let root_vol = root_volume(fs.clone()).await.unwrap();
221        let store = root_vol.new_volume("test_remote_vol", Default::default()).await.unwrap();
222
223        let unmounted = Arc::new(AtomicBool::new(false));
224        let unmounted_clone = unmounted.clone();
225        let unmount_fn: UnmountFn = Box::new(move || {
226            Box::pin(async move {
227                unmounted_clone.store(true, Ordering::Relaxed);
228                Ok(())
229            })
230        });
231
232        let remote_vol = FxRemoteVolume::new(store.clone(), None, Some(unmount_fn));
233
234        let (proxy, server_end) = create_proxy::<ffxfs::RemoteVolumeMarker>();
235        remote_vol.serve(server_end.into_stream());
236
237        assert!(!unmounted.load(Ordering::Relaxed));
238
239        // Call Close FIDL method.
240        let close_res = proxy.close().await.expect("close FIDL call failed");
241        assert!(close_res.is_ok());
242
243        // Verify unmount callback was invoked.
244        assert!(unmounted.load(Ordering::Relaxed));
245
246        fs.close().await.expect("close failed");
247    }
248
249    #[fuchsia::test(threads = 10)]
250    async fn test_remote_volume_open_file() {
251        const WRAPPING_KEY_ID: [u8; 16] = [1; 16];
252
253        let device = DeviceHolder::new(FakeDevice::new(16384, 512));
254        let fs = new_test_fs_builder().format(true).open(device).await.unwrap();
255        let crypt = Arc::new(new_insecure_crypt());
256        crypt.add_wrapping_key(WRAPPING_KEY_ID, [1; 32].into()).unwrap();
257
258        let root_vol = root_volume(fs.clone()).await.unwrap();
259        let store = root_vol
260            .new_volume(
261                "test_remote_vol",
262                fxfs::object_store::NewChildStoreOptions {
263                    options: fxfs::object_store::StoreOptions {
264                        crypt: Some(crypt.clone()),
265                        ..Default::default()
266                    },
267                    ..Default::default()
268                },
269            )
270            .await
271            .unwrap();
272
273        let (client_end, server_end) =
274            fidl::endpoints::create_endpoints::<ffxfs::RemoteVolumeMarker>();
275        let proxy = Arc::new(ffxfs::RemoteVolumeSynchronousProxy::new(client_end.into_channel()));
276        let fx_volume = Arc::new(
277            FxVolume::new(
278                std::sync::Weak::new(),
279                store.clone(),
280                zx::Event::create().koid().unwrap().raw_koid(),
281                "test_vol_server".to_string(),
282                Arc::new(refaults_vmo::PageFaultCounter::new().unwrap()),
283                Default::default(),
284            )
285            .unwrap(),
286        );
287
288        let remote_vol = FxRemoteVolume::new(store.clone(), Some(fx_volume.clone()), None);
289        remote_vol.serve(server_end.into_stream());
290
291        let mut transaction = store
292            .new_transaction(
293                fxfs::object_store::transaction::LockKeys::default(),
294                Options::default(),
295            )
296            .await
297            .unwrap();
298        let fscrypt_file = ObjectStore::create_object(
299            &store,
300            &mut transaction,
301            fxfs::object_store::HandleOptions::default(),
302            Some(WRAPPING_KEY_ID.into()),
303        )
304        .await
305        .unwrap();
306        let fscrypt_file_id = fscrypt_file.object_id();
307        transaction.commit().await.unwrap();
308
309        let payload = b"encrypted file read test";
310        let mut buf = fscrypt_file.allocate_buffer(payload.len()).await;
311        buf.as_mut_ptr_slice().copy_from_slice(payload);
312        fscrypt_file.write_or_append(Some(0), buf.as_ref()).await.unwrap();
313
314        let (file_proxy, file_server_end) = create_proxy::<fio::FileMarker>();
315        let file_info = proxy
316            .open_file(fscrypt_file_id, file_server_end, zx::MonotonicInstant::INFINITE)
317            .unwrap()
318            .expect("open_file failed");
319        let stream = file_info.stream.expect("Stream expected in FileInfo");
320        let stream_clone = stream.duplicate_handle(zx::Rights::SAME_RIGHTS).unwrap();
321        let read_bytes = fuchsia_async::unblock(move || {
322            stream_clone
323                .read_at_to_vec(zx::StreamReadOptions::empty(), 0, payload.len())
324                .expect("stream read failed")
325        })
326        .await;
327        assert_eq!(&read_bytes, payload);
328
329        drop(stream);
330        drop(file_proxy);
331        drop(proxy);
332
333        remote_vol.wait().await;
334        fs.close().await.expect("close failed");
335    }
336}
fxfs_platform_testing/fuchsia/remote_volume.rs

fxfs_platform_testing/fuchsia/
remote_volume.rs
