diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index f5ed4ff..668471a 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/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> Daemon { /// 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) -> DeskResponse { if let Err(refusal) = self.mind.require_instance(request.mind()) { @@ -245,15 +245,7 @@ impl> Daemon { } 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() }), } diff --git a/crates/huginn-daemon/src/desk_door.rs b/crates/huginn-daemon/src/desk_door.rs index b967351..72d9f38 100644 --- a/crates/huginn-daemon/src/desk_door.rs +++ b/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 { + inner: St, + deadline: Instant, +} + +impl Deadlined { + 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 Read for Deadlined { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.remaining(Timeout::Recv)?; + self.inner.read(buf) + } +} + +impl Write for Deadlined { + fn write(&mut self, buf: &[u8]) -> io::Result { + 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( 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>( 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) -> 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:#}"))), + } } diff --git a/crates/huginn-daemon/src/desk_door_tests.rs b/crates/huginn-daemon/src/desk_door_tests.rs index e48c765..60d09ad 100644 --- a/crates/huginn-daemon/src/desk_door_tests.rs +++ b/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] diff --git a/crates/huginn-mind/src/desk_wire.rs b/crates/huginn-mind/src/desk_wire.rs index 05db840..028f424 100644 --- a/crates/huginn-mind/src/desk_wire.rs +++ b/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` form, which no stored answer has. -pub fn answer_ref(answer: &ForkDeskAnswer) -> Option { - 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 { parse_n(key.strip_prefix(prefix(fork).as_str())?)