feat(maestro): daemon+registry ListDir RPC + broker relay + HTTP dirs endpoint (NAP-329)
Also fixes two pre-existing test call sites that NAP-328 left passing the old arity to register()/state_for() after adding home_dir.
This commit is contained in:
@@ -19,7 +19,7 @@ use std::sync::{Arc, Mutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use maestro_protocol::{
|
||||
AttachCandidates, CreateSessionRequest, DaemonOutbound, ReplyTo, RpcResult, Session,
|
||||
AttachCandidates, CreateSessionRequest, DaemonOutbound, DirListing, ReplyTo, RpcResult, Session,
|
||||
};
|
||||
use tokio::sync::{mpsc, oneshot};
|
||||
use uuid::Uuid;
|
||||
@@ -71,6 +71,8 @@ pub struct BrokerHandle {
|
||||
pending_create: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<Session>>>>>,
|
||||
/// Pending `ForwardKill` replies, keyed by `reply_to` (NAP-327).
|
||||
pending_kill: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<()>>>>>,
|
||||
/// Pending `ForwardListDir` replies, keyed by `reply_to` (NAP-329).
|
||||
pending_list_dir: Arc<Mutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<DirListing>>>>>,
|
||||
/// Whether the registry control link is currently up. Flipped by the
|
||||
/// registry-client task via [`BrokerHandle::set_connected`]. When down,
|
||||
/// `resolve_attach` fails fast instead of queueing onto an undrained
|
||||
@@ -92,6 +94,7 @@ impl BrokerHandle {
|
||||
pending: Arc::new(Mutex::new(HashMap::new())),
|
||||
pending_create: Arc::new(Mutex::new(HashMap::new())),
|
||||
pending_kill: Arc::new(Mutex::new(HashMap::new())),
|
||||
pending_list_dir: Arc::new(Mutex::new(HashMap::new())),
|
||||
connected: Arc::new(AtomicBool::new(false)),
|
||||
};
|
||||
(handle, out_rx)
|
||||
@@ -232,6 +235,63 @@ impl BrokerHandle {
|
||||
}
|
||||
}
|
||||
|
||||
/// Broker a cross-daemon directory listing: ask the registry to relay the
|
||||
/// `ListDir` to the daemon named by `daemon_id` (which may be a peer) and
|
||||
/// return its listing. Same fail-fast-while-disconnected + RAII-cleanup
|
||||
/// discipline as the other broker calls (NAP-329).
|
||||
pub async fn forward_list_dir(
|
||||
&self,
|
||||
daemon_id: String,
|
||||
path: String,
|
||||
) -> Result<DirListing, BrokerError> {
|
||||
if !self.connected.load(Ordering::Acquire) {
|
||||
return Err(BrokerError::NoControlLink);
|
||||
}
|
||||
let reply_to = Uuid::new_v4();
|
||||
let (tx, rx) = oneshot::channel();
|
||||
self.pending_list_dir.lock().unwrap().insert(reply_to, tx);
|
||||
let _guard = PendingGuard {
|
||||
map: &*self.pending_list_dir,
|
||||
key: reply_to,
|
||||
};
|
||||
if self
|
||||
.out_tx
|
||||
.try_send(DaemonOutbound::ForwardListDir {
|
||||
reply_to,
|
||||
daemon_id,
|
||||
path,
|
||||
})
|
||||
.is_err()
|
||||
{
|
||||
return Err(BrokerError::NoControlLink);
|
||||
}
|
||||
match tokio::time::timeout(RESOLVE_TIMEOUT, rx).await {
|
||||
Ok(Ok(RpcResult::Ok { value })) => Ok(value),
|
||||
Ok(Ok(RpcResult::Err { error })) => {
|
||||
Err(BrokerError::Registry(error.code, error.message))
|
||||
}
|
||||
Ok(Err(_)) => Err(BrokerError::ReplyClosed),
|
||||
Err(_) => Err(BrokerError::Timeout(RESOLVE_TIMEOUT)),
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve a waiting [`forward_list_dir`](Self::forward_list_dir); called by
|
||||
/// the registry-client task on a `ForwardListDirResult` frame.
|
||||
pub async fn complete_forward_list_dir(
|
||||
&self,
|
||||
reply_to: ReplyTo,
|
||||
result: RpcResult<DirListing>,
|
||||
) -> bool {
|
||||
let entry = self.pending_list_dir.lock().unwrap().remove(&reply_to);
|
||||
match entry {
|
||||
Some(tx) => tx.send(result).is_ok(),
|
||||
None => {
|
||||
tracing::debug!(?reply_to, "no waiter for ForwardListDirResult");
|
||||
false
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve a waiting [`forward_create`](Self::forward_create); called by the
|
||||
/// registry-client task on a `ForwardCreateResult` frame.
|
||||
pub async fn complete_forward_create(
|
||||
@@ -457,6 +517,62 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
// ----- NAP-329: forward_list_dir -----
|
||||
|
||||
/// A `forward_list_dir` issues a `ForwardListDir` frame and resolves with
|
||||
/// the listing once the registry's `ForwardListDirResult` is delivered.
|
||||
#[tokio::test]
|
||||
async fn forward_list_dir_round_trips_listing() {
|
||||
let (handle, mut out_rx) = BrokerHandle::new();
|
||||
handle.set_connected(true);
|
||||
|
||||
let h = handle.clone();
|
||||
let task =
|
||||
tokio::spawn(async move { h.forward_list_dir("konishi".into(), "/work".into()).await });
|
||||
|
||||
let reply_to = match out_rx.recv().await.expect("ForwardListDir went out") {
|
||||
DaemonOutbound::ForwardListDir {
|
||||
reply_to,
|
||||
daemon_id,
|
||||
path,
|
||||
} => {
|
||||
assert_eq!(daemon_id, "konishi");
|
||||
assert_eq!(path, "/work");
|
||||
reply_to
|
||||
}
|
||||
other => panic!("expected ForwardListDir, got {other:?}"),
|
||||
};
|
||||
handle
|
||||
.complete_forward_list_dir(
|
||||
reply_to,
|
||||
RpcResult::Ok {
|
||||
value: DirListing {
|
||||
path: std::path::PathBuf::from("/work"),
|
||||
entries: vec!["repo".into()],
|
||||
truncated: false,
|
||||
},
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
let got = task.await.unwrap().expect("list ok");
|
||||
assert_eq!(got.entries, vec!["repo".to_string()]);
|
||||
}
|
||||
|
||||
/// While disconnected, `forward_list_dir` fails fast and leaves no pending
|
||||
/// entry (NAP-263 / NAP-259).
|
||||
#[tokio::test]
|
||||
async fn forward_list_dir_fails_fast_while_disconnected() {
|
||||
let (handle, _out_rx) = BrokerHandle::new(); // starts disconnected
|
||||
assert!(matches!(
|
||||
handle
|
||||
.forward_list_dir("konishi".into(), "/work".into())
|
||||
.await,
|
||||
Err(BrokerError::NoControlLink)
|
||||
));
|
||||
assert_eq!(handle.pending_list_dir.lock().unwrap().len(), 0);
|
||||
}
|
||||
|
||||
/// NAP-270: if the registry never replies, `resolve_attach` resolves to
|
||||
/// `BrokerError::Timeout` after `RESOLVE_TIMEOUT` (driven here with a paused
|
||||
/// clock so the test is instant).
|
||||
|
||||
@@ -0,0 +1,188 @@
|
||||
//! Directory listing for the spawn-form working-directory autocomplete (NAP-329).
|
||||
//!
|
||||
//! A daemon answers `ListDir { path }` by listing the **immediate
|
||||
//! subdirectories** of one directory on its own host. The web client lists one
|
||||
//! directory at a time and filters the trailing path segment locally, so this
|
||||
//! never walks recursively and never returns files.
|
||||
//!
|
||||
//! Path expansion is done against the daemon's own `$HOME` so a client can send
|
||||
//! an abbreviated path (empty, `~`, or relative) and get a sensible result:
|
||||
//!
|
||||
//! - empty → home dir
|
||||
//! - `~` / `~/sub` → home / home + `sub`
|
||||
//! - relative `a/b` → home + `a/b`
|
||||
//! - absolute `/a/b` → as-is
|
||||
//!
|
||||
//! The pure [`list_dir`] is unit-tested in isolation; the registry-client task
|
||||
//! just wraps its `Result` in an `RpcResult` for the wire.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
|
||||
use maestro_protocol::{DirListing, ErrorInfo};
|
||||
|
||||
/// Maximum number of entries returned in a single listing. A directory with
|
||||
/// more subdirectories than this is reported `truncated`; the client narrows
|
||||
/// its query by typing more of the name. Bounds the wire payload regardless of
|
||||
/// how large a directory is.
|
||||
pub const MAX_ENTRIES: usize = 1000;
|
||||
|
||||
/// List the immediate subdirectories of `requested`, expanding it against
|
||||
/// `home_dir` first. Returns directory **names** only (not full paths), sorted
|
||||
/// case-insensitively, with hidden entries included.
|
||||
pub fn list_dir(home_dir: &str, requested: &str) -> Result<DirListing, ErrorInfo> {
|
||||
let dir = resolve(home_dir, requested);
|
||||
|
||||
let read = std::fs::read_dir(&dir).map_err(|e| io_error(&dir, &e))?;
|
||||
|
||||
let mut entries: Vec<String> = Vec::new();
|
||||
for entry in read.flatten() {
|
||||
if is_dir(&entry) {
|
||||
entries.push(entry.file_name().to_string_lossy().into_owned());
|
||||
}
|
||||
}
|
||||
// Case-insensitive sort so `Projects` and `projects` interleave the way a
|
||||
// user scanning the list expects.
|
||||
entries.sort_by_key(|s| s.to_lowercase());
|
||||
|
||||
let truncated = entries.len() > MAX_ENTRIES;
|
||||
entries.truncate(MAX_ENTRIES);
|
||||
|
||||
Ok(DirListing {
|
||||
path: dir,
|
||||
entries,
|
||||
truncated,
|
||||
})
|
||||
}
|
||||
|
||||
/// Expand an abbreviated client path against the daemon's home dir.
|
||||
fn resolve(home_dir: &str, requested: &str) -> PathBuf {
|
||||
let home = PathBuf::from(home_dir);
|
||||
let trimmed = requested.trim();
|
||||
if trimmed.is_empty() {
|
||||
return home;
|
||||
}
|
||||
if trimmed == "~" {
|
||||
return home;
|
||||
}
|
||||
if let Some(rest) = trimmed.strip_prefix("~/") {
|
||||
return home.join(rest);
|
||||
}
|
||||
let p = PathBuf::from(trimmed);
|
||||
if p.is_absolute() {
|
||||
p
|
||||
} else {
|
||||
home.join(p)
|
||||
}
|
||||
}
|
||||
|
||||
/// Whether a `DirEntry` denotes a directory, following symlinks so that a
|
||||
/// symlinked project dir still completes (shells do the same).
|
||||
fn is_dir(entry: &std::fs::DirEntry) -> bool {
|
||||
match entry.file_type() {
|
||||
Ok(ft) if ft.is_dir() => true,
|
||||
Ok(ft) if ft.is_symlink() => std::fs::metadata(entry.path())
|
||||
.map(|m| m.is_dir())
|
||||
.unwrap_or(false),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Map a `read_dir` failure to a stable [`ErrorInfo`] code the client can react
|
||||
/// to. `NotADirectory` isn't a stable `ErrorKind` across platforms, so we probe
|
||||
/// with a `metadata` call to distinguish "file, not dir" from "doesn't exist".
|
||||
fn io_error(dir: &Path, e: &std::io::Error) -> ErrorInfo {
|
||||
use std::io::ErrorKind;
|
||||
let (code, message) = match e.kind() {
|
||||
ErrorKind::NotFound => ("not_found", "no such directory"),
|
||||
ErrorKind::PermissionDenied => ("permission_denied", "permission denied"),
|
||||
_ if dir.is_file() => ("not_a_directory", "path is a file, not a directory"),
|
||||
_ => ("io_error", "could not read directory"),
|
||||
};
|
||||
ErrorInfo {
|
||||
code: code.into(),
|
||||
message: format!("{message}: {}", dir.display()),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::fs;
|
||||
|
||||
/// Build a temp home with a few subdirs (one hidden), a file, and a symlink
|
||||
/// to a dir. Returns the temp root path (its `Drop` cleans up).
|
||||
fn fixture() -> tempfile::TempDir {
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let home = tmp.path();
|
||||
fs::create_dir(home.join("projects")).unwrap();
|
||||
fs::create_dir(home.join("Downloads")).unwrap();
|
||||
fs::create_dir(home.join(".config")).unwrap();
|
||||
fs::write(home.join("notes.txt"), b"x").unwrap();
|
||||
#[cfg(unix)]
|
||||
std::os::unix::fs::symlink(home.join("projects"), home.join("link-to-projects")).unwrap();
|
||||
tmp
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lists_dirs_only_sorted_case_insensitively_including_hidden() {
|
||||
let tmp = fixture();
|
||||
let home = tmp.path().to_string_lossy();
|
||||
let listing = list_dir(&home, "").unwrap();
|
||||
// The file is excluded; dirs (incl. hidden + symlinked dir) are present.
|
||||
assert!(listing.entries.contains(&".config".to_string()));
|
||||
assert!(listing.entries.contains(&"projects".to_string()));
|
||||
assert!(listing.entries.contains(&"Downloads".to_string()));
|
||||
assert!(!listing.entries.contains(&"notes.txt".to_string()));
|
||||
#[cfg(unix)]
|
||||
assert!(listing.entries.contains(&"link-to-projects".to_string()));
|
||||
// Case-insensitive order: .config, Downloads, link-to-projects, projects
|
||||
let lower: Vec<String> = listing.entries.iter().map(|s| s.to_lowercase()).collect();
|
||||
let mut sorted = lower.clone();
|
||||
sorted.sort();
|
||||
assert_eq!(lower, sorted);
|
||||
assert!(!listing.truncated);
|
||||
assert_eq!(listing.path, tmp.path());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_and_tilde_resolve_to_home() {
|
||||
let tmp = fixture();
|
||||
let home = tmp.path().to_string_lossy();
|
||||
assert_eq!(list_dir(&home, "").unwrap().path, tmp.path());
|
||||
assert_eq!(list_dir(&home, "~").unwrap().path, tmp.path());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn relative_and_tilde_slash_join_onto_home() {
|
||||
let tmp = fixture();
|
||||
let home = tmp.path().to_string_lossy();
|
||||
let rel = list_dir(&home, "projects").unwrap();
|
||||
assert_eq!(rel.path, tmp.path().join("projects"));
|
||||
let tilde = list_dir(&home, "~/projects").unwrap();
|
||||
assert_eq!(tilde.path, tmp.path().join("projects"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn absolute_path_is_listed_as_is() {
|
||||
let tmp = fixture();
|
||||
let abs = tmp.path().join("projects");
|
||||
// home is irrelevant for an absolute path.
|
||||
let listing = list_dir("/nonexistent-home", &abs.to_string_lossy()).unwrap();
|
||||
assert_eq!(listing.path, abs);
|
||||
assert!(listing.entries.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn missing_dir_is_not_found() {
|
||||
let tmp = fixture();
|
||||
let err = list_dir(&tmp.path().to_string_lossy(), "does-not-exist").unwrap_err();
|
||||
assert_eq!(err.code, "not_found");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn file_target_is_not_a_directory() {
|
||||
let tmp = fixture();
|
||||
let err = list_dir(&tmp.path().to_string_lossy(), "notes.txt").unwrap_err();
|
||||
assert_eq!(err.code, "not_a_directory");
|
||||
}
|
||||
}
|
||||
@@ -12,6 +12,7 @@ pub mod catalog_mirror;
|
||||
pub mod cli;
|
||||
pub mod config;
|
||||
pub mod configure;
|
||||
pub mod dirlist;
|
||||
pub mod jwt;
|
||||
pub mod local_api;
|
||||
pub mod local_events;
|
||||
|
||||
@@ -25,13 +25,15 @@
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use axum::extract::{Path, State};
|
||||
use axum::extract::{Path, Query, State};
|
||||
use axum::http::StatusCode;
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::routing::{any, delete, get};
|
||||
use axum::{Json, Router};
|
||||
use maestro_protocol::{CreateSessionRequest, DaemonInfo, ErrorInfo, Session, PROTOCOL_VERSION};
|
||||
use serde::Serialize;
|
||||
use maestro_protocol::{
|
||||
CreateSessionRequest, DaemonInfo, DirListing, ErrorInfo, Session, PROTOCOL_VERSION,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::catalog_mirror::CatalogMirror;
|
||||
@@ -73,6 +75,7 @@ pub fn router(state: LocalApiState) -> Router {
|
||||
.route("/api/sessions", get(list_sessions).post(create_session))
|
||||
.route("/api/sessions/:id", delete(kill_session))
|
||||
.route("/api/daemons", get(list_daemons))
|
||||
.route("/api/daemons/:id/dirs", get(list_dir))
|
||||
.route("/api/web/info", get(web_info))
|
||||
.route("/api/events", any(events_ws))
|
||||
.with_state(state)
|
||||
@@ -97,6 +100,46 @@ async fn list_daemons(State(s): State<LocalApiState>) -> Json<Vec<DaemonInfo>> {
|
||||
Json(out)
|
||||
}
|
||||
|
||||
/// Query string for `GET /api/daemons/:id/dirs`. `path` is the directory to
|
||||
/// list (the parent of what the user is typing); empty/`~`/relative are
|
||||
/// resolved against the target daemon's home dir.
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct DirQuery {
|
||||
#[serde(default)]
|
||||
path: String,
|
||||
}
|
||||
|
||||
/// List one directory's immediate subdirectories on `daemon_id`'s host, for the
|
||||
/// spawn form's working-directory autocomplete (NAP-329). Lists locally when
|
||||
/// the target is this daemon; otherwise brokers a `ListDir` to the owning
|
||||
/// daemon over the registry control link.
|
||||
async fn list_dir(
|
||||
State(s): State<LocalApiState>,
|
||||
Path(daemon_id): Path<String>,
|
||||
Query(q): Query<DirQuery>,
|
||||
) -> Response {
|
||||
if daemon_id == s.manager.daemon_id() {
|
||||
return match crate::dirlist::list_dir(&s.home_dir, &q.path) {
|
||||
Ok(listing) => Json(listing).into_response(),
|
||||
Err(info) => error_response(dir_error_status(&info.code), &info.code, &info.message),
|
||||
};
|
||||
}
|
||||
match s.broker.forward_list_dir(daemon_id, q.path).await {
|
||||
Ok(listing) => Json::<DirListing>(listing).into_response(),
|
||||
Err(e) => broker_error_response(&e),
|
||||
}
|
||||
}
|
||||
|
||||
/// HTTP status for a local [`crate::dirlist`] error code.
|
||||
fn dir_error_status(code: &str) -> StatusCode {
|
||||
match code {
|
||||
"not_found" => StatusCode::NOT_FOUND,
|
||||
"permission_denied" => StatusCode::FORBIDDEN,
|
||||
"not_a_directory" => StatusCode::BAD_REQUEST,
|
||||
_ => StatusCode::BAD_GATEWAY,
|
||||
}
|
||||
}
|
||||
|
||||
async fn web_info(State(s): State<LocalApiState>) -> Json<WebInfo> {
|
||||
Json(WebInfo {
|
||||
daemon_id: s.manager.daemon_id().to_string(),
|
||||
|
||||
@@ -311,6 +311,23 @@ impl RegistryClient {
|
||||
DaemonInbound::ForwardKillResult { reply_to, result } => {
|
||||
self.broker.complete_forward_kill(reply_to, result).await;
|
||||
}
|
||||
DaemonInbound::ListDir { reply_to, path } => {
|
||||
// List a directory on this host for a client's spawn-form
|
||||
// autocomplete (NAP-329). Resolved against this daemon's home.
|
||||
let result = match crate::dirlist::list_dir(&self.home_dir, &path) {
|
||||
Ok(listing) => RpcResult::Ok { value: listing },
|
||||
Err(error) => RpcResult::Err { error },
|
||||
};
|
||||
sink.send(Message::Text(serde_json::to_string(
|
||||
&DaemonOutbound::ListDirResult { reply_to, result },
|
||||
)?))
|
||||
.await?;
|
||||
}
|
||||
DaemonInbound::ForwardListDirResult { reply_to, result } => {
|
||||
self.broker
|
||||
.complete_forward_list_dir(reply_to, result)
|
||||
.await;
|
||||
}
|
||||
DaemonInbound::RelaySetup { .. } => {
|
||||
// T3 (relay) is gated on NAP-238. Lands here only so newer
|
||||
// registries sending it don't crash this daemon on parse.
|
||||
|
||||
@@ -43,6 +43,7 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness {
|
||||
manager.clone(),
|
||||
hostname,
|
||||
daemon_url,
|
||||
"/home/test",
|
||||
catalog.clone(),
|
||||
events,
|
||||
broker.clone(),
|
||||
|
||||
@@ -7,12 +7,13 @@
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use axum::extract::{ws::WebSocketUpgrade, Path, State};
|
||||
use axum::extract::{ws::WebSocketUpgrade, Path, Query, State};
|
||||
use axum::http::StatusCode;
|
||||
use axum::response::{IntoResponse, Response};
|
||||
use axum::routing::{any, get};
|
||||
use axum::{Json, Router};
|
||||
use maestro_protocol::{ConnectInfo, CreateSessionRequest, DaemonInfo, Session};
|
||||
use maestro_protocol::{ConnectInfo, CreateSessionRequest, DaemonInfo, DirListing, Session};
|
||||
use serde::Deserialize;
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::events::events_ws;
|
||||
@@ -34,6 +35,7 @@ pub fn routes(state: ClientApiState) -> Router {
|
||||
// connection-orchestration spec §"UI hosting"). Only the JSON/WS API here.
|
||||
Router::new()
|
||||
.route("/api/daemons", get(list_daemons))
|
||||
.route("/api/daemons/:id/dirs", get(list_dir))
|
||||
.route("/api/sessions", get(list_sessions).post(create_session))
|
||||
.route("/api/sessions/:id", axum::routing::delete(delete_session))
|
||||
.route("/api/sessions/:id/connect-info", get(session_connect_info))
|
||||
@@ -56,6 +58,64 @@ async fn list_sessions(State(s): State<ClientApiState>) -> Json<Vec<Session>> {
|
||||
Json(s.registry.sessions().await)
|
||||
}
|
||||
|
||||
/// Query string for `GET /api/daemons/:id/dirs`. `path` is the directory to
|
||||
/// list; empty/`~`/relative are resolved against the target daemon's home dir.
|
||||
#[derive(Debug, Deserialize)]
|
||||
struct DirQuery {
|
||||
#[serde(default)]
|
||||
path: String,
|
||||
}
|
||||
|
||||
/// List one directory's immediate subdirectories on `daemon_id`'s host, for the
|
||||
/// web spawn form's working-directory autocomplete (NAP-329). Drives a `ListDir`
|
||||
/// RPC to the target daemon, same machinery + error mapping as `create_session`.
|
||||
async fn list_dir(
|
||||
State(s): State<ClientApiState>,
|
||||
Path(daemon_id): Path<String>,
|
||||
Query(q): Query<DirQuery>,
|
||||
) -> Response {
|
||||
let outcome =
|
||||
tokio::time::timeout(s.rpc_timeout, s.registry.list_dir(&daemon_id, q.path)).await;
|
||||
match outcome {
|
||||
Err(_) => (
|
||||
StatusCode::GATEWAY_TIMEOUT,
|
||||
Json(error_body("timeout", "daemon did not reply in time")),
|
||||
)
|
||||
.into_response(),
|
||||
Ok(Err(RegistryError::DaemonNotConnected(_))) => (
|
||||
StatusCode::NOT_FOUND,
|
||||
Json(error_body("unknown_daemon", "daemon is not connected")),
|
||||
)
|
||||
.into_response(),
|
||||
Ok(Err(RegistryError::Timeout(_))) => (
|
||||
StatusCode::GATEWAY_TIMEOUT,
|
||||
Json(error_body("timeout", "daemon did not reply in time")),
|
||||
)
|
||||
.into_response(),
|
||||
Ok(Err(e)) => (
|
||||
StatusCode::BAD_GATEWAY,
|
||||
Json(error_body("transport", &e.to_string())),
|
||||
)
|
||||
.into_response(),
|
||||
Ok(Ok(Err(rpc_err))) => (
|
||||
dir_error_status(&rpc_err.0.code),
|
||||
Json(error_body(&rpc_err.0.code, &rpc_err.0.message)),
|
||||
)
|
||||
.into_response(),
|
||||
Ok(Ok(Ok(listing))) => Json::<DirListing>(listing).into_response(),
|
||||
}
|
||||
}
|
||||
|
||||
/// HTTP status for a daemon-reported directory-listing error code (NAP-329).
|
||||
fn dir_error_status(code: &str) -> StatusCode {
|
||||
match code {
|
||||
"not_found" => StatusCode::NOT_FOUND,
|
||||
"permission_denied" => StatusCode::FORBIDDEN,
|
||||
"not_a_directory" => StatusCode::BAD_REQUEST,
|
||||
_ => StatusCode::BAD_GATEWAY,
|
||||
}
|
||||
}
|
||||
|
||||
async fn create_session(
|
||||
State(s): State<ClientApiState>,
|
||||
Json(body): Json<CreateSessionRequest>,
|
||||
|
||||
@@ -227,6 +227,20 @@ async fn dispatch_text(
|
||||
.forward_kill(daemon_id, reply_to, &target, id)
|
||||
.await;
|
||||
}
|
||||
DaemonOutbound::ListDirResult { reply_to, result } => {
|
||||
registry.complete_list_dir(reply_to, result).await;
|
||||
}
|
||||
DaemonOutbound::ForwardListDir {
|
||||
reply_to,
|
||||
daemon_id: target,
|
||||
path,
|
||||
} => {
|
||||
// Relay a directory listing to its owning daemon and route the reply
|
||||
// back to this (home) daemon — control-plane only (NAP-329).
|
||||
registry
|
||||
.forward_list_dir(daemon_id, reply_to, &target, path)
|
||||
.await;
|
||||
}
|
||||
DaemonOutbound::RelayAck { reply_to, .. } => {
|
||||
// T3 (relay) is gated on NAP-238. Land the variant so newer daemons
|
||||
// don't crash the connection; the registry doesn't act on it yet.
|
||||
|
||||
@@ -15,7 +15,7 @@ use std::sync::{Arc, Mutex as StdMutex};
|
||||
use std::time::Duration;
|
||||
|
||||
use maestro_protocol::{
|
||||
AttachCandidates, CreateSessionRequest, DaemonInbound, DaemonInfo, ErrorInfo,
|
||||
AttachCandidates, CreateSessionRequest, DaemonInbound, DaemonInfo, DirListing, ErrorInfo,
|
||||
PerimeterCandidate, RegistryEvent, ReplyTo, RpcResult, Session,
|
||||
};
|
||||
use tokio::sync::{broadcast, mpsc, oneshot, Mutex};
|
||||
@@ -108,6 +108,9 @@ pub struct Registry {
|
||||
/// drop — including when the awaiting HTTP handler is cancelled (NAP-259).
|
||||
pending_create: Arc<StdMutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<Session>>>>>,
|
||||
pending_kill: Arc<StdMutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<()>>>>>,
|
||||
/// Outstanding `ListDir` RPC waiters (NAP-329). Same `PendingGuard`
|
||||
/// discipline as `pending_create` / `pending_kill`.
|
||||
pending_list_dir: Arc<StdMutex<HashMap<ReplyTo, oneshot::Sender<RpcResult<DirListing>>>>>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
@@ -155,6 +158,7 @@ impl Registry {
|
||||
events,
|
||||
pending_create: Arc::new(StdMutex::new(HashMap::new())),
|
||||
pending_kill: Arc::new(StdMutex::new(HashMap::new())),
|
||||
pending_list_dir: Arc::new(StdMutex::new(HashMap::new())),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -487,6 +491,30 @@ impl Registry {
|
||||
.await;
|
||||
}
|
||||
|
||||
/// Relay a home daemon's `ForwardListDir` to the target daemon and push the
|
||||
/// `ForwardListDirResult` back to the home daemon (NAP-329). Reuses
|
||||
/// [`list_dir`](Self::list_dir), then routes the reply to `home_daemon_id`.
|
||||
pub async fn forward_list_dir(
|
||||
&self,
|
||||
home_daemon_id: &str,
|
||||
reply_to: ReplyTo,
|
||||
target_daemon_id: &str,
|
||||
path: String,
|
||||
) {
|
||||
let result = match self.list_dir(target_daemon_id, path).await {
|
||||
Ok(Ok(listing)) => RpcResult::Ok { value: listing },
|
||||
Ok(Err(RpcError(error))) => RpcResult::Err { error },
|
||||
Err(e) => RpcResult::Err {
|
||||
error: registry_error_info(&e),
|
||||
},
|
||||
};
|
||||
self.reply_to_daemon(
|
||||
home_daemon_id,
|
||||
DaemonInbound::ForwardListDirResult { reply_to, result },
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
/// Send a single frame to a connected daemon's outbound channel, dropping
|
||||
/// it silently if the daemon has gone away mid-flight (the home daemon's
|
||||
/// own RPC timeout will surface the failure).
|
||||
@@ -518,6 +546,19 @@ impl Registry {
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve a ListDir RPC reply. Returns true if a waiter was found (NAP-329).
|
||||
pub async fn complete_list_dir(
|
||||
&self,
|
||||
reply_to: ReplyTo,
|
||||
result: RpcResult<DirListing>,
|
||||
) -> bool {
|
||||
let entry = self.pending_list_dir.lock().unwrap().remove(&reply_to);
|
||||
match entry {
|
||||
Some(tx) => tx.send(result).is_ok(),
|
||||
None => false,
|
||||
}
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Outbound R→D requests (internal API for NAP-215)
|
||||
// -----------------------------------------------------------------------
|
||||
@@ -610,6 +651,40 @@ impl Registry {
|
||||
Ok(rpc)
|
||||
}
|
||||
|
||||
/// Forward a `ListDir` to the named daemon and wait for the matching
|
||||
/// `ListDirResult` (NAP-329). No catalog side-effects — listings aren't
|
||||
/// cached registry-side. Mirrors [`create_session`](Self::create_session).
|
||||
pub async fn list_dir(
|
||||
&self,
|
||||
daemon_id: &str,
|
||||
path: String,
|
||||
) -> Result<Result<DirListing, RpcError>, RegistryError> {
|
||||
let reply_to = Uuid::new_v4();
|
||||
let (tx, rx) = oneshot::channel();
|
||||
let outbound = {
|
||||
let state = self.inner.lock().await;
|
||||
state
|
||||
.daemons
|
||||
.get(daemon_id)
|
||||
.ok_or_else(|| RegistryError::DaemonNotConnected(daemon_id.to_string()))?
|
||||
.outbound
|
||||
.clone()
|
||||
};
|
||||
self.pending_list_dir.lock().unwrap().insert(reply_to, tx);
|
||||
let _guard = PendingGuard {
|
||||
map: &*self.pending_list_dir,
|
||||
key: reply_to,
|
||||
};
|
||||
if outbound
|
||||
.send(DaemonInbound::ListDir { reply_to, path })
|
||||
.await
|
||||
.is_err()
|
||||
{
|
||||
return Err(RegistryError::DaemonGone);
|
||||
}
|
||||
await_rpc(rx).await.map(rpc_into_result)
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Read-only views (mostly for tests)
|
||||
// -----------------------------------------------------------------------
|
||||
@@ -720,6 +795,7 @@ mod tests {
|
||||
"host".into(),
|
||||
"0.1.0".into(),
|
||||
format!("ws://{daemon_id}:7070"),
|
||||
"/home/test".into(),
|
||||
tx,
|
||||
)
|
||||
.await;
|
||||
@@ -994,6 +1070,7 @@ mod nap259_tests {
|
||||
"host".into(),
|
||||
"0.1.0".into(),
|
||||
"ws://127.0.0.1:1".into(),
|
||||
"/home/test".into(),
|
||||
tx,
|
||||
)
|
||||
.await;
|
||||
|
||||
Reference in New Issue
Block a user