Refuse an answer that will not fit, by name
A reply larger than one send could carry was handed to the hub anyway. The hub
refused it, the loop printed the error and served past, and the client waited
for an answer nothing would ever send. A real-sized read was lost in silence,
which is the one thing this surface promises not to do: a refusal is typed, and
that has to hold at every size.
`within_window` now encodes the reply and measures it against the window this
module configures, before a send it can already see will fail. Over the limit,
the answer becomes `MindRefusal::ResponseTooLarge { bytes, limit }` on the
response schema, reaching the client exactly as every other refusal does, with
the encoded size and the limit so a caller can narrow its own request. Nothing
is truncated, paginated or retried.
The arithmetic, measured rather than estimated. One `PipelineCutSpec` with every
list at the leaf's bound -- 256 file changes, 64 deletes, keeps, adds, builds,
tests, negative checks and operator checks, a full authority map and structural
delta -- is about 980,000 bytes of field content. As a `View` response, base64
in the operation envelope and MessagePack around that, it encodes to 1,315,551
bytes. The window is `MAX_PENDING_RELIABLE_PACKETS` (1024) times
`MAX_FRAGMENT_BYTES` (1200), so 1,228,800. A single document at the leaf's own
bounds therefore does not fit, by 86,751 bytes.
The window's values are unchanged by ruling: what a list read should return is
an open question, and raising the window would bake multi-megabyte answers in
as ordinary traffic before that question is answered. The module documentation
states 1,228,800 as the current bound rather than as a design target.
The test admits two cut specs into a real mind: one at 256 file changes and one
narrowed to 180, whose view encodes to 1,189,495 bytes. Over one loopback
session the query, the wide view and the open items are each refused by name,
and the narrow view is then delivered whole. Both halves are load-bearing: the
delivered answer sits 39,305 bytes under the limit, so `D22` and `D22L`, which
shrink the configured window, break it.
`D20` removes the gate, `D20L` measures against a larger number than the hub
carries, `D20L2` and `D20L3` collapse the two fields the refusal reports, `D21`
sends it as a failure envelope instead, and `D22`, `D22L` and `D22L2` shrink
each half of the window the gate trusts. All eight die. The whole suite, 51
entries and M0, is green.
Two reads pipelined on one session are still not both answered, and the module
documentation says so: 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 though each passes the gate alone. The gate measures one
answer against the window, not against what the session still has room for.
That is not fixed here.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
25a841dc6d
commit
9bbe746719
5 changed files with 503 additions and 32 deletions
|
|
@ -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<PipelineDocument> {
|
||||
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::<Vec<_>>();
|
||||
let shorts = |count: usize| (0..count).map(short).collect::<Vec<_>>();
|
||||
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<OwnedRedbMessagePackBackingStore, NoIndex>) {
|
||||
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.
|
||||
|
|
|
|||
|
|
@ -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<CultNetRudpServerHub> {
|
||||
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<S: MindStore, I: IndexSink<S>>(
|
|||
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<S: MindStore, I: IndexSink<S>>(
|
|||
}
|
||||
}
|
||||
|
||||
/// 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<S: MindStore, I: IndexSink<S>>(
|
|||
#[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<OwnedRedbMessagePackBackingStore, NoIndex>, 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<OwnedRedbMessagePackBackingStore, NoIndex>, 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
|
||||
|
|
|
|||
|
|
@ -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 },
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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": {
|
||||
|
|
|
|||
|
|
@ -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.'
|
||||
|
|
|
|||
Loading…
Reference in a new issue