Desk door fix batch: one deadline per connection (reads and writes spend what is left), the catalog reply bounded by the response limit, the dead index call and answer_ref deleted

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

crates/huginn-daemon/src/daemon.rs

@@ -5,7 +5,7 @@
 use chrono::{DateTime, Utc};
 use cultnet_rs::Selection;
 use crate::sources::forge::HttpForge;
-use huginn_mind::desk_wire::{DeskRequest, DeskResponse, answer_ref};
+use huginn_mind::desk_wire::{DeskRequest, DeskResponse};
 use huginn_mind::eureka_pipeline::{PipelineRef, Slug};
 use huginn_mind::wire::{ACTIVITY_REPOS_MAX, HuginnMindRequest, HuginnMindResponse, IndexStatus};
 use huginn_mind::{Members, Mind, MindRefusal, MindStore, PipelineAdmissionOutcome, SemanticQuery, SourceHits, SourceQuery};
@@ -237,7 +237,7 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
 
     /// One desk request, answered by the Mind method it names. The Mind is the instance's own
     /// (`mind` is checked against it); `deciders` is this call's list, handed to the Mind and
-    /// dropped. A fresh answer reaches the index as `Admit`'s writes do; a replay wrote nothing.
+    /// dropped. A desk answer is not indexed (`index_text` gives it no text), so nothing here tells the index.
     /// The reads wait on follow-up desk-door-reads and are refused until they exist.
     pub fn handle_desk(&mut self, request: DeskRequest, now: DateTime<Utc>) -> DeskResponse {
         if let Err(refusal) = self.mind.require_instance(request.mind()) {
@@ -245,15 +245,7 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
         }
         match request {
             DeskRequest::Answer { subject, authority, deciders, request, .. } => {
-                let (response, replayed) = self.mind.desk_answer(&deciders, subject, authority, request, now);
-                if !replayed
-                    && let DeskResponse::Answered { answer, .. } = &response
-                    && let Some(stored) = answer_ref(answer)
-                    && let Err(error) = self.index.committed(&self.mind, std::slice::from_ref(&stored))
-                {
-                    eprintln!("huginn: the index refused the landed write [{}]: {error:#}", stored.id.0);
-                }
-                response
+                self.mind.desk_answer(&deciders, subject, authority, request, now).0
             }
             DeskRequest::Forks { .. } | DeskRequest::Fork { .. } => DeskResponse::Refused(MindRefusal::Unavailable { detail: DESK_READS_UNBUILT.into() }),
         }

crates/huginn-daemon/src/desk_door.rs

@@ -5,13 +5,14 @@
 //! refused peer has nothing read and nothing written. The socket's life: bind, replace a stale
 //! one, refuse a live one or a file that is not a socket, mode 0660, unlink on drop. And the
 //! bounds: one request per connection, one reply, a header over `DESK_MAX_REQUEST_BYTES` refused
-//! before anything is allocated, `DESK_IO_TIMEOUT` on every read and write, a reply past
-//! `DESK_MAX_RESPONSE_BYTES` refused by name.
+//! before anything is allocated, one `DESK_CONNECTION_DEADLINE` for the whole connection (every
+//! read and write spends what is left of it, so a peer that trickles bytes is cut off at the
+//! deadline like one that is silent), a reply past `DESK_MAX_RESPONSE_BYTES` refused by name.
 //!
 //! It owns no rule. Which Mind method answers is `Daemon::handle_desk`'s; when the door is
 //! served is `serve::run`'s, once per pass in the thread that owns the Mind. A peer that is
-//! slow costs the loop at most `DESK_ACCEPTS_PER_PASS` connections of `2 * DESK_IO_TIMEOUT`
-//! each, and only the configured peer can reach that.
+//! slow costs the loop at most `DESK_ACCEPTS_PER_PASS` connections of `DESK_CONNECTION_DEADLINE`
+//! each, four seconds, and only the configured peer can reach that.
 //!
 //! The door keeps no request id: a retry after a lost reply is a new connection with the same
 //! request, and the stored answer is what replays it, across a restart as well.
@@ -23,7 +24,7 @@ use std::os::fd::AsFd;
 use std::os::unix::fs::{FileTypeExt, PermissionsExt};
 use std::os::unix::net::{UnixListener, UnixStream};
 use std::path::{Path, PathBuf};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use anyhow::{Context, Result, bail, ensure};
 use chrono::{DateTime, Utc};
@@ -38,12 +39,13 @@ use huginn_mind::desk_wire::{
 };
 use huginn_mind::envelope::{OperationFailure, decode_desk_request, encode_desk_failure, encode_desk_response};
 use huginn_mind::{MindRefusal, MindStore};
+use rustix::net::sockopt::{Timeout, set_socket_timeout};
 
 use crate::daemon::{Daemon, IndexSink};
 use crate::serve::{MAX_ECHOED_FIELD_BYTES, overlong_echo};
 
-/// How long the door waits on one read or one write of the configured peer.
-pub const DESK_IO_TIMEOUT: Duration = Duration::from_millis(500);
+/// How long one connection of the configured peer may hold the loop, reading and writing together.
+pub const DESK_CONNECTION_DEADLINE: Duration = Duration::from_secs(1);
 
 /// Connections one loop pass serves, so a peer that connects in a stream cannot hold the RUDP door off.
 pub const DESK_ACCEPTS_PER_PASS: u32 = 4;
@@ -145,11 +147,7 @@ impl DeskDoor {
                 }
             };
             taken += 1;
