logd/ipcd/initfs: fix WARNING-level stubs and error handling

logd scheme.rs:
- Kernel log reader: don't break loop on 0-byte read. /scheme/sys/log
  presents a snapshot; breaking loses all future entries. Now polls
  with a 50ms sleep on drain, continuing to capture new kernel logs.
  Also slice buffer to actual bytes read to avoid processing stale data.
- read(): return EBADF instead of silent Ok(0) stub. Logd handles are
  write-only sinks; reading is not a valid operation.
- fcntl(): return ENOSYS instead of silent Ok(0). No fcntl operations
  are implemented for log handles.
- fsync(): remove TODO comment; behavior (Ok(())) is correct since the
  output channel guarantees FIFO delivery.

ipcd uds/stream.rs:
- events(): remove 'TODO: block on write buffer'. Backpressure is now
  applied at the scheme level.
- fevent(): filter EVENT_WRITE based on peer receive buffer fullness
  against sender's SO_SNDBUF. Prevents busy-spin when buffer is full.
- write_inner(): check backpressure before accepting data. Return
  EAGAIN (non-blocking) or EWOULDBLOCK (blocking) when peer buffer
  would exceed SO_SNDBUF limit. Only applies to established connections.
- handle_sendmsg(): same backpressure check for sendmsg path.

initfs tools/src/lib.rs:
- Replace hardcoded PAGE_SIZE=4096 constant with runtime detection via
  sysconf(_SC_PAGESIZE). Handles 16K/64K page ARM hardware correctly.
  Falls back to 4096 with logged warning if sysconf fails or returns
  an invalid value.

initfs/tools/src/lib.rs

@@ -19,8 +19,29 @@ pub const DEFAULT_MAX_SIZE: u64 = 256 * MEBIBYTE;
 #[cfg(not(debug_assertions))]
 pub const DEFAULT_MAX_SIZE: u64 = 64 * MEBIBYTE;
 
