1use async_trait::async_trait;
2use serde::{Deserialize, Serialize};
3use std::fmt;
4use std::time::Duration;
5use vfs::engine::{
6 BlockKey, ChunkEdit, ChunkRange, ChunkRef, CreateInodeAttrs, DentryStat, InodeMeta, InodePatch,
7 MetadataStore, SnapshotId, VfsError, VfsResult,
8};
9
10pub const VFS_METADATA_EXT_NAMESPACE: &str = "secure-exec.vfs.metadata.v1";
11const CALLBACK_METADATA_TIMEOUT: Duration = Duration::from_secs(30);
12
13pub trait CallbackMetadataClient: Clone + Send + Sync + 'static {
14 type Ownership: Clone + Send + Sync + 'static;
15 type Error: fmt::Display;
16
17 fn invoke_metadata_callback(
18 &self,
19 ownership: Self::Ownership,
20 namespace: &str,
21 payload: Vec<u8>,
22 timeout: Duration,
23 ) -> Result<(String, Vec<u8>), Self::Error>;
24}
25
26#[derive(Clone)]
27pub struct CallbackMetadataStore<C>
28where
29 C: CallbackMetadataClient,
30{
31 requests: C,
32 ownership: C::Ownership,
33 mount_id: String,
34}
35
36impl<C> CallbackMetadataStore<C>
37where
38 C: CallbackMetadataClient,
39{
40 pub fn new(requests: C, ownership: C::Ownership, mount_id: String) -> Self {
41 Self {
42 requests,
43 ownership,
44 mount_id,
45 }
46 }
47
48 fn invoke(&self, method: MetadataCallbackMethod) -> VfsResult<MetadataCallbackResponse> {
49 let request = MetadataCallbackRequest {
50 mount_id: self.mount_id.clone(),
51 method,
52 };
53 let payload = serde_json::to_vec(&request).map_err(|error| {
54 VfsError::eio(format!(
55 "encode vfs metadata callback request for mount '{}': {error}",
56 self.mount_id
57 ))
58 })?;
59 let (namespace, payload) = self
60 .requests
61 .invoke_metadata_callback(
62 self.ownership.clone(),
63 VFS_METADATA_EXT_NAMESPACE,
64 payload,
65 CALLBACK_METADATA_TIMEOUT,
66 )
67 .map_err(Self::sidecar_error_to_vfs)?;
68 if namespace != VFS_METADATA_EXT_NAMESPACE {
69 return Err(VfsError::eio(format!(
70 "unexpected vfs metadata callback namespace '{namespace}'"
71 )));
72 }
73 let response: MetadataCallbackResponse =
74 serde_json::from_slice(&payload).map_err(|error| {
75 VfsError::eio(format!(
76 "decode vfs metadata callback response for mount '{}': {error}",
77 self.mount_id
78 ))
79 })?;
80 if let MetadataCallbackResponse::Err { code, message } = &response {
81 return Err(VfsError::new(code_from_string(code), message.clone()));
82 }
83 Ok(response)
84 }
85
86 fn sidecar_error_to_vfs(error: C::Error) -> VfsError {
87 VfsError::eio(error.to_string())
88 }
89}
90
91#[derive(Debug, Clone, Serialize, Deserialize)]
92#[serde(rename_all = "camelCase")]
93pub struct MetadataCallbackRequest {
94 pub mount_id: String,
95 pub method: MetadataCallbackMethod,
96}
97
98#[derive(Debug, Clone, Serialize, Deserialize)]
99#[serde(tag = "type", rename_all = "camelCase")]
100pub enum MetadataCallbackMethod {
101 Resolve {
102 path: String,
103 },
104 ResolveParent {
105 path: String,
106 },
107 Lstat {
108 path: String,
109 },
110 ListDir {
111 ino: u64,
112 },
113 Create {
114 parent: u64,
115 name: String,
116 attrs: CreateInodeAttrs,
117 },
118 Link {
119 parent: u64,
120 name: String,
121 target: u64,
122 },
123 Remove {
124 parent: u64,
125 name: String,
126 },
127 Rename {
128 src_parent: u64,
129 src: String,
130 dst_parent: u64,
131 dst: String,
132 },
133 SetAttr {
134 ino: u64,
135 patch: InodePatch,
136 },
137 CommitWrite {
138 ino: u64,
139 edits: Vec<ChunkEdit>,
140 new_size: u64,
141 },
142 GetChunks {
143 ino: u64,
144 range: ChunkRange,
145 },
146 Snapshot {
147 root: u64,
148 },
149 Fork {
150 snap: SnapshotId,
151 },
152 Gc,
153}
154
155#[derive(Debug, Clone, Serialize, Deserialize)]
156#[serde(tag = "type", rename_all = "camelCase")]
157pub enum MetadataCallbackResponse {
158 InodeMeta { meta: InodeMeta },
159 ResolveParent { parent: InodeMeta, name: String },
160 DentryStats { entries: Vec<DentryStat> },
161 Unit,
162 BlockKeys { keys: Vec<BlockKey> },
163 ChunkRefs { chunks: Vec<ChunkRef> },
164 Snapshot { id: SnapshotId },
165 Inode { ino: u64 },
166 Err { code: String, message: String },
167}
168
169#[async_trait]
170impl<C> MetadataStore for CallbackMetadataStore<C>
171where
172 C: CallbackMetadataClient,
173{
174 async fn resolve(&self, path: &str) -> VfsResult<InodeMeta> {
175 match self.invoke(MetadataCallbackMethod::Resolve {
176 path: path.to_owned(),
177 })? {
178 MetadataCallbackResponse::InodeMeta { meta } => Ok(meta),
179 other => Err(unexpected_response("resolve", other)),
180 }
181 }
182
183 async fn resolve_parent(&self, path: &str) -> VfsResult<(InodeMeta, String)> {
184 match self.invoke(MetadataCallbackMethod::ResolveParent {
185 path: path.to_owned(),
186 })? {
187 MetadataCallbackResponse::ResolveParent { parent, name } => Ok((parent, name)),
188 other => Err(unexpected_response("resolveParent", other)),
189 }
190 }
191
192 async fn lstat(&self, path: &str) -> VfsResult<InodeMeta> {
193 match self.invoke(MetadataCallbackMethod::Lstat {
194 path: path.to_owned(),
195 })? {
196 MetadataCallbackResponse::InodeMeta { meta } => Ok(meta),
197 other => Err(unexpected_response("lstat", other)),
198 }
199 }
200
201 async fn list_dir(&self, ino: u64) -> VfsResult<Vec<DentryStat>> {
202 match self.invoke(MetadataCallbackMethod::ListDir { ino })? {
203 MetadataCallbackResponse::DentryStats { entries } => Ok(entries),
204 other => Err(unexpected_response("listDir", other)),
205 }
206 }
207
208 async fn create(
209 &self,
210 parent: u64,
211 name: &str,
212 attrs: CreateInodeAttrs,
213 ) -> VfsResult<InodeMeta> {
214 match self.invoke(MetadataCallbackMethod::Create {
215 parent,
216 name: name.to_owned(),
217 attrs,
218 })? {
219 MetadataCallbackResponse::InodeMeta { meta } => Ok(meta),
220 other => Err(unexpected_response("create", other)),
221 }
222 }
223
224 async fn link(&self, parent: u64, name: &str, target: u64) -> VfsResult<()> {
225 match self.invoke(MetadataCallbackMethod::Link {
226 parent,
227 name: name.to_owned(),
228 target,
229 })? {
230 MetadataCallbackResponse::Unit => Ok(()),
231 other => Err(unexpected_response("link", other)),
232 }
233 }
234
235 async fn remove(&self, parent: u64, name: &str) -> VfsResult<Vec<BlockKey>> {
236 match self.invoke(MetadataCallbackMethod::Remove {
237 parent,
238 name: name.to_owned(),
239 })? {
240 MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
241 other => Err(unexpected_response("remove", other)),
242 }
243 }
244
245 async fn rename(
246 &self,
247 src_parent: u64,
248 src: &str,
249 dst_parent: u64,
250 dst: &str,
251 ) -> VfsResult<Vec<BlockKey>> {
252 match self.invoke(MetadataCallbackMethod::Rename {
253 src_parent,
254 src: src.to_owned(),
255 dst_parent,
256 dst: dst.to_owned(),
257 })? {
258 MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
259 other => Err(unexpected_response("rename", other)),
260 }
261 }
262
263 async fn set_attr(&self, ino: u64, patch: InodePatch) -> VfsResult<Vec<BlockKey>> {
264 match self.invoke(MetadataCallbackMethod::SetAttr { ino, patch })? {
265 MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
266 other => Err(unexpected_response("setAttr", other)),
267 }
268 }
269
270 async fn commit_write(
271 &self,
272 ino: u64,
273 edits: Vec<ChunkEdit>,
274 new_size: u64,
275 ) -> VfsResult<Vec<BlockKey>> {
276 match self.invoke(MetadataCallbackMethod::CommitWrite {
277 ino,
278 edits,
279 new_size,
280 })? {
281 MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
282 other => Err(unexpected_response("commitWrite", other)),
283 }
284 }
285
286 async fn get_chunks(&self, ino: u64, range: ChunkRange) -> VfsResult<Vec<ChunkRef>> {
287 match self.invoke(MetadataCallbackMethod::GetChunks { ino, range })? {
288 MetadataCallbackResponse::ChunkRefs { chunks } => Ok(chunks),
289 other => Err(unexpected_response("getChunks", other)),
290 }
291 }
292
293 async fn snapshot(&self, root: u64) -> VfsResult<SnapshotId> {
294 match self.invoke(MetadataCallbackMethod::Snapshot { root })? {
295 MetadataCallbackResponse::Snapshot { id } => Ok(id),
296 other => Err(unexpected_response("snapshot", other)),
297 }
298 }
299
300 async fn fork(&self, snap: SnapshotId) -> VfsResult<u64> {
301 match self.invoke(MetadataCallbackMethod::Fork { snap })? {
302 MetadataCallbackResponse::Inode { ino } => Ok(ino),
303 other => Err(unexpected_response("fork", other)),
304 }
305 }
306
307 async fn gc(&self) -> VfsResult<Vec<BlockKey>> {
308 match self.invoke(MetadataCallbackMethod::Gc)? {
309 MetadataCallbackResponse::BlockKeys { keys } => Ok(keys),
310 other => Err(unexpected_response("gc", other)),
311 }
312 }
313}
314
315fn unexpected_response(method: &str, response: MetadataCallbackResponse) -> VfsError {
316 VfsError::eio(format!(
317 "unexpected vfs metadata callback response for {method}: {response:?}"
318 ))
319}
320
321fn code_from_string(code: &str) -> &'static str {
322 match code {
323 "ENOENT" => "ENOENT",
324 "EEXIST" => "EEXIST",
325 "ENOTDIR" => "ENOTDIR",
326 "EISDIR" => "EISDIR",
327 "ELOOP" => "ELOOP",
328 "ENAMETOOLONG" => "ENAMETOOLONG",
329 "ENOTEMPTY" => "ENOTEMPTY",
330 "EOPNOTSUPP" => "EOPNOTSUPP",
331 "EROFS" => "EROFS",
332 "EINVAL" => "EINVAL",
333 "EACCES" => "EACCES",
334 "EPERM" => "EPERM",
335 "ENOSYS" => "ENOSYS",
336 _ => "EIO",
337 }
338}
339
340#[cfg(test)]
341mod tests {
342 use super::*;
343 use std::sync::{Arc, Mutex};
344 use vfs::engine::engines::{ChunkedFs, ChunkedFsOptions};
345 use vfs::engine::mem::{InMemoryMetadataStore, MemoryBlockStore};
346 use vfs::engine::{InodeType, MetadataStore, Storage, Timespec, VirtualFileSystem};
347
348 #[derive(Default)]
349 struct RecordingMetadataTransport {
350 requests: Mutex<Vec<MetadataCallbackRequest>>,
351 }
352
353 impl CallbackMetadataClient for Arc<RecordingMetadataTransport> {
354 type Ownership = String;
355 type Error = String;
356
357 fn invoke_metadata_callback(
358 &self,
359 _ownership: Self::Ownership,
360 namespace: &str,
361 payload: Vec<u8>,
362 _timeout: Duration,
363 ) -> Result<(String, Vec<u8>), Self::Error> {
364 assert_eq!(namespace, VFS_METADATA_EXT_NAMESPACE);
365 let callback_request: MetadataCallbackRequest =
366 serde_json::from_slice(&payload).expect("decode metadata callback");
367 self.requests
368 .lock()
369 .expect("lock request log")
370 .push(callback_request);
371
372 let now = Timespec { sec: 1, nsec: 2 };
373 let response = MetadataCallbackResponse::InodeMeta {
374 meta: InodeMeta {
375 ino: 7,
376 kind: InodeType::File,
377 mode: 0o644,
378 uid: 1000,
379 gid: 1000,
380 size: 5,
381 nlink: 1,
382 atime: now,
383 mtime: now,
384 ctime: now,
385 birthtime: now,
386 storage: Storage::Inline(b"hello".to_vec()),
387 symlink_target: None,
388 },
389 };
390 let payload = serde_json::to_vec(&response).expect("encode metadata callback response");
391 Ok((VFS_METADATA_EXT_NAMESPACE.to_string(), payload))
392 }
393 }
394
395 #[tokio::test]
396 async fn callback_metadata_store_sends_typed_ext_requests() {
397 let transport = Arc::new(RecordingMetadataTransport::default());
398 let store = CallbackMetadataStore::new(
399 transport.clone(),
400 "vm-owner".to_string(),
401 "mount-a".to_string(),
402 );
403
404 let meta = store
405 .resolve("/file.txt")
406 .await
407 .expect("resolve via callback");
408 assert_eq!(meta.ino, 7);
409 assert_eq!(meta.storage, Storage::Inline(b"hello".to_vec()));
410
411 let requests = transport.requests.lock().expect("lock request log");
412 assert_eq!(requests.len(), 1);
413 assert_eq!(requests[0].mount_id, "mount-a");
414 match &requests[0].method {
415 MetadataCallbackMethod::Resolve { path } => assert_eq!(path, "/file.txt"),
416 other => panic!("unexpected method: {other:?}"),
417 }
418 }
419
420 #[derive(Default)]
421 struct DelegatingMetadataTransport {
422 inner: InMemoryMetadataStore,
423 methods: Mutex<Vec<&'static str>>,
424 }
425
426 impl CallbackMetadataClient for Arc<DelegatingMetadataTransport> {
427 type Ownership = String;
428 type Error = String;
429
430 fn invoke_metadata_callback(
431 &self,
432 _ownership: Self::Ownership,
433 namespace: &str,
434 payload: Vec<u8>,
435 _timeout: Duration,
436 ) -> Result<(String, Vec<u8>), Self::Error> {
437 assert_eq!(namespace, VFS_METADATA_EXT_NAMESPACE);
438 let request: MetadataCallbackRequest =
439 serde_json::from_slice(&payload).map_err(|err| err.to_string())?;
440 let inner = self.inner.clone();
441 let response = std::thread::spawn(move || {
442 let runtime = tokio::runtime::Builder::new_current_thread()
443 .enable_all()
444 .build()
445 .map_err(|err| err.to_string())?;
446 runtime.block_on(handle_metadata_callback(inner, request.method))
447 })
448 .join()
449 .map_err(|_| "metadata callback thread panicked".to_string())??;
450 self.methods
451 .lock()
452 .expect("lock method log")
453 .push(method_name(&response));
454 let payload = serde_json::to_vec(&response).map_err(|err| err.to_string())?;
455 Ok((VFS_METADATA_EXT_NAMESPACE.to_string(), payload))
456 }
457 }
458
459 async fn handle_metadata_callback(
460 inner: InMemoryMetadataStore,
461 method: MetadataCallbackMethod,
462 ) -> Result<MetadataCallbackResponse, String> {
463 let result = match method {
464 MetadataCallbackMethod::Resolve { path } => inner
465 .resolve(&path)
466 .await
467 .map(|meta| MetadataCallbackResponse::InodeMeta { meta }),
468 MetadataCallbackMethod::ResolveParent { path } => inner
469 .resolve_parent(&path)
470 .await
471 .map(|(parent, name)| MetadataCallbackResponse::ResolveParent { parent, name }),
472 MetadataCallbackMethod::Lstat { path } => inner
473 .lstat(&path)
474 .await
475 .map(|meta| MetadataCallbackResponse::InodeMeta { meta }),
476 MetadataCallbackMethod::ListDir { ino } => inner
477 .list_dir(ino)
478 .await
479 .map(|entries| MetadataCallbackResponse::DentryStats { entries }),
480 MetadataCallbackMethod::Create {
481 parent,
482 name,
483 attrs,
484 } => inner
485 .create(parent, &name, attrs)
486 .await
487 .map(|meta| MetadataCallbackResponse::InodeMeta { meta }),
488 MetadataCallbackMethod::Link {
489 parent,
490 name,
491 target,
492 } => inner
493 .link(parent, &name, target)
494 .await
495 .map(|()| MetadataCallbackResponse::Unit),
496 MetadataCallbackMethod::Remove { parent, name } => inner
497 .remove(parent, &name)
498 .await
499 .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
500 MetadataCallbackMethod::Rename {
501 src_parent,
502 src,
503 dst_parent,
504 dst,
505 } => inner
506 .rename(src_parent, &src, dst_parent, &dst)
507 .await
508 .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
509 MetadataCallbackMethod::SetAttr { ino, patch } => inner
510 .set_attr(ino, patch)
511 .await
512 .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
513 MetadataCallbackMethod::CommitWrite {
514 ino,
515 edits,
516 new_size,
517 } => inner
518 .commit_write(ino, edits, new_size)
519 .await
520 .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
521 MetadataCallbackMethod::GetChunks { ino, range } => inner
522 .get_chunks(ino, range)
523 .await
524 .map(|chunks| MetadataCallbackResponse::ChunkRefs { chunks }),
525 MetadataCallbackMethod::Snapshot { root } => inner
526 .snapshot(root)
527 .await
528 .map(|id| MetadataCallbackResponse::Snapshot { id }),
529 MetadataCallbackMethod::Fork { snap } => inner
530 .fork(snap)
531 .await
532 .map(|ino| MetadataCallbackResponse::Inode { ino }),
533 MetadataCallbackMethod::Gc => inner
534 .gc()
535 .await
536 .map(|keys| MetadataCallbackResponse::BlockKeys { keys }),
537 };
538 Ok(
539 result.unwrap_or_else(|error| MetadataCallbackResponse::Err {
540 code: error.code().to_string(),
541 message: error.message().to_string(),
542 }),
543 )
544 }
545
546 fn method_name(response: &MetadataCallbackResponse) -> &'static str {
547 match response {
548 MetadataCallbackResponse::InodeMeta { .. } => "inodeMeta",
549 MetadataCallbackResponse::ResolveParent { .. } => "resolveParent",
550 MetadataCallbackResponse::DentryStats { .. } => "dentryStats",
551 MetadataCallbackResponse::Unit => "unit",
552 MetadataCallbackResponse::BlockKeys { .. } => "blockKeys",
553 MetadataCallbackResponse::ChunkRefs { .. } => "chunkRefs",
554 MetadataCallbackResponse::Snapshot { .. } => "snapshot",
555 MetadataCallbackResponse::Inode { .. } => "inode",
556 MetadataCallbackResponse::Err { .. } => "err",
557 }
558 }
559
560 #[tokio::test]
561 async fn callback_metadata_store_drives_chunked_filesystem_round_trip() {
562 let transport = Arc::new(DelegatingMetadataTransport::default());
563 let metadata = CallbackMetadataStore::new(
564 transport.clone(),
565 "vm-owner".to_string(),
566 "mount-a".to_string(),
567 );
568 let fs = ChunkedFs::with_options(
569 metadata,
570 MemoryBlockStore::new(),
571 ChunkedFsOptions {
572 inline_threshold: 1,
573 chunk_size: 4,
574 ..ChunkedFsOptions::default()
575 },
576 );
577
578 fs.write_file("/file.txt", b"abcdefgh").await.unwrap();
579 fs.pwrite("/file.txt", b"YY", 2).await.unwrap();
580 assert_eq!(fs.read_file("/file.txt").await.unwrap(), b"abYYefgh");
581 fs.truncate("/file.txt", 5).await.unwrap();
582 assert_eq!(fs.read_file("/file.txt").await.unwrap(), b"abYYe");
583
584 let methods = transport.methods.lock().expect("lock method log");
585 assert!(methods.contains(&"blockKeys"));
586 assert!(methods.contains(&"chunkRefs"));
587 }
588}