-            let served = serve_connection(stream, self.peer_uid, daemon, &self.registry, now, |stream| {
-                stream.set_nonblocking(false)?;
-                stream.set_read_timeout(Some(DESK_IO_TIMEOUT))?;
-                stream.set_write_timeout(Some(DESK_IO_TIMEOUT))
-            });
+            let served = serve_connection(stream, self.peer_uid, daemon, &self.registry, now, |stream| stream.set_nonblocking(false));
             if served != Served::Answered {
                 eprintln!("huginn: {served}");
             }
@@ -171,9 +169,49 @@ fn profile() -> CultNetTransportProfile {
     )
 }
 
-/// One connection: the peer first, before anything is read, then one frame in, one frame out.
-/// `ready` readies the stream for blocking, bounded I/O and runs only after the peer is known to
-/// be the configured one.
+/// A stream whose every read and write is bounded by what is left of one deadline: the socket's
+/// timeout is reset to the remainder before each call, so bytes arriving do not extend it.
+pub(crate) struct Deadlined<St> {
+    inner: St,
+    deadline: Instant,
+}
+
+impl<St: AsFd> Deadlined<St> {
+    pub(crate) fn new(inner: St, within: Duration) -> Self {
+        Self { inner, deadline: Instant::now() + within }
+    }
+
+    fn remaining(&self, which: Timeout) -> io::Result<()> {
+        // Whole milliseconds: rustix rounds a sub-microsecond tail up into an out-of-range tv_usec.
+        let left = Duration::from_millis(self.deadline.saturating_duration_since(Instant::now()).as_millis() as u64);
+        if left.is_zero() {
+            return Err(io::ErrorKind::TimedOut.into());
+        }
+        set_socket_timeout(&self.inner, which, Some(left)).map_err(io::Error::from)
+    }
+}
+
+impl<St: Read + AsFd> Read for Deadlined<St> {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        self.remaining(Timeout::Recv)?;
+        self.inner.read(buf)
+    }
+}
+
+impl<St: Write + AsFd> Write for Deadlined<St> {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.remaining(Timeout::Send)?;
+        self.inner.write(buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.inner.flush()
+    }
+}
+
+/// One connection: the peer first, before anything is read, then one frame in, one frame out,
+/// all inside `DESK_CONNECTION_DEADLINE`. `ready` readies the stream for blocking I/O and runs
+/// only after the peer is known to be the configured one.
 pub fn serve_connection<St, S, I>(
     stream: St,
     peer_uid: u32,
@@ -197,7 +235,7 @@ where
     if let Err(error) = ready(&stream) {
         return Served::Dropped(format!("the stream could not be readied: {error}"));
     }
-    let mut transport = TcpFramedTransportConnection::new(stream, profile());
+    let mut transport = TcpFramedTransportConnection::new(Deadlined::new(stream, DESK_CONNECTION_DEADLINE), profile());
     let frame = match transport.receive() {
         Ok(frame) => frame,
         Err(error) => return Served::Dropped(format!("no whole frame arrived: {error:#}")),
@@ -243,7 +281,14 @@ pub fn answer_desk_message<S: MindStore, I: IndexSink<S>>(
             Err(failure) => encode_desk_failure(message_id, operation, &failure, &runtime_id),
         },
         CultNetMessage::SchemaCatalogRequest { .. } => match registry.create_catalog_response(&message) {
-            Ok(response) => response,
+            Ok(reply) => bounded(reply, DESK_MAX_RESPONSE_BYTES, |size| CultNetMessage::Error {
+                error: match size {
+                    Ok(bytes) => format!("{DESK_SERVICE_ID} refuses a catalog reply of {bytes} bytes over {DESK_MAX_RESPONSE_BYTES}"),
+                    Err(detail) => detail,
+                },
+                code: None,
+                details: None,
+            }),
             Err(error) => CultNetMessage::Error { error: format!("{error:#}"), code: None, details: None },
         },
         _ => CultNetMessage::Error {
@@ -263,13 +308,21 @@ pub(crate) fn respond(response: &DeskResponse, message_id: &str, operation: &str
         Ok(reply) => reply,
         Err(error) => return not_encodable(format!("{error:#}")),
     };
-    let size = match encode_cultnet_message_to_vec(&reply, CultNetWireContract::CultNetSchemaV0) {
-        Ok(bytes) => bytes.len(),
-        Err(error) => return not_encodable(format!("{error:#}")),
-    };
-    if size <= limit {
-        return reply;
-    }
-    let refusal = DeskResponse::Refused(MindRefusal::ResponseTooLarge { bytes: size as u64, limit: limit as u64 });
-    encode_desk_response(message_id, operation, &refusal, runtime_id).unwrap_or_else(|error| not_encodable(format!("{error:#}")))
+    bounded(reply, limit, |size| match size {
+        Ok(bytes) => {
+            let refusal = DeskResponse::Refused(MindRefusal::ResponseTooLarge { bytes: bytes as u64, limit: limit as u64 });
+            encode_desk_response(message_id, operation, &refusal, runtime_id).unwrap_or_else(|error| not_encodable(format!("{error:#}")))
+        }
+        Err(detail) => not_encodable(detail),
+    })
+}
+
+/// `reply` itself when it encodes to at most `limit` bytes. Otherwise what `refuse` makes of the
+/// encoded size, or of the encoding error. Every reply the door sends passes here.
+fn bounded(reply: CultNetMessage, limit: usize, refuse: impl FnOnce(Result<usize, String>) -> CultNetMessage) -> CultNetMessage {
+    match encode_cultnet_message_to_vec(&reply, CultNetWireContract::CultNetSchemaV0) {
+        Ok(bytes) if bytes.len() <= limit => reply,
+        Ok(bytes) => refuse(Ok(bytes.len())),
+        Err(error) => refuse(Err(format!("{error:#}"))),
+    }
 }

crates/huginn-daemon/src/desk_door_tests.rs

@@ -30,7 +30,7 @@ use tempfile::TempDir;
 
 use crate::daemon::tests::{INSTANCE, OTHER, RecordingIndex, batch, campaign_seed, now, question_and_ruling, slug};
 use crate::daemon::Daemon;
-use crate::desk_door::{DeskDoor, Served, answer_desk_message, bind_desk, desk_registry, serve_connection};
+use crate::desk_door::{DESK_CONNECTION_DEADLINE, DeskDoor, Served, answer_desk_message, bind_desk, desk_registry, serve_connection};
 use crate::index::Backoff;
 use crate::serve::{Options, ServeOptions, bind, parse_options, run, schema_registry, startup};
 
@@ -342,17 +342,12 @@ mod door_smoke {
     fn a_counted_answer_over_the_socket_reads_back() {
         assert_ne!(my_uid(), 0, "the door refuses uid 0, so these tests cannot run as root");
         let mut rig = rig(my_uid());
-        let hash = rig.hash.clone();
+        let (before, hash) = (rig.state(), rig.hash.clone());
         let response = rig.serving(|path| reply_of(path, "m-1", &answer("x", DeskAuthority::Operator, &["x", "y"], wire(&hash, R1))));
         let DeskResponse::Answered { answer, counted_answer_id, .. } = &response else { panic!("{response:?}") };
         assert_eq!((answer.outcome.as_str(), counted_answer_id.as_deref(), answer.authority), ("Counted", Some("n1"), DeskAuthority::Operator));
-        let seen = rig.index.seen.lock().unwrap().clone();
-        assert_eq!(seen.len(), 1, "the fresh answer reached the index once");
-        let stored = huginn_mind::eureka_pipeline::PipelineRef {
-            kind: huginn_mind::eureka_pipeline::PipelineKind::DeskAnswer,
-            id: short(&seen[0].0[0]),
-        };
-        assert!(rig.daemon.mind().view(&stored).unwrap().is_some(), "the ref the index was given is the stored answer");
+        assert_eq!(rig.state().0, before.0 + 1, "the answer is stored");
+        assert!(rig.index.seen.lock().unwrap().is_empty(), "a desk answer is not indexed");
     }
 
     /// A pass takes at most `DESK_ACCEPTS_PER_PASS` waiting connections and says how many it took.
@@ -676,7 +671,7 @@ mod door_smoke {
         let (_, response) = decode_desk_response(&replayed).unwrap();
         assert_eq!(counted(response.unwrap()).0, "n1", "the retry is answered with the stored answer");
         assert_eq!(rig.state(), first_state, "the retry stored nothing");
-        assert_eq!(rig.index.seen.lock().unwrap().len(), 1, "the search index holds the answer once");
+        assert!(rig.index.seen.lock().unwrap().is_empty(), "a desk answer is not indexed");
     }
 
     #[test]
@@ -773,6 +768,7 @@ mod door_smoke {
 // ---- the bounds, pinned as the numbers the spec states ----
 
 mod bounds {
+    use crate::desk_door::Deadlined;
     use cultnet_rs::{CultNetSchemaKind, CultNetSchemaRegistration, CultNetSchemaRegistry};
     use huginn_mind::desk_wire::DESK_MAX_RESPONSE_BYTES;
 
@@ -815,6 +811,25 @@ mod bounds {
         });
     }
 
+    /// The write side of the deadline: a peer that stops reading is let go of at one second, the
+    /// socket's own send timeout reset to what is left before each write.
+    #[test]
+    fn a_peer_that_never_reads_is_cut_off_at_one_second() {
+        let (_peer, server) = UnixStream::pair().unwrap();
+        let mut stream = Deadlined::new(server, DESK_CONNECTION_DEADLINE);
+        let (chunk, started) = (vec![0_u8; 64 * 1024], Instant::now());
+        let error = loop {
+            if let Err(error) = stream.write(&chunk) {
+                break error;
+            }
+        };
+        let took = started.elapsed();
+        // The kernel's send timeout reads as WouldBlock; the deadline already spent reads as TimedOut.
+        assert!(matches!(error.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut), "{error:?}");
+        assert!(took >= Duration::from_millis(1000) && took < Duration::from_millis(1250), "the write was cut after {took:?}");
+        assert_eq!(stream.write(&chunk).unwrap_err().kind(), std::io::ErrorKind::TimedOut, "once spent, nothing more is written");
+    }
+
     /// A catalog is a reply like any other and is measured against the response bound: one that
     /// encodes to exactly the bound is sent, one byte more is refused naming the size.
     #[test]

crates/huginn-mind/src/desk_wire.rs

@@ -15,7 +15,7 @@
 
 use chrono::{DateTime, Utc};
 use eureka_pipeline::{
-    DeskAuthority, DeskOutcome, DeskRefusal, PipelineDeskAnswer, PipelineDocument, PipelineKind, PipelineQuestion, PipelineRef, RequestId, Sha256Hex, Short, Slug, Words,
+    DeskAuthority, DeskOutcome, DeskRefusal, PipelineDeskAnswer, PipelineDocument, PipelineKind, PipelineQuestion, RequestId, Sha256Hex, Short, Slug, Words,
 };
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
@@ -330,12 +330,6 @@ fn answer_key(fork: &str, n: u32) -> String {
     format!("{}{}", prefix(fork), n_text(n))
 }
 
-/// The record a projected answer was stored under: the ref the index is told about when the
-/// answer is fresh. `None` for an id outside the `n<seq>` form, which no stored answer has.
-pub fn answer_ref(answer: &ForkDeskAnswer) -> Option<PipelineRef> {
-    Some(PipelineRef { kind: PipelineKind::DeskAnswer, id: Short(answer_key(&answer.fork_id, parse_n(&answer.answer_id)?)) })
-}
-
 /// The sequence a key of `fork`'s answers names; `None` for another fork's key or any other text.
 fn answer_seq(fork: &str, key: &str) -> Option<u32> {
     parse_n(key.strip_prefix(prefix(fork).as_str())?)