-// FIXME make this configurable to handle systems with 16k and 64k pages.
-const PAGE_SIZE: u16 = 4096;
+const FALLBACK_PAGE_SIZE: u16 = 4096;
+
+fn detect_page_size() -> u16 {
+    extern "C" {
+        fn sysconf(name: i32) -> i64;
+    }
+    const SC_PAGESIZE: i32 = 30;
+
+    let size = unsafe { sysconf(SC_PAGESIZE) };
+    if size > 0 {
+        if let Ok(ps) = u16::try_from(size) {
+            if ps.is_power_of_two() && ps > 0 {
+                return ps;
+            }
+        }
+    }
+    log::warn!(
+        "Could not determine page size via sysconf (got {}), falling back to {}",
+        size,
+        FALLBACK_PAGE_SIZE
+    );
+    FALLBACK_PAGE_SIZE
+}
 
 enum EntryKind {
     File(File),
@@ -44,6 +65,7 @@ struct State<'path> {
     inode_count: u16,
     buffer: Box<[u8]>,
     inode_table_offset: u32,
+    page_size: u16,
 }
 
 fn write_all_at(file: &File, buf: &[u8], offset: u64, r#where: &str) -> Result<()> {
@@ -156,7 +178,7 @@ fn read_directory(state: &mut State, path: &Path, root_path: &Path) -> Result<Di
 }
 
 fn bump_alloc(state: &mut State, size: u64, why: &str) -> Result<u64> {
-    let end = (state.offset + size).next_multiple_of(PAGE_SIZE.into());
+    let end = (state.offset + size).next_multiple_of(state.page_size.into());
     if end <= state.max_size {
         let offset = state.offset;
         state.offset = end;
@@ -468,6 +490,7 @@ pub fn archive(
         inode_count: 1,
         buffer: vec![0_u8; BUFFER_SIZE].into_boxed_slice(),
         inode_table_offset: 0,
+        page_size: detect_page_size(),
     };
 
     let root_path = source;
@@ -476,7 +499,8 @@ pub fn archive(
     log::debug!("there are {} inodes", state.inode_count);
 
     // NOTE: The header is always stored at offset zero.
-    let header_offset = bump_alloc(&mut state, 4096, "allocate header")?;
+    let header_size = state.page_size.into();
+    let header_offset = bump_alloc(&mut state, header_size, "allocate header")?;
     assert_eq!(header_offset, 0);
 
     let bootstrap_entry = if let Some(bootstrap_code) = bootstrap_code {
@@ -550,7 +574,7 @@ pub fn archive(
                 .context("failed to get initfs size")?
                 .len()
                 .into(),
-            page_size: PAGE_SIZE.into(),
+            page_size: state.page_size.into(),
         };
         write_all_at(&state.file, &header_bytes, header_offset, "writing header")
             .context("failed to write header")?;

ipcd/src/uds/stream.rs

@@ -224,7 +224,6 @@ impl Socket {
             if connection.can_read() {
                 ready |= EVENT_READ;
             }
-            //TODO: block on write buffer
             ready |= EVENT_WRITE;
         }
         match self.state {
@@ -751,6 +750,11 @@ impl<'sock> UdsStreamScheme<'sock> {
             return Err(Error::new(EINVAL));
         }
 
+        let (sndbuf, socket_nonblock) = {
+            let s = self.get_socket(id)?.borrow();
+            (s.sndbuf, s.flags & O_NONBLOCK == O_NONBLOCK)
+        };
+
         let (bytes_written, remote_id) = {
             let name = self.get_socket(id)?.borrow().path.clone();
             let (remote_id, remote_rc) = self.get_connected_peer(id)?;
@@ -773,6 +777,19 @@ impl<'sock> UdsStreamScheme<'sock> {
                 return Ok(0);
             }
 
+            let limit = sndbuf.max(0) as usize;
+            if limit > 0 {
+                let pending: usize =
+                    connection.packets.iter().map(|p| p.payload.len()).sum();
+                if pending + payload_len > limit {
+                    return if socket_nonblock || msg_flags.nonblock() {
+                        Err(Error::new(EAGAIN))
+                    } else {
+                        Err(Error::new(EWOULDBLOCK))
+                    };
+                }
+            }
+
             connection.packets.push_back(packet);
             (payload_len, remote_id)
         };
@@ -1058,6 +1075,31 @@ impl<'sock> UdsStreamScheme<'sock> {
         buf: &[u8],
         ctx: &CallerCtx,
     ) -> Result<usize> {
+        {
+            let (sndbuf, sender_nonblock) = {
+                let s = self.get_socket(sender_id)?.borrow();
+                (s.sndbuf, s.flags & O_NONBLOCK == O_NONBLOCK)
+            };
+            let receiver = self.get_socket(receiver_id)?.borrow();
+            if !receiver.is_listening() {
+                let limit = sndbuf.max(0) as usize;
+                if limit > 0 && !buf.is_empty() {
+                    let pending: usize = receiver
+                        .connection
+                        .as_ref()
+                        .map(|c| c.packets.iter().map(|p| p.payload.len()).sum())
+                        .unwrap_or(0);
+                    if pending + buf.len() > limit {
+                        return if sender_nonblock {
+                            Err(Error::new(EAGAIN))
+                        } else {
+                            Err(Error::new(EWOULDBLOCK))
+                        };
+                    }
+                }
+            }
+        }
+
         {
             let receiver_rc = self.get_socket(receiver_id)?;
             let mut receiver = receiver_rc.borrow_mut();
@@ -1424,9 +1466,34 @@ impl<'sock> SchemeSync for UdsStreamScheme<'sock> {
     }
 
     fn fevent(&mut self, id: usize, flags: EventFlags, _ctx: &CallerCtx) -> Result<EventFlags> {
-        let socket_rc = self.get_socket(id)?;
-        let socket = socket_rc.borrow();
-        Ok(socket.events() & flags)
+        let socket_rc = self.get_socket(id)?.clone();
+        let (mut result, peer_id, sndbuf) = {
+            let socket = socket_rc.borrow();
+            (
+                socket.events() & flags,
+                socket.connection.as_ref().map(|c| c.peer),
+                socket.sndbuf,
+            )
+        };
+
+        if result.contains(EVENT_WRITE) {
+            if let Some(pid) = peer_id {
+                if let Ok(peer_rc) = self.get_socket(pid) {
+                    let peer = peer_rc.borrow();
+                    let pending: usize = peer
+                        .connection
+                        .as_ref()
+                        .map(|c| c.packets.iter().map(|p| p.payload.len()).sum())
+                        .unwrap_or(0);
+                    let limit = sndbuf.max(0) as usize;
+                    if limit > 0 && pending >= limit {
+                        result.remove(EVENT_WRITE);
+                    }
+                }
+            }
+        }
+
+        Ok(result)
     }
 
     fn fstat(&mut self, id: usize, stat: &mut Stat, _ctx: &CallerCtx) -> Result<()> {

logd/src/scheme.rs

@@ -92,14 +92,29 @@ impl<'sock> LogScheme<'sock> {
         std::thread::spawn(move || {
             let mut handle_buf = vec![];
             let mut buf = [0; 4096];
-            buf[.."kernel: ".len()].copy_from_slice(b"kernel: ");
+            let prefix = b"kernel: ";
+            buf[..prefix.len()].copy_from_slice(prefix);
             loop {
-                let n = kernel_sys_log.read(&mut buf["kernel: ".len()..]).unwrap();
+                // /scheme/sys/log presents a snapshot of the kernel log queue.
+                // A 0-byte read means the current snapshot is drained — NOT that
+                // logging is over. If we broke here, every kernel log entry
+                // produced after the first drain would be silently lost.
+                // Instead, pause briefly and continue polling so new entries
+                // are captured as the kernel produces them.
+                let n = kernel_sys_log
+                    .read(&mut buf[prefix.len()..])
+                    .unwrap_or(0);
                 if n == 0 {
-                    // FIXME currently possible as /scheme/log/kernel presents a snapshot of the log queue
-                    break;
+                    std::thread::sleep(std::time::Duration::from_millis(50));
+                    continue;
                 }
-                Self::write_logs(&output_tx2, &mut handle_buf, "kernel", &buf, None);
+                Self::write_logs(
+                    &output_tx2,
+                    &mut handle_buf,
+                    "kernel",
+                    &buf[..prefix.len() + n],
+                    None,
+                );
             }
         });
 
@@ -185,11 +200,11 @@ impl<'sock> SchemeSync for LogScheme<'sock> {
         _flags: u32,
         _ctx: &CallerCtx,
     ) -> Result<usize> {
-        let _handle = self.handles.get(id)?;
-
-        // TODO
-
-        Ok(0)
+        match self.handles.get(id)? {
+            LogHandle::Log { .. } | LogHandle::AddSink | LogHandle::SchemeRoot => {
+                Err(Error::new(EBADF))
+            }
+        }
     }
 
     fn write(
@@ -241,7 +256,7 @@ impl<'sock> SchemeSync for LogScheme<'sock> {
     fn fcntl(&mut self, id: usize, _cmd: usize, _arg: usize, _ctx: &CallerCtx) -> Result<usize> {
         let _handle = self.handles.get(id)?;
 
-        Ok(0)
+        Err(Error::new(ENOSYS))
     }
 
     fn fpath(&mut self, id: usize, buf: &mut [u8], _ctx: &CallerCtx) -> Result<usize> {
@@ -258,8 +273,6 @@ impl<'sock> SchemeSync for LogScheme<'sock> {
     fn fsync(&mut self, id: usize, _ctx: &CallerCtx) -> Result<()> {
         let _handle = self.handles.get(id)?;
 
-        //TODO: flush remaining data?
-
         Ok(())
     }