Hare/wip-rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: unify object observation API

Browse Source
This commit is contained in:
Hare
2026-09-30 18:19:12 +09:00
parent 2e4196a1ec
commit ea7b0f00ea
32 changed files with 1035 additions and 1251 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/wip-fs-server/src/lib.rs
+9 -9
View File
@@ -51,10 +51,10 @@ pub struct ResourceLimits {
pub max_file_read_bytes: usize,
/// Maximum bytes accepted by one file replacement.
pub max_file_write_bytes: usize,
/// Maximum accepted `fetch_tree` depth.
pub max_tree_depth: u32,
/// Maximum accepted `fetch_tree` node count, including its root.
pub max_tree_nodes: u32,
/// Maximum accepted `observe` depth.
pub max_observation_depth: u32,
/// Maximum accepted `observe` node count, including its root.
pub max_observation_nodes: u32,
/// Maximum simultaneously pinned projected filesystem handles.
pub max_open_handles: usize,
/// Maximum bytes accepted for one HTTP request head, including its terminator.
@@ -72,8 +72,8 @@ impl Default for ResourceLimits {
.expect("default wire limits are nonzero"),
max_file_read_bytes: 1024 * 1024,
max_file_write_bytes: 1024 * 1024,
max_tree_depth: 32,
max_tree_nodes: 1024,
max_observation_depth: 32,
max_observation_nodes: 1024,
max_open_handles: 2048,
max_http_header_bytes: 16 * 1024,
max_concurrent_requests: 32,
@@ -125,7 +125,7 @@ impl ServerConfig {
}
if self.limits.max_file_read_bytes == 0
|| self.limits.max_file_write_bytes == 0
|| self.limits.max_tree_nodes == 0
|| self.limits.max_observation_nodes == 0
|| self.limits.max_open_handles == 0
|| self.limits.max_http_header_bytes == 0
|| self.limits.max_concurrent_requests == 0
@@ -133,7 +133,7 @@ impl ServerConfig {
{
return Err(ConfigError::ZeroLimit);
}
if self.limits.max_open_handles < self.limits.max_tree_nodes as usize {
if self.limits.max_open_handles < self.limits.max_observation_nodes as usize {
return Err(ConfigError::InsufficientHandles);
}
if self.limits.wire.max_response_bytes() < 512 {
@@ -171,7 +171,7 @@ impl std::fmt::Display for ConfigError {
"endpoint base path must be absolute, canonical, and query-free"
}
Self::ZeroLimit => "all resource limits and the request timeout must be nonzero",
Self::InsufficientHandles => "max_open_handles must cover max_tree_nodes",
Self::InsufficientHandles => "max_open_handles must cover max_observation_nodes",
Self::ResponseLimitTooSmall => "max_response_bytes must be at least 512",
})
}
crates/wip-fs-server/src/projection.rs
+102 -110
View File
@@ -1,4 +1,3 @@
use std::collections::VecDeque;
use std::ffi::CStr;
use std::mem::MaybeUninit;
use std::os::fd::{AsRawFd, OwnedFd};
@@ -17,8 +16,8 @@ use wip_host::{
};
use wip_interface as wip;
use wip_protocol::{
FetchInterfaceRequest, FetchInterfaceResponse, FetchTreeRequest, FetchTreeResponse, Object,
ObjectTreeNode, ProtocolError, ProtocolErrorCode,
FetchInterfaceRequest, FetchInterfaceResponse, ObjectObservation, ObserveRequest,
ObserveResponse, ProtocolError, ProtocolErrorCode,
};
use crate::interfaces::{
@@ -223,37 +222,60 @@ impl Projection {
.protocol_response()
}
pub(crate) fn fetch_tree(
pub(crate) fn observe(
&self,
context: &ExecutionContext,
request: &FetchTreeRequest,
) -> Result<FetchTreeResponse, ProtocolError> {
request: &ObserveRequest,
) -> Result<ObserveResponse, ProtocolError> {
request.validate().map_err(|_| ProtocolError {
code: ProtocolErrorCode::InvalidRequest,
message: "invalid fetch_tree request".into(),
message: "invalid observe request".into(),
})?;
if request.depth > self.limits.max_tree_depth || request.limit > self.limits.max_tree_nodes
{
if request.depth > self.limits.max_observation_depth {
return Err(ProjectionError::Resource.protocol());
}
let root = self
.resolve(context, &request.path)
.map_err(ProjectionError::protocol)?;
let root_object = self
.publication(root.clone(), context.clone())
.protocol_object();
let mut objects = vec![root_object];
let mut children: Vec<Vec<usize>> = vec![Vec::new()];
let mut queue = VecDeque::from([(0_usize, root, 0_u32)]);
let mut truncated = false;
let mut node_count = 0_u32;
let response = self.observe_entry(context, root, 0, request.depth, &mut node_count)?;
request
.validate_response(&response)
.map_err(|_| ProtocolError {
code: ProtocolErrorCode::Internal,
message: "Host produced an invalid observation response".into(),
})?;
Ok(response)
}
while let Some((parent_index, parent, depth)) = queue.pop_front() {
check_context(context).map_err(ProjectionError::protocol)?;
if parent.kind != EntryKind::Directory {
continue;
}
let directory_fd = reopen(parent.fd(), OFlags::RDONLY | OFlags::DIRECTORY)
fn observe_entry(
&self,
context: &ExecutionContext,
entry: Entry,
depth: u32,
maximum_depth: u32,
node_count: &mut u32,
) -> Result<ObjectObservation, ProtocolError> {
check_context(context).map_err(ProjectionError::protocol)?;
*node_count = node_count.saturating_add(1);
if *node_count > self.limits.max_observation_nodes {
return Err(ProjectionError::Resource.protocol());
}
let object = self
.publication(entry.clone(), context.clone())
.protocol_object();
if depth == maximum_depth {
return Ok(ObjectObservation {
object,
children: None,
});
}
let mut children = Vec::new();
if entry.kind == EntryKind::Directory {
let directory_fd = reopen(entry.fd(), OFlags::RDONLY | OFlags::DIRECTORY)
.map_err(ProjectionError::protocol)?;
let mut buffer = [MaybeUninit::<u8>::uninit(); 8192];
let mut directory = RawDir::new(directory_fd, &mut buffer);
@@ -271,50 +293,25 @@ impl Projection {
if name.contains('/') || name.is_empty() || name == "." || name == ".." {
continue;
}
let child = match self.open_child(context, parent.fd(), name) {
let child = match self.open_child(context, entry.fd(), name) {
Ok(child) => child,
Err(ProjectionError::NotFound) => continue,
Err(error) => return Err(error.protocol()),
};
if depth == request.depth || objects.len() == request.limit as usize {
truncated = true;
break;
}
let child_index = objects.len();
objects.push(
self.publication(child.clone(), context.clone())
.protocol_object(),
);
children.push(Vec::new());
children[parent_index].push(child_index);
if child.kind == EntryKind::Directory {
queue.push_back((child_index, child, depth + 1));
}
}
if truncated && objects.len() == request.limit as usize {
break;
children.push(self.observe_entry(
context,
child,
depth + 1,
maximum_depth,
node_count,
)?);
}
}
fn build(index: usize, objects: &[Object], children: &[Vec<usize>]) -> ObjectTreeNode {
ObjectTreeNode {
object: objects[index].clone(),
children: children[index]
.iter()
.map(|child| build(*child, objects, children))
.collect(),
}
}
let response = FetchTreeResponse {
root: build(0, &objects, &children),
truncated,
};
response.validate_for(request).map_err(|_| ProtocolError {
code: ProtocolErrorCode::Internal,
message: "Host produced an invalid tree response".into(),
})?;
Ok(response)
Ok(ObjectObservation {
object,
children: Some(children),
})
}
fn publication(
@@ -809,7 +806,7 @@ mod tests {
use wip_host::{
CallErrorKind, CallRequest, ExecutionContext, InterfaceTarget, ObjectProvider, Target,
};
use wip_protocol::{FetchTreeRequest, ProtocolErrorCode, Value};
use wip_protocol::{ObserveRequest, ProtocolErrorCode, Value};
use super::{FailureInjection, FlockOperation, Projection};
use crate::{AccessMode, ResourceLimits};
@@ -850,23 +847,29 @@ mod tests {
.expect("create symlink");
let projection = projection(temporary.path(), AccessMode::ReadOnly);
let root = projection.fetch("/").expect("fetch root");
let root = projection.observe_object("/").expect("observe root");
assert_eq!(root.name, "");
assert_eq!(root.description.as_deref(), Some("Filesystem directory"));
assert!(root.interfaces.is_empty());
assert!(root.validator.is_none());
let file = projection.fetch("/file.txt").expect("fetch file");
let file = projection
.observe_object("/file.txt")
.expect("observe file");
assert_eq!(file.interfaces.len(), 1);
assert!(file.validator.is_some());
assert!(file.r#ref.is_none());
let directory = projection.fetch("/dir").expect("fetch directory");
let directory = projection
.observe_object("/dir")
.expect("observe directory");
assert!(directory.interfaces.is_empty());
#[cfg(unix)]
{
let link = projection.fetch("/link").expect("fetch link itself");
let link = projection
.observe_object("/link")
.expect("observe link itself");
assert_eq!(
link.description.as_deref(),
Some("Symbolic link (not followed)")
@@ -874,7 +877,7 @@ mod tests {
assert_eq!(link.interfaces.len(), 1);
assert_eq!(
projection
.fetch("/link/shadow")
.observe_object("/link/shadow")
.expect_err("must not follow")
.kind(),
CallErrorKind::NotFound
@@ -892,18 +895,21 @@ mod tests {
}
assert_eq!(
projection.fetch("/missing").expect_err("missing").kind(),
projection
.observe_object("/missing")
.expect_err("missing")
.kind(),
CallErrorKind::NotFound
);
}
#[test]
fn reads_are_bounded_and_tree_truncation_is_exact() {
fn reads_are_bounded_and_observation_is_complete() {
let temporary = tempdir().expect("temporary root");
fs::write(temporary.path().join("a"), b"abcdef").expect("write file");
fs::write(temporary.path().join("b"), b"b").expect("write file");
let projection = projection(temporary.path(), AccessMode::ReadOnly);
let file = projection.fetch("/a").expect("fetch file");
let file = projection.observe_object("/a").expect("observe file");
let result = call(
&projection,
"/a",
@@ -919,30 +925,17 @@ mod tests {
assert_eq!(result.result, Value::Bytes(b"cde".to_vec()));
let complete = projection
.fetch_tree(
.observe(
&ExecutionContext::new(),
&FetchTreeRequest {
&ObserveRequest {
path: "/".into(),
depth: 1,
limit: 3,
},
)
.expect("complete tree");
assert!(!complete.truncated);
assert_eq!(complete.root.children.len(), 2);
let limited = projection
.fetch_tree(
&ExecutionContext::new(),
&FetchTreeRequest {
path: "/".into(),
depth: 1,
limit: 2,
},
)
.expect("limited tree");
assert!(limited.truncated);
assert_eq!(limited.root.children.len(), 1);
.expect("complete observation");
let children = complete.children.as_ref().expect("root children observed");
assert_eq!(children.len(), 2);
assert!(children.iter().all(|child| child.children.is_none()));
}
#[test]
@@ -951,7 +944,7 @@ mod tests {
let path = temporary.path().join("file");
fs::write(&path, b"old").expect("write file");
let projection = projection(temporary.path(), AccessMode::ReadWrite);
let observed = projection.fetch("/file").expect("fetch file");
let observed = projection.observe_object("/file").expect("observe file");
assert_eq!(observed.interfaces.len(), 2);
let write_interface = observed.interfaces[1].clone();
@@ -981,7 +974,9 @@ mod tests {
assert_eq!(stale.kind(), CallErrorKind::ValidatorMismatch);
assert_eq!(fs::read(&path).expect("read replacement"), b"replacement");
let current = projection.fetch("/file").expect("fetch replacement");
let current = projection
.observe_object("/file")
.expect("observe replacement");
let result = call(
&projection,
"/file",
@@ -1001,7 +996,7 @@ mod tests {
let path = temporary.path().join("file");
fs::write(&path, b"old").expect("write file");
let projection = projection(temporary.path(), AccessMode::ReadWrite);
let observed = projection.fetch("/file").expect("fetch file");
let observed = projection.observe_object("/file").expect("observe file");
let request = || {
call(
&projection,
@@ -1029,25 +1024,23 @@ mod tests {
}
#[test]
fn large_directory_stops_after_proving_truncation() {
fn large_directory_fails_without_partial_observation() {
let temporary = tempdir().expect("temporary root");
for index in 0..2_000 {
fs::write(temporary.path().join(format!("entry-{index}")), b"x")
.expect("write directory entry");
}
let projection = projection(temporary.path(), AccessMode::ReadOnly);
let tree = projection
.fetch_tree(
let error = projection
.observe(
&ExecutionContext::new(),
&FetchTreeRequest {
&ObserveRequest {
path: "/".into(),
depth: 1,
limit: 2,
},
)
.expect("bounded tree");
assert!(tree.truncated);
assert_eq!(tree.root.children.len(), 1);
.expect_err("complete observation exceeds node budget");
assert_eq!(error.code, ProtocolErrorCode::ResourceLimitExceeded);
}
#[cfg(unix)]
@@ -1065,22 +1058,22 @@ mod tests {
let projection = projection(temporary.path(), AccessMode::ReadOnly);
let tree = projection
.fetch_tree(
.observe(
&ExecutionContext::new(),
&FetchTreeRequest {
&ObserveRequest {
path: "/".into(),
depth: 1,
limit: 10,
},
)
.expect("list root");
assert_eq!(tree.root.children.len(), 1);
assert_eq!(tree.root.children[0].object.name, "socket");
let children = tree.children.as_ref().expect("root children observed");
assert_eq!(children.len(), 1);
assert_eq!(children[0].object.name, "socket");
assert_eq!(
tree.root.children[0].object.description.as_deref(),
children[0].object.description.as_deref(),
Some("Unsupported filesystem entry")
);
assert!(tree.root.children[0].object.interfaces.is_empty());
assert!(children[0].object.interfaces.is_empty());
}
#[test]
@@ -1091,12 +1084,11 @@ mod tests {
let cancellation = wip_host::CancellationToken::new();
cancellation.cancel();
let error = projection
.fetch_tree(
.observe(
&ExecutionContext::new().with_cancellation(cancellation),
&FetchTreeRequest {
&ObserveRequest {
path: "/".into(),
depth: 1,
limit: 2,
},
)
.expect_err("cancelled traversal");
@@ -1108,7 +1100,7 @@ mod tests {
let temporary = tempdir().expect("temporary root");
let projection = projection(temporary.path(), AccessMode::ReadOnly);
let error = projection
.fetch("/bad\0name")
.observe_object("/bad\0name")
.expect_err("NUL cannot name a backend entry");
assert_eq!(error.kind(), CallErrorKind::NotFound);
}
@@ -1144,7 +1136,7 @@ mod tests {
};
let projection = Projection::open(temporary.path(), AccessMode::ReadOnly, limits)
.expect("open projection");
let observed = projection.fetch("/file").expect("fetch file");
let observed = projection.observe_object("/file").expect("observe file");
let started = std::time::Instant::now();
let error = call(
&projection,
crates/wip-fs-server/src/server.rs
+19 -45
View File
@@ -10,12 +10,10 @@ use wip_host::{ExecutionContext, ObjectProvider};
use wip_https::{
BatchRequestKind, BatchResponse, BatchResponseItem, BatchSuccess, CodecError, EncodeHttpError,
Endpoint, RequestMetadataError, ResourceLimitKind, Route, decode_batch_request,
decode_call_operation_metadata, decode_fetch_interface_request, decode_fetch_request,
decode_fetch_tree_request, encode_batch_error_response,
encode_batch_resource_limit_error_response, encode_batch_response,
encode_call_operation_response, encode_fetch_interface_response, encode_fetch_response,
encode_fetch_tree_response, encode_protocol_error_response,
encode_resource_limit_error_response, validate_http_request,
decode_call_operation_metadata, decode_fetch_interface_request, decode_observe_request,
encode_batch_error_response, encode_batch_resource_limit_error_response, encode_batch_response,
encode_call_operation_response, encode_fetch_interface_response, encode_observe_response,
encode_protocol_error_response, encode_resource_limit_error_response, validate_http_request,
};
use wip_protocol::{ProtocolError, ProtocolErrorCode, ProtocolInteraction};
@@ -78,10 +76,9 @@ impl WipFsServer {
}
let context = ExecutionContext::new()
.with_deadline(deadline)
.with_resource_budget(u64::from(self.config.limits.max_tree_nodes));
.with_resource_budget(u64::from(self.config.limits.max_observation_nodes));
match route {
Route::Fetch => self.fetch(request.body(), &context),
Route::FetchTree => self.fetch_tree(request.body(), &context),
Route::Observe => self.observe(request.body(), &context),
Route::FetchInterface => self.fetch_interface(request.body()),
Route::CallOperation => self.call_operation(request.body(), &context),
Route::Batch => self.batch(request.body(), &context),
@@ -110,8 +107,7 @@ impl WipFsServer {
fn route(&self, path: &str) -> Option<Route> {
[
Route::Fetch,
Route::FetchTree,
Route::Observe,
Route::FetchInterface,
Route::CallOperation,
Route::Batch,
@@ -120,33 +116,17 @@ impl WipFsServer {
.find(|route| self.endpoint.route(*route).path() == path)
}
fn fetch(&self, body: &[u8], context: &ExecutionContext) -> Response<Vec<u8>> {
let request = match decode_fetch_request(body, self.config.limits.wire) {
fn observe(&self, body: &[u8], context: &ExecutionContext) -> Response<Vec<u8>> {
let request = match decode_observe_request(body, self.config.limits.wire) {
Ok(request) => request,
Err(error) => return self.codec_error(ProtocolInteraction::Fetch, error),
Err(error) => return self.codec_error(ProtocolInteraction::Observe, error),
};
match self
.projection
.fetch_with_context(context, &request.path)
.map_err(|error| error.protocol_error())
{
Ok(object) => encode_fetch_response(&request, &object, self.config.limits.wire)
.unwrap_or_else(|error| self.read_encode_error(ProtocolInteraction::Fetch, error)),
Err(error) => self.protocol_error(ProtocolInteraction::Fetch, &error, None),
}
}
fn fetch_tree(&self, body: &[u8], context: &ExecutionContext) -> Response<Vec<u8>> {
let request = match decode_fetch_tree_request(body, self.config.limits.wire) {
Ok(request) => request,
Err(error) => return self.codec_error(ProtocolInteraction::FetchTree, error),
};
match self.projection.fetch_tree(context, &request) {
Ok(tree) => encode_fetch_tree_response(&request, &tree, self.config.limits.wire)
match self.projection.observe(context, &request) {
Ok(tree) => encode_observe_response(&request, &tree, self.config.limits.wire)
.unwrap_or_else(|error| {
self.read_encode_error(ProtocolInteraction::FetchTree, error)
self.read_encode_error(ProtocolInteraction::Observe, error)
}),
Err(error) => self.protocol_error(ProtocolInteraction::FetchTree, &error, None),
Err(error) => self.protocol_error(ProtocolInteraction::Observe, &error, None),
}
}
@@ -236,15 +216,10 @@ impl WipFsServer {
.iter()
.map(|item| {
let outcome = match &item.request {
BatchRequestKind::Fetch(request) => self
BatchRequestKind::Observe(request) => self
.projection
.fetch_with_context(context, &request.path)
.map(BatchSuccess::Fetch)
.map_err(|error| error.protocol_error()),
BatchRequestKind::FetchTree(request) => self
.projection
.fetch_tree(context, request)
.map(BatchSuccess::FetchTree),
.observe(context, request)
.map(BatchSuccess::Observe),
BatchRequestKind::FetchInterface(request) => self
.projection
.fetch_interface(request)
@@ -674,11 +649,10 @@ fn write_all_with_deadline(
fn interaction(route: Route) -> ProtocolInteraction {
match route {
Route::Fetch => ProtocolInteraction::Fetch,
Route::FetchTree => ProtocolInteraction::FetchTree,
Route::Observe => ProtocolInteraction::Observe,
Route::FetchInterface => ProtocolInteraction::FetchInterface,
Route::CallOperation => ProtocolInteraction::CallOperation,
Route::Batch => ProtocolInteraction::Fetch,
Route::Batch => ProtocolInteraction::Observe,
}
}