diff --git a/crates/huginn-daemon/src/daemon.rs b/crates/huginn-daemon/src/daemon.rs index c468b39..9ab2dd8 100644 --- a/crates/huginn-daemon/src/daemon.rs +++ b/crates/huginn-daemon/src/daemon.rs @@ -104,8 +104,10 @@ pub(crate) mod tests { use super::*; use chrono::TimeZone; use huginn_mind::epiphany_pipeline::{ - Date, OrgRepo, PipelineDocument, PipelineInstance, PipelineKind, PipelineQuestion, PipelineRuling, - QuestionOption, RulingAuthority, Short, + AuthorityMap, CodeLocation, CutDelete, CutVerification, Date, DocRef, FileChange, Line, NegativeCheck, + OrgRepo, PipelineCampaign, PipelineCutSpec, PipelineDocument, PipelineInstance, PipelineKind, + PipelineQuestion, PipelineRuling, PipelineStewardship, QuestionOption, RulingAuthority, Sha, Short, + StructuralDelta, VerificationTest, }; use huginn_mind::wire::MindStatus; use huginn_mind::{ @@ -183,6 +185,134 @@ pub(crate) mod tests { (question, ruling) } + pub(crate) const REPO: &str = "GameCult/Epiphany"; + + /// A `Line` and a `Short` at the leaf's own bound, so a document built out + /// of them is as wide as any writer may make it. + pub(crate) fn line(n: usize) -> Line { + Line(format!("{n:0>1000}")) + } + + pub(crate) fn short(n: usize) -> Short { + Short(format!("{n:0>200}")) + } + + fn sha() -> Sha { + Sha("5f98228d".into()) + } + + fn doc_ref() -> DocRef { + DocRef { path: Short("notes/target.md".into()), start_line: 1, end_line: 9, commit: sha() } + } + + /// The three documents a cut spec needs before it may be admitted: the + /// mind's identity, its stewardship of the repo, and the campaign that + /// names the repo. + pub(crate) fn campaign_seed() -> Vec { + vec![ + identity(INSTANCE), + PipelineDocument::Stewardship(PipelineStewardship { + instance: slug(INSTANCE), + repo: OrgRepo(REPO.into()), + sequence: 1, + assigned_on: Date("2026-09-16".into()), + note: "assigned".into(), + }), + PipelineDocument::Campaign(PipelineCampaign { + slug: slug(CAMPAIGN), + title: Short("Eureka pipeline state".into()), + repos: vec![OrgRepo(REPO.into())], + working_branch: Short("codex/eureka-pipeline-state".into()), + target_doc: doc_ref(), + }), + ] + } + + /// One cut spec with every list the leaf bounds filled to its maximum and + /// every text at its own: `file_changes` file changes, 64 deletes, keeps, + /// adds, builds, tests, negative checks and operator checks, a full + /// authority map and a full structural delta. At 256 file changes that is + /// roughly a megabyte of field content, which is what a real-sized answer + /// looks like at the leaf's bounds rather than at a fixture's convenience. + /// `depends_on`, `rulings` and `questions` stay empty because each is a + /// reference admission resolves. + pub(crate) fn cut_spec(cut: &str, file_changes: usize) -> PipelineDocument { + let lines = |count: usize| (0..count).map(line).collect::>(); + let shorts = |count: usize| (0..count).map(short).collect::>(); + let location = |n: usize| CodeLocation { path: short(n), line: 1, end_line: Some(9) }; + PipelineDocument::CutSpec(PipelineCutSpec { + campaign: slug(CAMPAIGN), + cut: cut.into(), + revision: 1, + title: short(0), + repo: OrgRepo(REPO.into()), + branch: short(1), + base: sha(), + depends_on: vec![], + first: lines(16), + deletes: (0..64).map(|n| CutDelete { path: short(n), lines: 9, note: line(n) }).collect(), + keeps_moves: lines(64), + adds: lines(64), + file_changes: (0..file_changes).map(|n| FileChange { location: location(n), change: line(n) }).collect(), + authority_map: Some(AuthorityMap { + owner: line(0), + inputs: lines(16), + outputs: lines(16), + derived_state: lines(16), + forbidden_writers: lines(16), + shared_paths: lines(16), + deletion_line: line(1), + }), + verification: CutVerification { + builds: lines(64), + tests: (0..64).map(|n| VerificationTest { name: short(n), pins: line(n) }).collect(), + negative: (0..64).map(|n| NegativeCheck { pattern: short(n), scope: line(n) }).collect(), + operator: lines(64), + }, + estimate: StructuralDelta { + lines_added: 0, + lines_removed: 0, + dependencies_added: shorts(64), + dependencies_removed: shorts(64), + formats_added: shorts(64), + formats_removed: shorts(64), + targets_added: shorts(64), + targets_removed: shorts(64), + }, + rulings: vec![], + questions: vec![], + }) + } + + /// The cut spec at the leaf's own maximum, and one narrow enough that its + /// answer still fits a single send. `WIDE_CHANGES` is the leaf's bound; + /// `FITTING_CHANGES` is chosen so the view of that spec lands inside the + /// window with less than a packet-count's slack, which is what makes the + /// window's own value load-bearing rather than merely generous. + pub(crate) const WIDE_CUT: &str = "wide"; + pub(crate) const WIDE_CHANGES: usize = 256; + pub(crate) const FITTING_CUT: &str = "fits"; + pub(crate) const FITTING_CHANGES: usize = 180; + + /// A daemon whose mind holds the campaign seed, one cut spec as wide as the + /// leaf permits, and one narrow enough to answer, so a read over it is a + /// real-sized answer either way. + pub(crate) fn seeded_wide() -> (TempDir, Daemon) { + let root = tempfile::tempdir().unwrap(); + let mut daemon = Daemon::open(root.path(), &slug(INSTANCE)).unwrap(); + for documents in [ + campaign_seed(), + vec![cut_spec(WIDE_CUT, WIDE_CHANGES), cut_spec(FITTING_CUT, FITTING_CHANGES)], + ] { + let outcome = daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), now()); + assert!( + matches!(outcome, HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. })), + "{outcome:?}" + ); + } + (root, daemon) + } + /// A daemon over a real redb mind in a temporary directory, with its /// identity admitted. The directory is returned because dropping it /// removes the store. diff --git a/crates/huginn-daemon/src/serve.rs b/crates/huginn-daemon/src/serve.rs index 6c6ac65..851d765 100644 --- a/crates/huginn-daemon/src/serve.rs +++ b/crates/huginn-daemon/src/serve.rs @@ -5,10 +5,35 @@ //! Stated limits. A socket error and a hostile datagram are indistinguishable //! at `receive_event_once`, so both are logged and served past: a dead socket //! spins with logging rather than exiting, and Cut 14's health check is the -//! observer. The reliable window is `max_pending_reliable_packets` times -//! `max_fragment_bytes` per session, about 1.2 MB in flight; a page larger -//! than that fails to send and the client times out rather than receiving a -//! typed answer. +//! observer. +//! +//! One answer may be `MAX_RESPONSE_BYTES` encoded, 1,228,800 today: the +//! reliable window one session holds, `MAX_PENDING_RELIABLE_PACKETS` packets +//! of `MAX_FRAGMENT_BYTES`. A larger answer is refused by name, +//! `MindRefusal::ResponseTooLarge`, carrying the encoded size and the limit, +//! and the caller narrows its own request. This is the current bound and not a +//! design target: a single cut spec at the leaf's own bounds does not fit it, +//! and what a list read should return is an open question, so a client author +//! should expect the number to move. +//! +//! Four things this cut does not do, which a client has nowhere else to learn: +//! +//! - A session idle for `ServeOptions::session_timeout`, 30 seconds, is +//! removed here while the client still believes it is connected; its next +//! request is answered by nothing. A client that intends to stay must speak +//! inside that window or reconnect. +//! - An envelope CultNet's own `validate_message` rejects is dropped by the +//! hub before this module sees it, so it is never answered. +//! - Shutdown neither drains what is in flight nor disconnects its peers: the +//! loop stops and the sockets close under whatever was queued. +//! - `response-not-encodable` is authored here and is unreachable in practice; +//! every response type encodes. +//! +//! Two reads pipelined on one session are not both answered: the window is +//! per session and counts what is unacknowledged, so a second large reply sent +//! before the first is acknowledged fails to send even when each passes the +//! size gate on its own. The gate measures one answer against the window, not +//! against what the session still has room for. use std::collections::BTreeMap; use std::net::{SocketAddr, UdpSocket}; @@ -21,13 +46,14 @@ use chrono::{DateTime, Utc}; use cultnet_rs::{ CULTNET_OPERATION_CONNECTION_ID, CultNetMessage, CultNetRudpServerEvent, CultNetRudpServerHub, CultNetRudpServerHubOptions, CultNetSchemaKind, CultNetSchemaRegistration, CultNetSchemaRegistry, - CultNetWireContract, decode_cultnet_message_from_slice, + CultNetWireContract, decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, }; use huginn_mind::epiphany_pipeline::Slug; use huginn_mind::wire::{ - MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA, MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, + HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA, + MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, }; -use huginn_mind::{MindStore, OwnedRedbMessagePackBackingStore}; +use huginn_mind::{MindRefusal, MindStore, OwnedRedbMessagePackBackingStore}; use crate::daemon::{Daemon, IndexSink, NoIndex}; use crate::envelope::{decode_request, encode_failure, encode_response}; @@ -87,14 +113,23 @@ pub fn startup( Ok((daemon, hub, schema_registry()?)) } +/// The bytes one reliable packet carries, and the packets one session may hold +/// unacknowledged. Their product is the largest answer a single send can +/// carry, and `answer` measures every reply against it rather than handing the +/// hub a send it can already see will fail. +pub const MAX_FRAGMENT_BYTES: u32 = 1200; +pub const MAX_PENDING_RELIABLE_PACKETS: u32 = 1024; +/// 1,228,800 bytes. +pub const MAX_RESPONSE_BYTES: u64 = MAX_FRAGMENT_BYTES as u64 * MAX_PENDING_RELIABLE_PACKETS as u64; + /// One non-blocking socket serving the operation connection id. The hub drops /// every datagram carrying another one. pub fn bind(addr: SocketAddr, runtime_id: &str) -> Result { let socket = UdpSocket::bind(addr).with_context(|| format!("binding {addr}"))?; socket.set_nonblocking(true)?; let mut options = CultNetRudpServerHubOptions::new(runtime_id, socket, CULTNET_OPERATION_CONNECTION_ID); - options.max_fragment_bytes = Some(1200); - options.max_pending_reliable_packets = Some(1024); + options.max_fragment_bytes = Some(MAX_FRAGMENT_BYTES); + options.max_pending_reliable_packets = Some(MAX_PENDING_RELIABLE_PACKETS); options.max_peers = 256; CultNetRudpServerHub::new(options) } @@ -133,18 +168,8 @@ pub fn answer>( Ok((message_id, request)) => { let operation = request.operation(); let response = daemon.handle(request, now); - match encode_response(&message_id, operation, &response, &runtime_id) { - Ok(message) => message, - Err(error) => encode_failure( - &message_id, - operation, - &crate::envelope::OperationFailure { - code: "response-not-encodable".into(), - message: format!("{error:#}"), - }, - &runtime_id, - ), - } + let reply = encode_or_fail(&message_id, operation, &response, &runtime_id); + within_window(reply, &message_id, operation, &runtime_id) } Err(failure) => encode_failure(message_id, operation, &failure, &runtime_id), }, @@ -160,6 +185,63 @@ pub fn answer>( } } +/// One response in its envelope, or the failure that says it did not encode. +fn encode_or_fail( + message_id: &str, + operation: &str, + response: &HuginnMindResponse, + runtime_id: &str, +) -> CultNetMessage { + match encode_response(message_id, operation, response, runtime_id) { + Ok(message) => message, + Err(error) => encode_failure( + message_id, + operation, + &crate::envelope::OperationFailure { + code: "response-not-encodable".into(), + message: format!("{error:#}"), + }, + runtime_id, + ), + } +} + +/// The reply, or a typed refusal saying it will not fit. The hub cuts a reply +/// into `MAX_FRAGMENT_BYTES` packets and refuses a send needing more than +/// `MAX_PENDING_RELIABLE_PACKETS` of them; that refusal is a transport error +/// the client never sees, so it waits for an answer nothing will send. The +/// size is therefore measured here, against the window this module configured, +/// before a send that can already be seen to fail is attempted, and the client +/// is told by name how large the answer was and how large one may be. +/// +/// Nothing is truncated, paginated or retried: the caller narrows its own +/// request. The refusal rides the response schema like every other refusal, so +/// it is not a second thing for a client to parse. +fn within_window(reply: CultNetMessage, message_id: &str, operation: &str, runtime_id: &str) -> CultNetMessage { + let encoded = match encode_cultnet_message_to_vec(&reply, CultNetWireContract::CultNetSchemaV0) { + Ok(bytes) => bytes.len() as u64, + Err(error) => { + return encode_failure( + message_id, + operation, + &crate::envelope::OperationFailure { + code: "response-not-encodable".into(), + message: format!("{error:#}"), + }, + runtime_id, + ); + } + }; + if encoded <= MAX_RESPONSE_BYTES { + return reply; + } + let refusal = HuginnMindResponse::Refused(MindRefusal::ResponseTooLarge { + bytes: encoded, + limit: MAX_RESPONSE_BYTES, + }); + encode_or_fail(message_id, operation, &refusal, runtime_id) +} + /// Until `stopping`: expire what timed out, resend what was not acknowledged, /// then answer every frame waiting on the socket. A hostile datagram and a /// departed session are logged and served past, never fatal. This is the @@ -217,7 +299,11 @@ pub fn run>( #[cfg(test)] mod tests { use super::*; - use crate::daemon::tests::{INSTANCE, OTHER, batch, identity, now, slug}; + use crate::daemon::tests::{ + CAMPAIGN, FITTING_CHANGES, FITTING_CUT, INSTANCE, OTHER, WIDE_CHANGES, WIDE_CUT, batch, identity, now, + seeded_wide, slug, + }; + use huginn_mind::epiphany_pipeline::{PipelineKind, PipelineRef}; use crate::envelope::{FAILURE_SCHEMA, decode_response, encode_request}; use base64::Engine; use base64::engine::general_purpose::STANDARD; @@ -436,6 +522,147 @@ mod tests { assert_eq!(documents(&mut daemon), 1, "the admission that crossed the wire landed"); } + /// An answer larger than one send can carry is a refusal by name, not a + /// transport failure the client waits out. The wide document is one cut spec + /// with every list the leaf bounds filled: about a megabyte of field + /// content, which base64 in the operation envelope carries past the window + /// this module configures, so a single read of it is already over. The + /// fitting one is the same document narrowed until its answer lands inside + /// the window, and it is delivered whole. + /// + /// Both halves are load-bearing. Without the refused reads a daemon that + /// never measured would pass; without the delivered one a daemon that + /// refused everything, or configured a window smaller than the number it + /// measures against, would pass too. The delivered answer sits within one + /// packet-count's slack of the limit, so shrinking the configured window + /// breaks it. + #[test] + fn an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client() { + let (_root, mut daemon) = seeded_wide(); + let registry = schema_registry().unwrap(); + + let sized = |daemon: &mut Daemon, id: PipelineRef| { + let view = daemon.handle(HuginnMindRequest::View { instance: slug(INSTANCE), id }, now()); + let message = encode_response("m-0", "view", &view, "huginn-yggdrasil").unwrap(); + encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap().len() as u64 + }; + let wide = spec_ref(&mut daemon, WIDE_CUT); + let fitting = spec_ref(&mut daemon, FITTING_CUT); + let (over, under) = (sized(&mut daemon, wide.clone()), sized(&mut daemon, fitting.clone())); + eprintln!("a view of {WIDE_CHANGES} file changes encodes to {over} bytes, of {FITTING_CHANGES} to {under}"); + assert!(over > MAX_RESPONSE_BYTES, "a cut spec at the leaf's bounds is {over} bytes, over {MAX_RESPONSE_BYTES}"); + assert!(under <= MAX_RESPONSE_BYTES, "the narrowed spec is {under} bytes and must fit"); + assert!( + MAX_RESPONSE_BYTES - under < 64 * MAX_FRAGMENT_BYTES as u64, + "the delivered answer must sit close enough to the limit that the window's value is load-bearing" + ); + + let mut hub = bind("127.0.0.1:0".parse().unwrap(), &daemon.runtime_id()).unwrap(); + let endpoint = format!("rudp://{}", hub.local_addr().unwrap()); + let mut client = CultMesh::create_rudp_client_for_endpoint( + "eureka-state-wide".to_string(), + CULTNET_OPERATION_CONNECTION_ID, + &endpoint, + CultMeshRudpSocketOptions::default(), + ) + .unwrap(); + client.connect(Vec::new()).unwrap(); + for _ in 0..200 { + while hub.receive_event_once().unwrap().is_some() {} + let _ = client.receive_once(); + if client.connected() { + break; + } + std::thread::sleep(Duration::from_millis(2)); + } + assert!(client.connected()); + + let reads = [ + ("m-q", HuginnMindRequest::Query { instance: slug(INSTANCE), query: PipelineQuery::default() }), + ("m-v", HuginnMindRequest::View { instance: slug(INSTANCE), id: wide }), + ("m-o", HuginnMindRequest::OpenItems { instance: slug(INSTANCE), campaign: slug(CAMPAIGN) }), + ]; + let stopping = Arc::new(AtomicBool::new(false)); + let loop_stopping = Arc::clone(&stopping); + let serving = std::thread::spawn(move || { + let result = run(&mut daemon, &mut hub, ®istry, &loop_stopping, &ServeOptions::default()); + (result, daemon) + }); + + // One read at a time: the window is per session and counts what is + // still unacknowledged, so two large answers in flight together are a + // separate limit this gate does not measure. + for (message_id, request) in reads { + client.send_schema_message(&encode_request(message_id, &request, None).unwrap()).unwrap(); + let mut reply = None; + for _ in 0..500 { + client.poll_resends().unwrap(); + if let Some(message) = client.receive_schema_message_once().unwrap() { + reply = Some(message); + break; + } + std::thread::sleep(Duration::from_millis(2)); + } + let reply = reply.unwrap_or_else(|| panic!("{message_id} got no reply")); + let (correlation, answered) = decode_response(&reply).unwrap(); + assert_eq!(&correlation, message_id); + let HuginnMindResponse::Refused(MindRefusal::ResponseTooLarge { bytes, limit }) = + answered.expect("a refusal is an answer, not an envelope failure") + else { + panic!("{message_id} was not refused for its size"); + }; + assert_eq!(limit, MAX_RESPONSE_BYTES); + assert!(bytes > limit, "{message_id} answered {bytes} bytes against a {limit} limit"); + } + + // A real-sized answer that does fit is delivered whole, over the same + // session that was just refused twice: the gate measures the answer, and + // the window it measures against is one the hub actually carries. + let delivered = [ + ("m-f", HuginnMindRequest::View { instance: slug(INSTANCE), id: fitting.clone() }), + ("m-w", HuginnMindRequest::Whoami), + ]; + for (message_id, request) in delivered { + client.send_schema_message(&encode_request(message_id, &request, None).unwrap()).unwrap(); + let mut reply = None; + for _ in 0..2000 { + client.poll_resends().unwrap(); + if let Some(message) = client.receive_schema_message_once().unwrap() { + reply = Some(message); + break; + } + std::thread::sleep(Duration::from_millis(2)); + } + let reply = reply.unwrap_or_else(|| panic!("{message_id} got no reply")); + let (correlation, answered) = decode_response(&reply).unwrap(); + assert_eq!(&correlation, message_id); + match answered.unwrap() { + HuginnMindResponse::View(Some(view)) => assert_eq!(view.id, fitting), + HuginnMindResponse::Whoami(status) => assert_eq!(status.documents, 5), + other => panic!("{message_id} got {other:?}"), + } + } + + stopping.store(true, Ordering::Relaxed); + serving.join().unwrap().0.unwrap(); + } + + /// One cut spec's own reference, by its cut label. + fn spec_ref(daemon: &mut Daemon, cut: &str) -> PipelineRef { + let query = PipelineQuery { + kinds: vec![PipelineKind::CutSpec], + cut: Some(cut.into()), + ..PipelineQuery::default() + }; + let HuginnMindResponse::Query(page) = + daemon.handle(HuginnMindRequest::Query { instance: slug(INSTANCE), query }, now()) + else { + panic!("expected a page"); + }; + assert_eq!(page.matched, 1, "one cut spec is labelled {cut}"); + page.items[0].id.clone() + } + /// Ruling 15 as an order in the one place both happen. Both gates are shut /// at once: the mind is held and the port is held. A startup that opens /// first says why the mind refused; one that binds first can only say the diff --git a/crates/huginn-mind/src/refusal.rs b/crates/huginn-mind/src/refusal.rs index 2087414..21476a0 100644 --- a/crates/huginn-mind/src/refusal.rs +++ b/crates/huginn-mind/src/refusal.rs @@ -54,6 +54,15 @@ pub enum MindRefusal { FindingWithoutEvidence, UnknownInvariant { label: String }, NotStewarded { repo: String }, + /// The one refusal a mind never raises: the answer does not fit the + /// transport that would have carried it. It lives here because a refusal + /// rides the response schema and there is one of those, so a client reads + /// it the way it reads every other refusal rather than parsing a second + /// vocabulary. `bytes` is the encoded answer and `limit` is what one send + /// may carry, both so a caller can narrow its own request; nothing is + /// truncated or paginated on its behalf. The daemon owns it and the + /// transport's own bound decides it; no rule of the mind is involved. + ResponseTooLarge { bytes: u64, limit: u64 }, Unavailable { detail: String }, } diff --git a/schemas/cultnet/huginn.mind_response.v1.schema.json b/schemas/cultnet/huginn.mind_response.v1.schema.json index 89e5db5..aa7b901 100644 --- a/schemas/cultnet/huginn.mind_response.v1.schema.json +++ b/schemas/cultnet/huginn.mind_response.v1.schema.json @@ -1201,6 +1201,35 @@ "NotStewarded" ] }, + { + "description": "The one refusal a mind never raises: the answer does not fit the\ntransport that would have carried it. It lives here because a refusal\nrides the response schema and there is one of those, so a client reads\nit the way it reads every other refusal rather than parsing a second\nvocabulary. `bytes` is the encoded answer and `limit` is what one send\nmay carry, both so a caller can narrow its own request; nothing is\ntruncated or paginated on its behalf. The daemon owns it and the\ntransport's own bound decides it; no rule of the mind is involved.", + "type": "object", + "properties": { + "ResponseTooLarge": { + "type": "object", + "properties": { + "bytes": { + "type": "integer", + "format": "uint64", + "minimum": 0 + }, + "limit": { + "type": "integer", + "format": "uint64", + "minimum": 0 + } + }, + "required": [ + "bytes", + "limit" + ] + } + }, + "additionalProperties": false, + "required": [ + "ResponseTooLarge" + ] + }, { "type": "object", "properties": { diff --git a/tools/eureka-cut10-mutations.psd1 b/tools/eureka-cut10-mutations.psd1 index 55acba6..ebe8dff 100644 --- a/tools/eureka-cut10-mutations.psd1 +++ b/tools/eureka-cut10-mutations.psd1 @@ -1,4 +1,4 @@ -# Cut 10 mutations, D1-D19: the CultNet surface's rules. Each rule has a +# Cut 10 mutations, D1-D22: the CultNet surface's rules. Each rule has a # revert, `Dn`, which removes the rule, and a loosening, `DnL`, which weakens # it rather than removing it, because a plain revert is the easy target. Every # entry names the test that must fail while it is applied. Run through @@ -156,13 +156,9 @@ Rule = 'A refusal is data in the response, never a transport failure the client must parse out of a failure envelope.' Test = 'serve::tests::a_malformed_envelope_is_answered_with_a_failure_and_touches_no_mind' File = 'crates/huginn-daemon/src/serve.rs' - Old = @' - let response = daemon.handle(request, now); - match encode_response(&message_id, operation, &response, &runtime_id) { -'@ + Old = ' let reply = encode_or_fail(&message_id, operation, &response, &runtime_id);' New = @' - let response = daemon.handle(request, now); - if let huginn_mind::wire::HuginnMindResponse::Refused(refusal) = &response { + if let HuginnMindResponse::Refused(refusal) = &response { return encode_failure( &message_id, operation, @@ -173,7 +169,7 @@ &runtime_id, ); } - match encode_response(&message_id, operation, &response, &runtime_id) { + let reply = encode_or_fail(&message_id, operation, &response, &runtime_id); '@ } @{ @@ -237,6 +233,86 @@ Old = ' if operation != request.operation() {' New = ' if operation.len() != request.operation().len() {' } + @{ + Id = 'D20' + Rule = 'An answer larger than one send can carry is refused by name, not handed to a send that fails silently.' + Test = 'serve::tests::an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client' + File = 'crates/huginn-daemon/src/serve.rs' + Old = ' if encoded <= MAX_RESPONSE_BYTES {' + New = ' if true {' + } + @{ + Id = 'D20L' + Rule = 'It is measured against the window the hub was configured with, not against a number chosen to admit it.' + Test = 'serve::tests::an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client' + File = 'crates/huginn-daemon/src/serve.rs' + Old = ' if encoded <= MAX_RESPONSE_BYTES {' + New = ' if encoded <= MAX_RESPONSE_BYTES * 4 {' + } + @{ + Id = 'D20L2' + Rule = 'The refusal carries the answer''s own encoded size, so a caller can tell how far it overshot.' + Test = 'serve::tests::an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client' + File = 'crates/huginn-daemon/src/serve.rs' + Old = ' bytes: encoded,' + New = ' bytes: MAX_RESPONSE_BYTES,' + } + @{ + Id = 'D20L3' + Rule = 'And it carries the limit itself, not the size again, so the two are not one number twice.' + Test = 'serve::tests::an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client' + File = 'crates/huginn-daemon/src/serve.rs' + Old = ' limit: MAX_RESPONSE_BYTES,' + New = ' limit: encoded,' + } + @{ + Id = 'D21' + Rule = 'The refusal rides the response schema like every other refusal, not a failure envelope the client must parse apart.' + Test = 'serve::tests::an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client' + File = 'crates/huginn-daemon/src/serve.rs' + Old = @' + let refusal = HuginnMindResponse::Refused(MindRefusal::ResponseTooLarge { + bytes: encoded, + limit: MAX_RESPONSE_BYTES, + }); + encode_or_fail(message_id, operation, &refusal, runtime_id) +'@ + New = @' + encode_failure( + message_id, + operation, + &crate::envelope::OperationFailure { + code: "response-too-large".into(), + message: format!("{encoded} bytes against a {MAX_RESPONSE_BYTES} limit"), + }, + runtime_id, + ) +'@ + } + @{ + Id = 'D22' + Rule = 'The window the gate measures against is the window the hub carries: a smaller one sends nothing and says nothing.' + Test = 'serve::tests::an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client' + File = 'crates/huginn-daemon/src/serve.rs' + Old = ' options.max_pending_reliable_packets = Some(MAX_PENDING_RELIABLE_PACKETS);' + New = ' options.max_pending_reliable_packets = Some(MAX_PENDING_RELIABLE_PACKETS / 2);' + } + @{ + Id = 'D22L' + Rule = 'Not merely close to it: a window short by a few dozen packets still drops an answer the gate admitted.' + Test = 'serve::tests::an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client' + File = 'crates/huginn-daemon/src/serve.rs' + Old = ' options.max_pending_reliable_packets = Some(MAX_PENDING_RELIABLE_PACKETS);' + New = ' options.max_pending_reliable_packets = Some(MAX_PENDING_RELIABLE_PACKETS - 64);' + } + @{ + Id = 'D22L2' + Rule = 'The fragment size is the window''s other half: a smaller one carries fewer bytes in the same packet count.' + Test = 'serve::tests::an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client' + File = 'crates/huginn-daemon/src/serve.rs' + Old = ' options.max_fragment_bytes = Some(MAX_FRAGMENT_BYTES);' + New = ' options.max_fragment_bytes = Some(MAX_FRAGMENT_BYTES - 64);' + } @{ Id = 'D17' Rule = 'A refusal `view` raised crosses the dispatch as itself, not rewrapped as unavailable.'