v5.3: initnsmgr event-driven deferred retry on O_NONBLOCK (Design B)

Implement Design B from local/docs/INITNSMGR-CONCURRENCY-DESIGN.md:
initnsmgr now uses O_NONBLOCK on openat to provider daemons and
parks requests when the provider is not ready, instead of
blocking the entire request loop.

Previously: open_scheme_resource did a blocking syscall::openat
to the provider. If the provider was briefly not servicing its
socket (descheduled under load, mid-tick, in a one-shot startup
window), the openat blocked, the loop stopped, and every other
request queued behind it. The wedge surfaced at whatever boot
stage was in flight (e.g. 'ahcid' in the logs) - that stage was
the symptom location, not the cause.

Now:
- openat to a provider is called with O_NONBLOCK.
- On Ok(fd), return the fd immediately.
- On EAGAIN, park the request in a bounded PendingOpens queue
  (tag, cap_fd, reference, flags, fcntl_flags).
- On any other error, return the error to the caller (preserves
  existing behavior).
- After each next_request cycle, retry ALL parked opens with
  O_NONBLOCK. Any that succeed are written back to the original
  request via the deferred Response API.
- The park queue is bounded (64 entries). On overflow, the
  request falls back to the synchronous path (same as before) -
  never unbounded growth.

This implements the traffic-driven retry model: as long as new
requests arrive (e.g. processes starting), parked opens are
re-attempted frequently. When the slow provider finally responds,
all parked opens succeed in the next retry sweep.

The kernel change (commit f5baa05d on submodule/kernel) is the
prerequisite: it implements O_NONBLOCK on OpenAt so the openat
syscall returns EAGAIN when the provider hasn't immediately
answered instead of blocking.

Preserves all existing handlers (dup, unlinkat, on_close,
on_sendfd, getdents, fstat, namespace:/"" branches) unchanged.
Only the openat -> open_scheme_resource path is changed.

Per local/AGENTS.md:
- No new branches (work on submodule/base)
- No stubs, no todo!/unimplemented!
- Cat 2 fork - changes committed to submodule/base branch

Closes v5.3 Design B (base side). Kernel side is the companion
commit on submodule/kernel.

bootstrap/src/initnsmgr.rs

@@ -1,3 +1,4 @@
+use alloc::collections::VecDeque;
 use alloc::string::{String, ToString};
 use alloc::sync::Arc;
 use alloc::vec::Vec;
@@ -17,12 +18,13 @@ use redox_rt::proc::FdGuard;
 // local/docs/INITNSMGR-CONCURRENCY-DESIGN.md.
 use redox_rt::sync::Mutex;
 use redox_scheme::{
-    CallerCtx, OpenResult, RequestKind, Response, SendFdRequest, SignalBehavior, Socket,
-    scheme::{SchemeState, SchemeSync},
+    CallerCtx, Id, IntoTag, OpenResult, RequestKind, Response, SendFdRequest, SignalBehavior,
+    Socket, Tag,
+    scheme::{Op, OpFdPathLike, SchemeState, SchemeSync},
 };
 use syscall::Stat;
 use syscall::dirent::{DirEntry, DirentBuf, DirentKind};
-use syscall::{CallFlags, FobtainFdFlags, error::*, schemev2::NewFdFlags};
+use syscall::{CallFlags, FobtainFdFlags, error::*, flag::O_NONBLOCK, schemev2::NewFdFlags};
 
 #[derive(Debug, Clone)]
 struct Namespace {
@@ -102,12 +104,23 @@ enum Handle {
     List(NamespaceAccess),
 }
 
+struct PendingOpen {
+    tag: Tag,
+    cap_fd: Arc<FdGuard>,
+    reference: String,
+    flags: usize,
+    fcntl_flags: u32,
+}
+
+const MAX_PENDING_OPENS: usize = 64;
+
 pub struct NamespaceScheme<'sock> {
     socket: &'sock Socket,
     handles: HashMap<usize, Handle>,
     root_namespace: Namespace,
     next_id: usize,
     scheme_creation_cap: FdGuard,
+    pending_opens: VecDeque<PendingOpen>,
 }
 
 const HIGH_PERMISSIONS: NsPermissions = NsPermissions::SCHEME_CREATE;
@@ -124,6 +137,7 @@ impl<'sock> NamespaceScheme<'sock> {
             root_namespace: Namespace { schemes },
             next_id: 0,
             scheme_creation_cap,
+            pending_opens: VecDeque::new(),
         }
     }
 
@@ -216,6 +230,120 @@ impl<'sock> NamespaceScheme<'sock> {
         self.next_id += 1;
         Ok(next_id)
     }
+
+    fn handle_scheme_open_nonblock(
+        &mut self,
+        openat: OpFdPathLike<usize>,
+    ) -> Option<Response> {
+        let ns_fd = openat.fd;
+        let fcntl_flags = openat.fcntl_flags;
+        let flags = *openat.flags();
+        let path_owned = openat.path().to_string();
+        let tag = openat.into_tag();
+
+        let redox_path = match RedoxPath::from_absolute(&path_owned) {
+            Some(rp) => rp,
+            None => return Some(Response::new(Err(Error::new(EINVAL)), tag)),
+        };
+        let (scheme_part, reference_part) = match redox_path.as_parts() {
+            Some(parts) => parts,
+            None => return Some(Response::new(Err(Error::new(EINVAL)), tag)),
+        };
+        let scheme_str = scheme_part.as_ref().to_string();
+        let reference_str = reference_part.as_ref().to_string();
+
+        let ns_access = match self.handles.get(&ns_fd) {
+            Some(Handle::Access(access)) => access.clone(),
+            _ => return Some(Response::new(Err(Error::new(ENOENT)), tag)),
+        };
+
+        let cap_fd = {
+            let ns = ns_access.namespace.lock();
+            match ns.get_scheme_fd(&scheme_str) {
+                Some(arc_fd) => arc_fd.clone(),
+                None => {
+                    log::info!("Scheme {:?} not found in namespace", scheme_str);
+                    return Some(Response::new(Err(Error::new(ENODEV)), tag));
+                }
+            }
+        };
+
+        let client_nonblock = flags & O_NONBLOCK != 0;
+        let nonblock_flags = flags | O_NONBLOCK;
+
+        match syscall::openat(
+            cap_fd.as_raw_fd(),
+            &reference_str,
+            nonblock_flags,
+            fcntl_flags as usize,
+        ) {
+            Ok(scheme_fd) => Some(Response::open_dup_like(
+                Ok(OpenResult::OtherScheme { fd: scheme_fd }),
+                tag,
+            )),
+            Err(e) if e.errno == EAGAIN && !client_nonblock => {
+                if self.pending_opens.len() >= MAX_PENDING_OPENS {
+                    let result = syscall::openat(
+                        cap_fd.as_raw_fd(),
+                        &reference_str,
+                        flags,
+                        fcntl_flags as usize,
+                    );
+                    Some(Response::open_dup_like(
+                        result.map(|fd| OpenResult::OtherScheme { fd }),
+                        tag,
+                    ))
+                } else {
+                    self.pending_opens.push_back(PendingOpen {
+                        tag,
+                        cap_fd,
+                        reference: reference_str,
+                        flags,
+                        fcntl_flags,
+                    });
+                    None
+                }
+            }
+            Err(e) => Some(Response::open_dup_like(Err(e), tag)),
+        }
+    }
+
+    fn retry_pending_opens(&mut self, socket: &Socket) {
+        let mut pending = mem::take(&mut self.pending_opens);
+
+        while let Some(p) = pending.pop_front() {
+            let nonblock_flags = p.flags | O_NONBLOCK;
+            match syscall::openat(
+                p.cap_fd.as_raw_fd(),
+                &p.reference,
+                nonblock_flags,
+                p.fcntl_flags as usize,
+            ) {
+                Ok(scheme_fd) => {
+                    let resp = Response::open_dup_like(
+                        Ok(OpenResult::OtherScheme { fd: scheme_fd }),
+                        p.tag,
+                    );
+                    let _ = socket.write_response(resp, SignalBehavior::Restart);
+                }
+                Err(e) if e.errno == EAGAIN => {
+                    pending.push_back(p);
+                }
+                Err(e) => {
+                    let resp = Response::open_dup_like(Err(e), p.tag);
+                    let _ = socket.write_response(resp, SignalBehavior::Restart);
+                }
+            }
+        }
+
+        self.pending_opens = pending;
+    }
+
+    fn cancel_pending_open(&mut self, id: Id) {
+        if let Some(idx) = self.pending_opens.iter().position(|p| p.tag.id() == id) {
+            let _ = self.pending_opens.remove(idx);
+        }
+    }
 }
 
 impl<'sock> SchemeSync for NamespaceScheme<'sock> {
@@ -506,6 +634,20 @@ where
     T::from_le_bytes_slice(buffer)
 }
 
+/// Returns `true` when the path targets a scheme other than `namespace` itself.
+/// Such opens are intercepted for nonblocking deferred retry; everything else
+/// (`"namespace:…"`, `"namespace:"`, listing) falls through to `handle_sync`.
+fn is_scheme_forwarded(path: &str) -> bool {
+    let Some(rp) = RedoxPath::from_absolute(path) else {
+        return false;
+    };
+    let Some((scheme, _)) = rp.as_parts() else {
+        return false;
+    };
+    let s = scheme.as_ref();
+    !s.is_empty() && s != "namespace"
+}
+
 pub fn run(
     sync_pipe: FdGuard,
     socket: Socket,
@@ -540,16 +682,40 @@ pub fn run(
             break;
         };
         match req.kind() {
-            RequestKind::Call(req) => {
-                let resp = req.handle_sync(&mut scheme, &mut state);
+            RequestKind::Call(call_req) => {
+                let caller = call_req.caller();
+                let resp_opt = match call_req.op() {
+                    Ok(op) => {
+                        let intercept = matches!(
+                            &op,
+                            Op::OpenAt(ref o) if is_scheme_forwarded(o.path())
+                        );
+                        if intercept {
+                            match op {
+                                Op::OpenAt(openat) => {
+                                    scheme.handle_scheme_open_nonblock(openat)
+                                }
+                                _ => unreachable!(),
+                            }
+                        } else {
+                            Some(op.handle_sync(caller, &mut scheme, &mut state))
+                        }
+                    }
+                    Err(this) => Some(Response::new(Err(Error::new(ENOSYS)), this)),
+                };
 
-                if !socket
-                    .write_response(resp, SignalBehavior::Restart)
-                    .expect("bootstrap: failed to write scheme response to kernel")
-                {
-                    break;
+                if let Some(resp) = resp_opt {
+                    if !socket
+                        .write_response(resp, SignalBehavior::Restart)
+                        .expect("bootstrap: failed to write scheme response to kernel")
+                    {
+                        break;
+                    }
                 }
             }
+            RequestKind::Cancellation(cancel_req) => {
+                scheme.cancel_pending_open(cancel_req.id);
+            }
             RequestKind::OnClose { id } => scheme.on_close(id),
             RequestKind::SendFd(sendfd_request) => {
                 let result = scheme.on_sendfd(&sendfd_request);
@@ -564,6 +730,8 @@ pub fn run(
 
             _ => (),
         }
+
+        scheme.retry_pending_opens(&socket);
     }
 
     unreachable!()