BP-3 1: an oversize answer is deferred to the body plane
deliver replaces within_window as the one owner of how an answer leaves:
whole when the envelope fits the window, deferred when the payload is within
MAX_DEFERRED_BODY_BYTES (64 MiB), refused by name past it. A deferred answer
is HuginnMindResponse::Deferred(DeferredAnswer { manifest }), the reference's
CultMeshCdnArtifactManifest; the chunks are asked for on the session the client
holds and answered by cultnet-rs's answer_content_chunk_request. The body is
held by DeferredBodies (256 MiB budget, sliding 60 s TTL, least-recently-
touched eviction, shared chunks stored once), in memory only. The mind knows
nothing of deferral.
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
9673ab8c9e
commit
7636737d60
7 changed files with 774 additions and 316 deletions
270
crates/huginn-daemon/src/bodies.rs
Normal file
270
crates/huginn-daemon/src/bodies.rs
Normal file
|
|
@ -0,0 +1,270 @@
|
|||
//! The deferred bodies: answers too large for one send, held for the client to
|
||||
//! fetch chunk by chunk. Ephemeral transport state, never a document store and
|
||||
//! never world truth: it is lost when the process exits, and a client that
|
||||
//! finds a body gone asks the operation again.
|
||||
//!
|
||||
//! Identity is the manifest's `contentHash`, so the same bytes deferred twice
|
||||
//! are one body. Chunking, hashing and the manifest are `cultnet-rs`'s; what is
|
||||
//! owned here is only how long and how much is kept.
|
||||
//!
|
||||
//! A body expires when it has been untouched for the TTL, where retaining it
|
||||
//! again and serving any of its chunks both touch it. When the stored bytes
|
||||
//! exceed the budget the body touched longest ago is evicted first. A chunk
|
||||
//! that two bodies share is stored once and lives until the last of them goes.
|
||||
//! Time is an argument, so nothing here reads a clock.
|
||||
|
||||
use std::cell::Cell;
|
||||
use std::collections::{BTreeSet, HashMap};
|
||||
use std::time::Duration;
|
||||
|
||||
use chrono::{DateTime, Utc};
|
||||
use cultnet_rs::{CultMeshCdnArtifactManifest, CultMeshCdnChunk};
|
||||
|
||||
/// The size of one chunk of a deferred body. Four fit one send with room for a
|
||||
/// control-plane answer beside them on the same session.
|
||||
pub const DEFERRED_CHUNK_BYTES: usize = 256 * 1024;
|
||||
/// The largest answer that is deferred rather than refused, measured on the
|
||||
/// encoded payload.
|
||||
pub const MAX_DEFERRED_BODY_BYTES: u64 = 64 * 1024 * 1024;
|
||||
/// What the daemon keeps across all deferred bodies.
|
||||
pub const DEFERRED_BUDGET_BYTES: u64 = 256 * 1024 * 1024;
|
||||
|
||||
struct Chunk {
|
||||
bytes: Vec<u8>,
|
||||
/// The content hashes of the retained bodies that reference this chunk.
|
||||
owners: BTreeSet<String>,
|
||||
}
|
||||
|
||||
struct Body {
|
||||
chunk_hashes: BTreeSet<String>,
|
||||
touched_at: Cell<DateTime<Utc>>,
|
||||
/// Orders touches when the clock does not: the largest is the newest.
|
||||
touch_order: Cell<u64>,
|
||||
}
|
||||
|
||||
pub struct DeferredBodies {
|
||||
budget: u64,
|
||||
ttl: chrono::Duration,
|
||||
chunks: HashMap<String, Chunk>,
|
||||
bodies: HashMap<String, Body>,
|
||||
stored: u64,
|
||||
touches: Cell<u64>,
|
||||
}
|
||||
|
||||
impl DeferredBodies {
|
||||
pub fn new(budget: u64, ttl: Duration) -> Self {
|
||||
Self {
|
||||
budget,
|
||||
ttl: chrono::Duration::from_std(ttl).unwrap_or(chrono::Duration::MAX),
|
||||
chunks: HashMap::new(),
|
||||
bodies: HashMap::new(),
|
||||
stored: 0,
|
||||
touches: Cell::new(0),
|
||||
}
|
||||
}
|
||||
|
||||
/// Bytes held: each distinct chunk once.
|
||||
#[cfg(test)]
|
||||
pub fn stored_bytes(&self) -> u64 {
|
||||
self.stored
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub fn body_count(&self) -> usize {
|
||||
self.bodies.len()
|
||||
}
|
||||
|
||||
/// Holds a body, or, when its bytes are already held, touches it and
|
||||
/// stores nothing. Then evicts, least recently touched first, while the
|
||||
/// stored bytes exceed the budget and another body could be evicted
|
||||
/// instead: the body just retained is the last to go, and one that alone
|
||||
/// exceeds the budget is kept whole.
|
||||
pub fn retain(&mut self, manifest: &CultMeshCdnArtifactManifest, chunks: Vec<CultMeshCdnChunk>, now: DateTime<Utc>) {
|
||||
let content_hash = manifest.content_hash.clone();
|
||||
if let Some(body) = self.bodies.get(&content_hash) {
|
||||
self.touch(body, now);
|
||||
return;
|
||||
}
|
||||
let mut chunk_hashes = BTreeSet::new();
|
||||
for chunk in chunks {
|
||||
chunk_hashes.insert(chunk.chunk_hash.clone());
|
||||
let held = self.chunks.entry(chunk.chunk_hash).or_insert_with(|| {
|
||||
self.stored += chunk.payload.len() as u64;
|
||||
Chunk { bytes: chunk.payload, owners: BTreeSet::new() }
|
||||
});
|
||||
held.owners.insert(content_hash.clone());
|
||||
}
|
||||
let body = Body { chunk_hashes, touched_at: Cell::new(now), touch_order: Cell::new(0) };
|
||||
self.touch(&body, now);
|
||||
self.bodies.insert(content_hash, body);
|
||||
while self.stored > self.budget && self.bodies.len() > 1 {
|
||||
let oldest = self
|
||||
.bodies
|
||||
.iter()
|
||||
.min_by_key(|(_, body)| body.touch_order.get())
|
||||
.map(|(hash, _)| hash.clone())
|
||||
.expect("more than one body is held");
|
||||
self.drop_body(&oldest);
|
||||
}
|
||||
}
|
||||
|
||||
/// One chunk's bytes by its normalised hash, touching every body that
|
||||
/// references it. Serving a chunk is what keeps its body alive.
|
||||
pub fn chunk(&self, hash: &str, now: DateTime<Utc>) -> Option<&[u8]> {
|
||||
let chunk = self.chunks.get(hash)?;
|
||||
for owner in &chunk.owners {
|
||||
self.touch(&self.bodies[owner], now);
|
||||
}
|
||||
Some(&chunk.bytes)
|
||||
}
|
||||
|
||||
/// Drops every body untouched for the TTL as of `now`. A clock that reads
|
||||
/// earlier than a touch expires nothing.
|
||||
pub fn expire(&mut self, now: DateTime<Utc>) {
|
||||
let expired: Vec<String> = self
|
||||
.bodies
|
||||
.iter()
|
||||
.filter(|(_, body)| now.signed_duration_since(body.touched_at.get()) >= self.ttl)
|
||||
.map(|(hash, _)| hash.clone())
|
||||
.collect();
|
||||
for hash in expired {
|
||||
self.drop_body(&hash);
|
||||
}
|
||||
}
|
||||
|
||||
fn touch(&self, body: &Body, now: DateTime<Utc>) {
|
||||
body.touched_at.set(now);
|
||||
self.touches.set(self.touches.get() + 1);
|
||||
body.touch_order.set(self.touches.get());
|
||||
}
|
||||
|
||||
fn drop_body(&mut self, content_hash: &str) {
|
||||
let Some(body) = self.bodies.remove(content_hash) else { return };
|
||||
for hash in body.chunk_hashes {
|
||||
let chunk = self.chunks.get_mut(&hash).expect("a retained body's chunks are held");
|
||||
chunk.owners.remove(content_hash);
|
||||
if chunk.owners.is_empty() {
|
||||
self.stored -= chunk.bytes.len() as u64;
|
||||
self.chunks.remove(&hash);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use cultnet_rs::pack_content;
|
||||
|
||||
fn at(seconds: i64) -> DateTime<Utc> {
|
||||
DateTime::from_timestamp(1_800_000_000 + seconds, 0).unwrap()
|
||||
}
|
||||
|
||||
/// A body of `pieces` chunks of 4 bytes each, distinct per `tag`.
|
||||
fn body(tag: u8, pieces: usize) -> (CultMeshCdnArtifactManifest, Vec<CultMeshCdnChunk>) {
|
||||
let mut bytes = Vec::new();
|
||||
for piece in 0..pieces {
|
||||
bytes.extend_from_slice(&[tag, piece as u8, 7, 7]);
|
||||
}
|
||||
pack_content("t", "package", "", "", "", &bytes, 4).unwrap()
|
||||
}
|
||||
|
||||
fn held(store: &DeferredBodies, manifest: &CultMeshCdnArtifactManifest, now: DateTime<Utc>) -> bool {
|
||||
manifest.chunks.iter().all(|chunk| store.chunk(&chunk.chunk_hash, now).is_some())
|
||||
}
|
||||
|
||||
const TTL: Duration = Duration::from_secs(60);
|
||||
|
||||
/// The same bytes deferred twice are one body, counted once; serving a
|
||||
/// chunk slides the expiry; an untouched body is gone exactly at the TTL.
|
||||
#[test]
|
||||
fn retention_is_idempotent_and_sliding() {
|
||||
let mut store = DeferredBodies::new(1_000, TTL);
|
||||
let (manifest, chunks) = body(1, 3);
|
||||
store.retain(&manifest, chunks.clone(), at(0));
|
||||
store.retain(&manifest, chunks, at(1));
|
||||
assert_eq!((store.body_count(), store.stored_bytes()), (1, 12), "one body, its bytes counted once");
|
||||
|
||||
// Retaining again at t=1 touched it, so it lives to t=61, not t=60.
|
||||
store.expire(at(60));
|
||||
assert_eq!(store.body_count(), 1, "the second retain refreshed it");
|
||||
// Serving a chunk at t=60 touches it, so t=119 is inside a fresh TTL.
|
||||
assert!(store.chunk(&manifest.chunks[0].chunk_hash, at(60)).is_some());
|
||||
store.expire(at(119));
|
||||
assert_eq!(store.body_count(), 1, "serving a chunk slid the expiry");
|
||||
store.expire(at(120));
|
||||
assert_eq!((store.body_count(), store.stored_bytes()), (0, 0), "untouched for the TTL is gone");
|
||||
|
||||
// An untouched body is gone at the TTL and not before it.
|
||||
let (other, chunks) = body(2, 1);
|
||||
store.retain(&other, chunks, at(200));
|
||||
store.expire(at(259));
|
||||
assert_eq!(store.body_count(), 1);
|
||||
store.expire(at(260));
|
||||
assert_eq!(store.body_count(), 0);
|
||||
}
|
||||
|
||||
/// Time going backwards expires nothing.
|
||||
#[test]
|
||||
fn a_clock_behind_the_last_touch_expires_nothing() {
|
||||
let mut store = DeferredBodies::new(1_000, TTL);
|
||||
let (manifest, chunks) = body(1, 1);
|
||||
store.retain(&manifest, chunks, at(100));
|
||||
store.expire(at(0));
|
||||
assert_eq!(store.body_count(), 1);
|
||||
}
|
||||
|
||||
/// Over the budget the body touched longest ago goes first, not the
|
||||
/// newest and not the first inserted, and a chunk two bodies share stays
|
||||
/// while either lives.
|
||||
#[test]
|
||||
fn eviction_is_least_recently_touched_and_bounded_in_bytes() {
|
||||
// Budget 12 bytes: three chunks of 4.
|
||||
let mut store = DeferredBodies::new(12, TTL);
|
||||
let (a, chunks_a) = body(1, 1);
|
||||
let (b, chunks_b) = body(2, 1);
|
||||
let (c, chunks_c) = body(3, 1);
|
||||
store.retain(&a, chunks_a, at(0));
|
||||
store.retain(&b, chunks_b, at(1));
|
||||
// The oldest-inserted body is touched last, so insertion order and
|
||||
// recency disagree about who goes.
|
||||
assert!(store.chunk(&a.chunks[0].chunk_hash, at(2)).is_some());
|
||||
store.retain(&c, chunks_c, at(3));
|
||||
assert_eq!(store.stored_bytes(), 12);
|
||||
let (d, chunks_d) = body(4, 1);
|
||||
store.retain(&d, chunks_d, at(4));
|
||||
assert_eq!(store.stored_bytes(), 12, "bounded by the budget");
|
||||
assert!(!held(&store, &b, at(5)), "the least recently touched went");
|
||||
assert!(held(&store, &a, at(5)) && held(&store, &c, at(5)) && held(&store, &d, at(5)));
|
||||
|
||||
// A shared chunk survives its first owner.
|
||||
let mut store = DeferredBodies::new(12, TTL);
|
||||
let (x, chunks_x) = body(5, 2); // chunks x0 x1
|
||||
store.retain(&x, chunks_x, at(0));
|
||||
let (y, mut chunks_y) = body(6, 1); // chunk y0, plus x1 shared
|
||||
let mut manifest_y = y.clone();
|
||||
manifest_y.content_hash = "shared-owner".into();
|
||||
chunks_y.push(CultMeshCdnChunk {
|
||||
chunk_hash: x.chunks[1].chunk_hash.clone(),
|
||||
payload: vec![5, 1, 7, 7],
|
||||
});
|
||||
store.retain(&manifest_y, chunks_y, at(1));
|
||||
assert_eq!(store.stored_bytes(), 12, "the shared chunk is stored once");
|
||||
// Evict x (older): its private chunk goes, the shared one stays.
|
||||
let (z, chunks_z) = body(7, 1);
|
||||
store.retain(&z, chunks_z, at(2));
|
||||
assert_eq!(store.body_count(), 2, "x was evicted");
|
||||
assert!(store.chunk(&x.chunks[0].chunk_hash, at(3)).is_none(), "x's private chunk is gone");
|
||||
assert!(store.chunk(&x.chunks[1].chunk_hash, at(3)).is_some(), "the shared chunk lives with y");
|
||||
}
|
||||
|
||||
/// A body that alone exceeds the budget is kept whole rather than
|
||||
/// evicting itself, so a manifest handed out is never a lie.
|
||||
#[test]
|
||||
fn a_body_over_the_budget_by_itself_is_kept() {
|
||||
let mut store = DeferredBodies::new(4, TTL);
|
||||
let (manifest, chunks) = body(1, 3);
|
||||
store.retain(&manifest, chunks, at(0));
|
||||
assert!(held(&store, &manifest, at(0)));
|
||||
}
|
||||
}
|
||||
|
|
@ -6,12 +6,15 @@
|
|||
//! request": it dispatches, it asks `Mind::require_instance` for every read
|
||||
//! that names an instance, and it compares nothing itself. `envelope` owns "is
|
||||
//! this a request at all": a message it cannot decode is answered with a typed
|
||||
//! failure and touches no mind. `serve` owns "which session gets which reply".
|
||||
//! failure and touches no mind. `serve` owns "which session gets which reply"
|
||||
//! and whether an answer is delivered whole, deferred or refused; `bodies`
|
||||
//! owns how long and how much of a deferred answer is held.
|
||||
//!
|
||||
//! Every admission rule, the instance check, the receipt, the derived status
|
||||
//! and the wire vocabulary are `huginn-mind`'s. The leaf and the store type
|
||||
//! are reached through `huginn_mind`, so one crate pins one revision of each.
|
||||
|
||||
pub mod bodies;
|
||||
pub mod daemon;
|
||||
pub mod envelope;
|
||||
pub mod serve;
|
||||
|
|
|
|||
|
|
@ -7,14 +7,23 @@
|
|||
//! spins with logging rather than exiting, and Cut 14's health check is the
|
||||
//! observer.
|
||||
//!
|
||||
//! One answer may be `MAX_RESPONSE_BYTES` encoded, 1,228,800 today: the
|
||||
//! One send carries `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.
|
||||
//! of `MAX_FRAGMENT_BYTES`. `deliver` decides every answer's plane: one that
|
||||
//! fits goes whole; a larger one whose encoded payload is at most
|
||||
//! `MAX_DEFERRED_BODY_BYTES` is held by `DeferredBodies` and answered with
|
||||
//! `HuginnMindResponse::Deferred`, a `CultMeshCdnArtifactManifest`, and the
|
||||
//! client fetches the payload with `cultnet_rs::fetch_content`, one
|
||||
//! `cultmesh.content_chunk_request.v1` at a time on the session it already
|
||||
//! holds, exactly as the C# reference's `CultMeshLegacyRudpContentServer` is
|
||||
//! asked. The fetched bytes are the named MessagePack of the
|
||||
//! `HuginnMindResponse` the mind gave. A body the daemon will not defer is
|
||||
//! refused by name, `MindRefusal::ResponseTooLarge`, carrying the payload size
|
||||
//! and `MAX_DEFERRED_BODY_BYTES`, and the caller narrows its own request. A
|
||||
//! deferred body is memory only: it expires untouched after
|
||||
//! `ServeOptions::deferred_ttl`, is evicted past `DEFERRED_BUDGET_BYTES`, and
|
||||
//! is gone when the process exits, so a chunk request that finds it gone is
|
||||
//! answered `found: false` and the client asks the operation again.
|
||||
//!
|
||||
//! Four things this cut does not do, which a client has nowhere else to learn:
|
||||
//!
|
||||
|
|
@ -50,31 +59,43 @@ use std::sync::atomic::{AtomicBool, Ordering};
|
|||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context, Result, ensure};
|
||||
use base64::Engine;
|
||||
use base64::engine::general_purpose::STANDARD;
|
||||
use chrono::{DateTime, Utc};
|
||||
use cultnet_rs::{
|
||||
CULTNET_OPERATION_CONNECTION_ID, CultNetMessage, CultNetRudpServerEvent, CultNetRudpServerHub,
|
||||
CultNetRudpServerHubOptions, CultNetSchemaKind, CultNetSchemaRegistration, CultNetSchemaRegistry,
|
||||
CultNetWireContract, decode_cultnet_message_from_slice, encode_cultnet_message_to_vec,
|
||||
CultNetWireContract, answer_content_chunk_request, decode_cultnet_message_from_slice,
|
||||
encode_cultnet_message_to_vec, pack_content,
|
||||
};
|
||||
use huginn_mind::epiphany_pipeline::Slug;
|
||||
use huginn_mind::wire::{
|
||||
HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA,
|
||||
DeferredAnswer, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA,
|
||||
MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID,
|
||||
};
|
||||
use huginn_mind::{MindRefusal, MindStore, OwnedRedbMessagePackBackingStore};
|
||||
|
||||
use crate::bodies::{
|
||||
DEFERRED_BUDGET_BYTES, DEFERRED_CHUNK_BYTES, DeferredBodies, MAX_DEFERRED_BODY_BYTES,
|
||||
};
|
||||
use crate::daemon::{Daemon, IndexSink, NoIndex};
|
||||
use crate::envelope::{decode_request, encode_failure, encode_response};
|
||||
use crate::envelope::{OperationFailure, decode_request, encode_failure, encode_response};
|
||||
|
||||
/// What the loop does when nothing is waiting.
|
||||
/// What the loop does when nothing is waiting. A deferred body untouched for
|
||||
/// `deferred_ttl`, twice the session timeout, is dropped.
|
||||
pub struct ServeOptions {
|
||||
pub session_timeout: Duration,
|
||||
pub idle_sleep: Duration,
|
||||
pub deferred_ttl: Duration,
|
||||
}
|
||||
|
||||
impl Default for ServeOptions {
|
||||
fn default() -> Self {
|
||||
Self { session_timeout: Duration::from_secs(30), idle_sleep: Duration::from_millis(2) }
|
||||
Self {
|
||||
session_timeout: Duration::from_secs(30),
|
||||
idle_sleep: Duration::from_millis(2),
|
||||
deferred_ttl: Duration::from_secs(60),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -164,20 +185,24 @@ pub fn schema_registry() -> Result<CultNetSchemaRegistry> {
|
|||
|
||||
/// One message in, one message out. Nothing here reaches a mind except through
|
||||
/// `Daemon::handle`, and an envelope that does not decode never gets that far.
|
||||
/// A chunk request is answered from `bodies` alone, on the session it came in
|
||||
/// on; every operation's reply leaves through `deliver`.
|
||||
pub fn answer<S: MindStore, I: IndexSink<S>>(
|
||||
daemon: &mut Daemon<S, I>,
|
||||
registry: &CultNetSchemaRegistry,
|
||||
bodies: &mut DeferredBodies,
|
||||
message: CultNetMessage,
|
||||
now: DateTime<Utc>,
|
||||
) -> CultNetMessage {
|
||||
let runtime_id = daemon.runtime_id();
|
||||
bodies.expire(now);
|
||||
match &message {
|
||||
CultNetMessage::OperationRequest { message_id, operation, .. } => match decode_request(&message) {
|
||||
Ok((message_id, request)) => {
|
||||
let operation = request.operation();
|
||||
let response = daemon.handle(request, now);
|
||||
let reply = encode_or_fail(&message_id, operation, &response, &runtime_id);
|
||||
within_window(reply, &message_id, operation, &runtime_id)
|
||||
deliver(reply, &message_id, operation, &runtime_id, bodies, now)
|
||||
}
|
||||
Err(failure) => encode_failure(message_id, operation, &failure, &runtime_id),
|
||||
},
|
||||
|
|
@ -185,9 +210,13 @@ pub fn answer<S: MindStore, I: IndexSink<S>>(
|
|||
Ok(response) => response,
|
||||
Err(error) => CultNetMessage::Error { error: format!("{error:#}"), code: None, details: None },
|
||||
},
|
||||
CultNetMessage::ContentChunkRequest { .. } => {
|
||||
answer_content_chunk_request(&message, |hash| bodies.chunk(hash, now))
|
||||
}
|
||||
_ => CultNetMessage::Error {
|
||||
error: format!(
|
||||
"{MIND_SERVICE_ID} answers cultnet.operation_request.v0 and cultnet.schema_catalog_request.v0"
|
||||
"{MIND_SERVICE_ID} answers cultnet.operation_request.v0, cultnet.schema_catalog_request.v0 \
|
||||
and cultmesh.content_chunk_request.v1"
|
||||
),
|
||||
code: None,
|
||||
details: None,
|
||||
|
|
@ -216,40 +245,73 @@ fn encode_or_fail(
|
|||
}
|
||||
}
|
||||
|
||||
/// 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.
|
||||
/// How one answer leaves: whole, deferred or refused, in that order, and
|
||||
/// nothing else decides. 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 envelope is therefore measured here, against
|
||||
/// the window this module configured, before a send that can already be seen to
|
||||
/// fail is attempted.
|
||||
///
|
||||
/// 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 {
|
||||
/// A reply that fits is returned as it is. A larger one whose payload, the
|
||||
/// named MessagePack of the response, is at most `MAX_DEFERRED_BODY_BYTES` is
|
||||
/// packed by CultNet's content plane, retained in `bodies`, and answered with
|
||||
/// `Deferred` and the manifest. That answer is a small fixed-shape envelope and
|
||||
/// is never itself measured or deferred. A larger payload is refused by name
|
||||
/// with its own size and the deferral bound: nothing is truncated or paginated,
|
||||
/// and the caller narrows its own request. Both answers ride the response
|
||||
/// schema like every other, so a client parses no second vocabulary.
|
||||
fn deliver(
|
||||
reply: CultNetMessage,
|
||||
message_id: &str,
|
||||
operation: &str,
|
||||
runtime_id: &str,
|
||||
bodies: &mut DeferredBodies,
|
||||
now: DateTime<Utc>,
|
||||
) -> CultNetMessage {
|
||||
let not_encodable = |message: String| {
|
||||
encode_failure(
|
||||
message_id,
|
||||
operation,
|
||||
&OperationFailure { code: "response-not-encodable".into(), message },
|
||||
runtime_id,
|
||||
)
|
||||
};
|
||||
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,
|
||||
);
|
||||
}
|
||||
Err(error) => return not_encodable(format!("{error:#}")),
|
||||
};
|
||||
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)
|
||||
let payload = match &reply {
|
||||
CultNetMessage::OperationResponse { payload, .. } => STANDARD.decode(payload),
|
||||
_ => return not_encodable("only an operation response is deferred".into()),
|
||||
};
|
||||
let payload = match payload {
|
||||
Ok(bytes) => bytes,
|
||||
Err(error) => return not_encodable(format!("the payload is not base64: {error}")),
|
||||
};
|
||||
let size = payload.len() as u64;
|
||||
if size > MAX_DEFERRED_BODY_BYTES {
|
||||
let refusal = HuginnMindResponse::Refused(MindRefusal::ResponseTooLarge {
|
||||
bytes: size,
|
||||
limit: MAX_DEFERRED_BODY_BYTES,
|
||||
});
|
||||
return encode_or_fail(message_id, operation, &refusal, runtime_id);
|
||||
}
|
||||
let (manifest, chunks) = pack_content(
|
||||
"huginn.mind_response",
|
||||
"package",
|
||||
"",
|
||||
"application/vnd.gamecult.huginn.mind-response+msgpack",
|
||||
&now.to_rfc3339(),
|
||||
&payload,
|
||||
DEFERRED_CHUNK_BYTES,
|
||||
)
|
||||
.expect("the deferral constants are a valid pack");
|
||||
bodies.retain(&manifest, chunks, now);
|
||||
encode_or_fail(message_id, operation, &HuginnMindResponse::Deferred(DeferredAnswer { manifest }), runtime_id)
|
||||
}
|
||||
|
||||
/// Until `stopping`: expire what timed out, resend what was not acknowledged,
|
||||
|
|
@ -264,6 +326,7 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
|
|||
options: &ServeOptions,
|
||||
) -> Result<()> {
|
||||
let timeout_ms = options.session_timeout.as_millis() as u64;
|
||||
let mut bodies = DeferredBodies::new(DEFERRED_BUDGET_BYTES, options.deferred_ttl);
|
||||
while !stopping.load(Ordering::Relaxed) {
|
||||
hub.remove_timed_out_sessions(timeout_ms);
|
||||
if let Err(error) = hub.poll_resends() {
|
||||
|
|
@ -293,7 +356,7 @@ pub fn run<S: MindStore, I: IndexSink<S>>(
|
|||
continue;
|
||||
}
|
||||
};
|
||||
let reply = answer(daemon, registry, message, Utc::now());
|
||||
let reply = answer(daemon, registry, &mut bodies, message, Utc::now());
|
||||
if let Err(error) = hub.send_schema_message(&session, &reply) {
|
||||
eprintln!("huginn: {} did not receive its reply: {error:#}", session.remote_addr);
|
||||
}
|
||||
|
|
@ -314,9 +377,7 @@ mod tests {
|
|||
};
|
||||
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;
|
||||
use cultnet_rs::{CultMesh, CultMeshRudpSocketOptions};
|
||||
use cultnet_rs::{CultMesh, CultMeshCdnArtifactManifest, CultMeshRudpSocketOptions, fetch_content};
|
||||
use huginn_mind::wire::{HuginnMindRequest, HuginnMindResponse, MindStatus};
|
||||
use cultnet_rs::{FieldPredicate, Selection};
|
||||
use huginn_mind::{MindRefusal, PipelineAdmissionOutcome, PipelinePageItems};
|
||||
|
|
@ -391,14 +452,14 @@ mod tests {
|
|||
("m-5b", request("m-5b", MIND_SERVICE_ID, "Admit", MIND_REQUEST_SCHEMA, admit), "operation-mismatch"),
|
||||
];
|
||||
for (id, message, code) in cases {
|
||||
let reply = answer(&mut daemon, ®istry, message, now());
|
||||
let reply = answer(&mut daemon, ®istry, &mut fresh_bodies(), message, now());
|
||||
assert_eq!(rejected(&reply).0, id, "the client's own correlation key is echoed");
|
||||
assert_eq!(rejected(&reply).1, code);
|
||||
}
|
||||
// The sixth code, `not-an-operation-request`, is the codec's and is
|
||||
// pinned in `envelope::tests`: `answer` never produces it, because a
|
||||
// message of another family is answered with `Error` instead.
|
||||
let reply = answer(&mut daemon, ®istry, CultNetMessage::Error { error: "hello".into(), code: None, details: None }, now());
|
||||
let reply = answer(&mut daemon, ®istry, &mut fresh_bodies(), CultNetMessage::Error { error: "hello".into(), code: None, details: None }, now());
|
||||
let CultNetMessage::Error { error, .. } = &reply else { panic!("expected an error, got {reply:?}") };
|
||||
assert!(
|
||||
!error.is_empty() && error.contains("cultnet.operation_request.v0") && error.contains("cultnet.schema_catalog_request.v0"),
|
||||
|
|
@ -410,7 +471,7 @@ mod tests {
|
|||
// A mind's refusal is an answer on the response schema, not a failure
|
||||
// of the envelope: the client decodes it as the typed refusal it is.
|
||||
let read = HuginnMindRequest::Query { instance: slug(OTHER), selection: Selection::default(), semantic: None };
|
||||
let reply = answer(&mut daemon, ®istry, encode_request("m-r", &read, None).unwrap(), now());
|
||||
let reply = answer(&mut daemon, ®istry, &mut fresh_bodies(), encode_request("m-r", &read, None).unwrap(), now());
|
||||
let CultNetMessage::OperationResponse { status, payload_schema, .. } = &reply else {
|
||||
panic!("expected an operation response, got {reply:?}");
|
||||
};
|
||||
|
|
@ -428,7 +489,7 @@ mod tests {
|
|||
schema_ids: None,
|
||||
kinds: None,
|
||||
};
|
||||
let reply = answer(&mut daemon, ®istry, catalog, now());
|
||||
let reply = answer(&mut daemon, ®istry, &mut fresh_bodies(), catalog, now());
|
||||
let CultNetMessage::SchemaCatalogResponse { message_id, schemas } = reply else {
|
||||
panic!("expected a catalog response");
|
||||
};
|
||||
|
|
@ -532,29 +593,109 @@ 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.
|
||||
fn encoded_len(message: &CultNetMessage) -> u64 {
|
||||
encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0).unwrap().len() as u64
|
||||
}
|
||||
|
||||
const TTL: Duration = Duration::from_secs(60);
|
||||
|
||||
fn fresh_bodies() -> DeferredBodies {
|
||||
DeferredBodies::new(DEFERRED_BUDGET_BYTES, TTL)
|
||||
}
|
||||
|
||||
type Client = cultnet_rs::CultNetRudpSocketTransportConnection;
|
||||
type BodyDaemon = Daemon<OwnedRedbMessagePackBackingStore, NoIndex>;
|
||||
|
||||
/// One request on the client's own session and its reply, the way
|
||||
/// `fetch_content` is asked for each chunk.
|
||||
fn exchange(client: &mut Client, message: &CultNetMessage) -> Result<CultNetMessage> {
|
||||
client.send_schema_message(message)?;
|
||||
for _ in 0..3000 {
|
||||
client.poll_resends()?;
|
||||
if let Some(reply) = client.receive_schema_message_once()? {
|
||||
return Ok(reply);
|
||||
}
|
||||
std::thread::sleep(Duration::from_millis(2));
|
||||
}
|
||||
anyhow::bail!("no reply")
|
||||
}
|
||||
|
||||
fn deferred_manifest(reply: &CultNetMessage) -> CultMeshCdnArtifactManifest {
|
||||
match decode_response(reply).unwrap().1.expect("an answer, not an envelope failure") {
|
||||
HuginnMindResponse::Deferred(DeferredAnswer { manifest }) => manifest,
|
||||
other => panic!("expected a deferred answer, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
fn chunk_request(manifest: &CultMeshCdnArtifactManifest, index: usize) -> CultNetMessage {
|
||||
let chunk = &manifest.chunks[index];
|
||||
CultNetMessage::ContentChunkRequest {
|
||||
message_id: format!("c-{index}"),
|
||||
chunk_hash: chunk.chunk_hash.clone(),
|
||||
record_key: chunk.record_key.clone(),
|
||||
expected_size_bytes: chunk.size_bytes,
|
||||
}
|
||||
}
|
||||
|
||||
fn found(reply: &CultNetMessage) -> bool {
|
||||
let CultNetMessage::ContentChunkResponse { found, .. } = reply else {
|
||||
panic!("expected a content chunk response, got {reply:?}");
|
||||
};
|
||||
*found
|
||||
}
|
||||
|
||||
fn document_query() -> HuginnMindRequest {
|
||||
HuginnMindRequest::Query {
|
||||
instance: slug(INSTANCE),
|
||||
selection: Selection { projection: "document".into(), ..Selection::default() },
|
||||
semantic: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// One operation asked of `answer` at an injected time, and the manifest of
|
||||
/// the deferred answer it gives back.
|
||||
fn ask_deferred(
|
||||
daemon: &mut BodyDaemon,
|
||||
bodies: &mut DeferredBodies,
|
||||
message_id: &str,
|
||||
read: &HuginnMindRequest,
|
||||
at: DateTime<Utc>,
|
||||
) -> CultMeshCdnArtifactManifest {
|
||||
let registry = schema_registry().unwrap();
|
||||
deferred_manifest(&answer(daemon, ®istry, bodies, encode_request(message_id, read, None).unwrap(), at))
|
||||
}
|
||||
|
||||
/// Whether the first chunk of a manifest is still served at an injected time.
|
||||
fn first_chunk_found(
|
||||
daemon: &mut BodyDaemon,
|
||||
bodies: &mut DeferredBodies,
|
||||
manifest: &CultMeshCdnArtifactManifest,
|
||||
at: DateTime<Utc>,
|
||||
) -> bool {
|
||||
let registry = schema_registry().unwrap();
|
||||
found(&answer(daemon, ®istry, bodies, chunk_request(manifest, 0), at))
|
||||
}
|
||||
|
||||
/// An answer larger than one send can carry is deferred to the body plane
|
||||
/// and fetches, on the same session, to the answer the mind gave. 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
|
||||
/// envelope 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.
|
||||
/// The client holds the `Deferred` answer before it fetches, so the size
|
||||
/// the daemon decided on is observed at the decision and not only
|
||||
/// downstream of the fetch.
|
||||
#[test]
|
||||
fn an_answer_too_large_for_one_send_is_a_typed_refusal_that_reaches_the_client() {
|
||||
fn an_oversize_answer_arrives_deferred_and_fetches_to_the_same_answer() {
|
||||
let (_root, mut daemon) = seeded_wide();
|
||||
let registry = schema_registry().unwrap();
|
||||
|
||||
let sized = |daemon: &mut Daemon<OwnedRedbMessagePackBackingStore, NoIndex>, id: PipelineRef| {
|
||||
let sized = |daemon: &mut BodyDaemon, 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
|
||||
encoded_len(&message)
|
||||
};
|
||||
let wide = spec_ref(&mut daemon, WIDE_CUT);
|
||||
let fitting = spec_ref(&mut daemon, FITTING_CUT);
|
||||
|
|
@ -567,6 +708,13 @@ mod tests {
|
|||
"the delivered answer must sit close enough to the limit that the window's value is load-bearing"
|
||||
);
|
||||
|
||||
let reads = [
|
||||
("m-q", document_query()),
|
||||
("m-v", HuginnMindRequest::View { instance: slug(INSTANCE), id: wide }),
|
||||
];
|
||||
let expected: Vec<HuginnMindResponse> =
|
||||
reads.iter().map(|(_, read)| daemon.handle(read.clone(), now())).collect();
|
||||
|
||||
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(
|
||||
|
|
@ -587,20 +735,6 @@ mod tests {
|
|||
}
|
||||
assert!(client.connected());
|
||||
|
||||
let reads = [
|
||||
// The whole documents: a header page is small by construction, so
|
||||
// the transport's bound is measured on the projection that can
|
||||
// exceed it.
|
||||
(
|
||||
"m-q",
|
||||
HuginnMindRequest::Query {
|
||||
instance: slug(INSTANCE),
|
||||
selection: Selection { projection: "document".into(), ..Selection::default() },
|
||||
semantic: None,
|
||||
},
|
||||
),
|
||||
("m-v", HuginnMindRequest::View { instance: slug(INSTANCE), id: wide }),
|
||||
];
|
||||
let stopping = Arc::new(AtomicBool::new(false));
|
||||
let loop_stopping = Arc::clone(&stopping);
|
||||
let serving = std::thread::spawn(move || {
|
||||
|
|
@ -609,50 +743,30 @@ mod tests {
|
|||
});
|
||||
|
||||
// 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");
|
||||
// still unacknowledged, so the fetch is one chunk in flight.
|
||||
for ((message_id, read), expected) in reads.into_iter().zip(expected) {
|
||||
let reply = exchange(&mut client, &encode_request(message_id, &read, None).unwrap()).unwrap();
|
||||
assert!(encoded_len(&reply) <= MAX_RESPONSE_BYTES, "the deferred answer fits one send");
|
||||
assert_eq!(decode_response(&reply).unwrap().0, message_id);
|
||||
let manifest = deferred_manifest(&reply);
|
||||
let bytes_expected = rmp_serde::to_vec_named(&expected).unwrap();
|
||||
assert_eq!(manifest.size_bytes as usize, bytes_expected.len(), "{message_id}: the manifest is of the answer");
|
||||
|
||||
let bytes = fetch_content(&manifest, MAX_DEFERRED_BODY_BYTES, |request| exchange(&mut client, &request))
|
||||
.unwrap();
|
||||
assert_eq!(bytes, bytes_expected, "{message_id}: byte-identical to the answer the mind gave");
|
||||
assert_eq!(rmp_serde::from_slice::<HuginnMindResponse>(&bytes).unwrap(), expected);
|
||||
assert!(!matches!(expected, HuginnMindResponse::Deferred(_)));
|
||||
}
|
||||
|
||||
// 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.
|
||||
// session that just fetched two bodies.
|
||||
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 reply = exchange(&mut client, &encode_request(message_id, &request, None).unwrap()).unwrap();
|
||||
let (correlation, answered) = decode_response(&reply).unwrap();
|
||||
assert_eq!(&correlation, message_id);
|
||||
match answered.unwrap() {
|
||||
|
|
@ -666,69 +780,6 @@ mod tests {
|
|||
serving.join().unwrap().0.unwrap();
|
||||
}
|
||||
|
||||
fn encoded_len(message: &CultNetMessage) -> u64 {
|
||||
encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0).unwrap().len() as u64
|
||||
}
|
||||
|
||||
/// A reply carrying `payload_len` bytes of payload under a chosen
|
||||
/// `message_id`: the envelope a view answer travels in, with the document
|
||||
/// replaced by filler, so a size can be chosen instead of found. The
|
||||
/// `message_id` is part of the encoded envelope too, so pinning the gate
|
||||
/// against one length alone (every other fixture uses `m-0`) hides a gate
|
||||
/// that hard-codes that length's contribution instead of measuring it.
|
||||
fn synthetic_with_id(message_id: &str, payload_len: usize) -> CultNetMessage {
|
||||
synthetic_full(message_id, "view", "huginn-yggdrasil", payload_len)
|
||||
}
|
||||
|
||||
/// N4: `operation` and `source_runtime_id` are in the encoded envelope
|
||||
/// too, exactly as `message_id` is, so a boundary fixture that only ever
|
||||
/// varies `message_id` cannot tell a gate that genuinely measures the
|
||||
/// encoded bytes from one that hard-codes a formula shaped to fit that
|
||||
/// one dimension. `synthetic_with_id` is this with the operation and
|
||||
/// runtime id `the_gates_boundary_*` tests were pinned to before N4.
|
||||
fn synthetic_full(message_id: &str, operation: &str, runtime_id: &str, payload_len: usize) -> CultNetMessage {
|
||||
CultNetMessage::OperationResponse {
|
||||
message_id: message_id.into(),
|
||||
service_id: MIND_SERVICE_ID.into(),
|
||||
operation: operation.into(),
|
||||
status: "accepted".into(),
|
||||
payload_schema: MIND_RESPONSE_SCHEMA.into(),
|
||||
payload_encoding: "messagepack-base64".into(),
|
||||
payload: "A".repeat(payload_len),
|
||||
diagnostics: vec![],
|
||||
source_runtime_id: Some(runtime_id.into()),
|
||||
}
|
||||
}
|
||||
|
||||
fn synthetic(payload_len: usize) -> CultNetMessage {
|
||||
synthetic_with_id("m-0", payload_len)
|
||||
}
|
||||
|
||||
/// One whose encoded envelope is exactly `target` bytes, under a chosen
|
||||
/// `message_id`. The envelope's own overhead is measured rather than
|
||||
/// assumed: the length prefix is the same width on either side of a
|
||||
/// megabyte, so one probe fixes it.
|
||||
fn sized_exactly_with_id(message_id: &str, target: u64) -> CultNetMessage {
|
||||
sized_exactly_full(message_id, "view", "huginn-yggdrasil", target)
|
||||
}
|
||||
|
||||
/// N4's own probe: `sized_exactly_with_id` generalised over the
|
||||
/// operation and the runtime id as well as the message id, so a fixture
|
||||
/// can pin the gate's boundary at a combination a formula shaped to fit
|
||||
/// `view`/`huginn-yggdrasil` alone was never measured against.
|
||||
fn sized_exactly_full(message_id: &str, operation: &str, runtime_id: &str, target: u64) -> CultNetMessage {
|
||||
let probe = 1_200_000_usize;
|
||||
let at_probe = encoded_len(&synthetic_full(message_id, operation, runtime_id, probe));
|
||||
let message =
|
||||
synthetic_full(message_id, operation, runtime_id, (probe as i64 + (target as i64 - at_probe as i64)) as usize);
|
||||
assert_eq!(encoded_len(&message), target);
|
||||
message
|
||||
}
|
||||
|
||||
fn sized_exactly(target: u64) -> CultNetMessage {
|
||||
sized_exactly_with_id("m-0", target)
|
||||
}
|
||||
|
||||
/// A connected session whose accept has been acknowledged, so its window
|
||||
/// holds `MAX_PENDING_RELIABLE_PACKETS` and not one less. See the module
|
||||
/// documentation: a test that measures the window exactly and does not
|
||||
|
|
@ -769,150 +820,224 @@ mod tests {
|
|||
(client, session.unwrap())
|
||||
}
|
||||
|
||||
/// The gate's boundary is the transport's, byte for byte. The test above
|
||||
/// proves the two agree at the sizes real documents reach, which leaves
|
||||
/// the width of a whole document between the largest answer it delivers
|
||||
/// and the smallest it refuses; this one leaves nothing. A reply whose
|
||||
/// encoded envelope is exactly `MAX_RESPONSE_BYTES` passes the gate
|
||||
/// unchanged and is accepted by a hub `bind` configured; one byte more is
|
||||
/// refused by both.
|
||||
///
|
||||
/// So the gate measures the encoded envelope, which is what the hub
|
||||
/// fragments, and not the payload string inside it, which is 231 bytes
|
||||
/// smaller; and it admits the limit rather than stopping one short of it.
|
||||
/// Neither is visible to a test whose answers sit tens of thousands of
|
||||
/// bytes from the boundary.
|
||||
///
|
||||
/// The refusal is measured on the same scale: it is one packet, so the
|
||||
/// answer that says an answer did not fit always fits itself. And the
|
||||
/// accepted reply is shown to have filled the window it was measured
|
||||
/// against, because a one-byte reply after it is refused.
|
||||
#[test]
|
||||
fn the_gates_boundary_is_the_transports_boundary_byte_for_byte() {
|
||||
let at = sized_exactly(MAX_RESPONSE_BYTES);
|
||||
let over = sized_exactly(MAX_RESPONSE_BYTES + 1);
|
||||
assert_eq!(within_window(at.clone(), "m-0", "view", "huginn-yggdrasil"), at, "exactly the limit passes");
|
||||
/// A reply carrying a chosen payload string, with every field the envelope
|
||||
/// echoes chosen too, so a size can be chosen instead of found.
|
||||
fn reply_with_payload(message_id: &str, operation: &str, runtime_id: &str, payload: String) -> CultNetMessage {
|
||||
CultNetMessage::OperationResponse {
|
||||
message_id: message_id.into(),
|
||||
service_id: MIND_SERVICE_ID.into(),
|
||||
operation: operation.into(),
|
||||
status: "accepted".into(),
|
||||
payload_schema: MIND_RESPONSE_SCHEMA.into(),
|
||||
payload_encoding: "messagepack-base64".into(),
|
||||
payload,
|
||||
diagnostics: vec![],
|
||||
source_runtime_id: Some(runtime_id.into()),
|
||||
}
|
||||
}
|
||||
|
||||
let refused = within_window(over.clone(), "m-0", "view", "huginn-yggdrasil");
|
||||
/// One whose encoded envelope is exactly `target` bytes and whose payload is
|
||||
/// valid base64, under a `message_id` that is `stem` followed by as many
|
||||
/// zeros as make the payload a whole number of base64 quanta. The envelope's
|
||||
/// own overhead is measured, not assumed: the length prefix is the same
|
||||
/// width on either side of a megabyte, so one probe fixes it. The id is
|
||||
/// echoed, so it is part of the result.
|
||||
fn sized_exactly(stem: &str, operation: &str, runtime_id: &str, target: u64) -> (String, CultNetMessage) {
|
||||
let probe = 1_200_000_usize;
|
||||
for zeros in 0..4 {
|
||||
let message_id = format!("{stem}{}", "0".repeat(zeros));
|
||||
let at_probe = encoded_len(&reply_with_payload(&message_id, operation, runtime_id, "A".repeat(probe)));
|
||||
let payload_len = (probe as i64 + (target as i64 - at_probe as i64)) as usize;
|
||||
if payload_len % 4 == 0 {
|
||||
let message = reply_with_payload(&message_id, operation, runtime_id, "A".repeat(payload_len));
|
||||
assert_eq!(encoded_len(&message), target);
|
||||
return (message_id, message);
|
||||
}
|
||||
}
|
||||
unreachable!("one of four consecutive id lengths aligns the payload")
|
||||
}
|
||||
|
||||
/// The boundary is the transport's, byte for byte, at every shape of
|
||||
/// envelope. A reply whose encoded envelope is exactly `MAX_RESPONSE_BYTES`
|
||||
/// is delivered unchanged and a hub `bind` configured accepts it; one byte
|
||||
/// more is deferred, not refused, and a hub refuses it.
|
||||
///
|
||||
/// So the threshold measures the encoded envelope, which is what the hub
|
||||
/// fragments, and not the payload string inside it, which is a couple of
|
||||
/// hundred bytes smaller; it admits the limit rather than stopping one
|
||||
/// short of it; and it is the window itself and not a margin under it. The
|
||||
/// echoed ids, operations and runtime ids vary independently and in length,
|
||||
/// so a formula fitted to one envelope shape is measured where it was never
|
||||
/// fitted. The accepted reply is shown to have filled the window it was
|
||||
/// measured against, because a one-byte reply after it is refused.
|
||||
#[test]
|
||||
fn the_delivery_boundary_is_the_transports_boundary_byte_for_byte() {
|
||||
let long = format!("m-{}", "1".repeat(38));
|
||||
for (stem, operation, runtime_id) in [
|
||||
("m-", "view", "huginn-yggdrasil"),
|
||||
(long.as_str(), "view", "huginn-yggdrasil"),
|
||||
("m-", "whoami", "huginn-thought-cage"),
|
||||
("m-", "whoami", "huginn-yggdrasil"),
|
||||
("m-", "view", "huginn-thought-cage"),
|
||||
] {
|
||||
let (id_at, at) = sized_exactly(stem, operation, runtime_id, MAX_RESPONSE_BYTES);
|
||||
let (id_over, over) = sized_exactly(stem, operation, runtime_id, MAX_RESPONSE_BYTES + 1);
|
||||
let mut bodies = fresh_bodies();
|
||||
let case = format!("{operation}/{runtime_id}/{}", stem.len());
|
||||
|
||||
let delivered = deliver(at.clone(), &id_at, operation, runtime_id, &mut bodies, now());
|
||||
assert_eq!(delivered, at, "{case}: exactly the limit is delivered whole");
|
||||
assert_eq!(bodies.body_count(), 0, "{case}: nothing was retained for it");
|
||||
|
||||
let deferred = deliver(over.clone(), &id_over, operation, runtime_id, &mut bodies, now());
|
||||
let (correlation, answered) = decode_response(&deferred).unwrap();
|
||||
assert_eq!(correlation, id_over, "{case}");
|
||||
let Some(HuginnMindResponse::Deferred(DeferredAnswer { manifest })) = answered.ok() else {
|
||||
panic!("{case}: one byte over the limit was not deferred: {deferred:?}");
|
||||
};
|
||||
assert_eq!(bodies.body_count(), 1, "{case}");
|
||||
assert!(
|
||||
encoded_len(&deferred) <= MAX_FRAGMENT_BYTES as u64 * 4,
|
||||
"{case}: the deferred answer is a few packets"
|
||||
);
|
||||
let CultNetMessage::OperationResponse { payload, .. } = &over else { unreachable!() };
|
||||
assert_eq!(manifest.size_bytes as usize, STANDARD.decode(payload).unwrap().len(), "{case}");
|
||||
|
||||
if stem == "m-" && operation == "view" && runtime_id == "huginn-yggdrasil" {
|
||||
let mut hub = bind("127.0.0.1:0".parse().unwrap(), "huginn-yggdrasil").unwrap();
|
||||
let (_client, session) = settled_session(&mut hub);
|
||||
let error = hub.send_schema_message(&session, &over).unwrap_err();
|
||||
assert!(format!("{error:#}").contains("queue is full"), "limit + 1 was accepted: {error:#}");
|
||||
hub.send_schema_message(&session, &at).expect("exactly the limit is accepted on an empty window");
|
||||
hub.send_schema_message(&session, &reply_with_payload("m-0", "view", runtime_id, "A".into()))
|
||||
.expect_err("the accepted answer filled the window, so nothing follows it");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The backstop stays, measured against the deferral bound and on the
|
||||
/// payload: a body of exactly `MAX_DEFERRED_BODY_BYTES` is deferred, one
|
||||
/// byte more is refused by name with its own size and that bound, and the
|
||||
/// deferred answer at the bound still fits one send, so it is never itself
|
||||
/// deferred.
|
||||
#[test]
|
||||
fn an_answer_over_the_deferral_bound_is_refused_by_name() {
|
||||
let payload_of = |len: u64| STANDARD.encode(vec![0_u8; len as usize]);
|
||||
let mut bodies = fresh_bodies();
|
||||
|
||||
let at = reply_with_payload("m-b", "view", "huginn-yggdrasil", payload_of(MAX_DEFERRED_BODY_BYTES));
|
||||
let deferred = deliver(at, "m-b", "view", "huginn-yggdrasil", &mut bodies, now());
|
||||
let manifest = deferred_manifest(&deferred);
|
||||
assert_eq!(manifest.size_bytes as u64, MAX_DEFERRED_BODY_BYTES);
|
||||
assert_eq!(manifest.chunks.len() as u64, MAX_DEFERRED_BODY_BYTES / DEFERRED_CHUNK_BYTES as u64);
|
||||
assert!(encoded_len(&deferred) <= MAX_RESPONSE_BYTES, "at the bound the deferred answer still fits one send");
|
||||
assert_eq!(bodies.body_count(), 1);
|
||||
drop(bodies);
|
||||
|
||||
let over = reply_with_payload("m-b", "view", "huginn-yggdrasil", payload_of(MAX_DEFERRED_BODY_BYTES + 1));
|
||||
let mut empty = fresh_bodies();
|
||||
let refused = deliver(over, "m-b", "view", "huginn-yggdrasil", &mut empty, now());
|
||||
let (correlation, answered) = decode_response(&refused).unwrap();
|
||||
assert_eq!(correlation, "m-0");
|
||||
assert_eq!(correlation, "m-b");
|
||||
let HuginnMindResponse::Refused(MindRefusal::ResponseTooLarge { bytes, limit }) =
|
||||
answered.expect("a refusal is an answer, not an envelope failure")
|
||||
else {
|
||||
panic!("one byte over the limit was not refused: {refused:?}");
|
||||
panic!("one byte over the deferral bound was not refused: {refused:?}");
|
||||
};
|
||||
assert_eq!((bytes, limit), (MAX_RESPONSE_BYTES + 1, MAX_RESPONSE_BYTES));
|
||||
let refusal_bytes = encoded_len(&refused);
|
||||
assert_eq!((bytes, limit), (MAX_DEFERRED_BODY_BYTES + 1, MAX_DEFERRED_BODY_BYTES));
|
||||
assert!(encoded_len(&refused) <= MAX_FRAGMENT_BYTES as u64, "the refusal is one packet");
|
||||
assert_eq!(empty.body_count(), 0, "a refused answer retains nothing");
|
||||
}
|
||||
|
||||
/// An oversize reply whose payload cannot be read as the response's bytes
|
||||
/// is a failure of the envelope, named, and never deferred.
|
||||
#[test]
|
||||
fn an_oversize_reply_that_is_not_a_response_is_not_deferred() {
|
||||
let mut bodies = fresh_bodies();
|
||||
let not_base64 = reply_with_payload("m-x", "view", "huginn-yggdrasil", "A".repeat(MAX_RESPONSE_BYTES as usize) + "!");
|
||||
let not_a_response =
|
||||
CultNetMessage::Error { error: "e".repeat(MAX_RESPONSE_BYTES as usize + 1), code: None, details: None };
|
||||
for reply in [not_base64, not_a_response] {
|
||||
assert!(encoded_len(&reply) > MAX_RESPONSE_BYTES);
|
||||
let answered = deliver(reply, "m-x", "view", "huginn-yggdrasil", &mut bodies, now());
|
||||
assert_eq!(rejected(&answered).1, "response-not-encodable");
|
||||
}
|
||||
assert_eq!(bodies.body_count(), 0);
|
||||
}
|
||||
|
||||
/// Lifecycle end to end through the daemon's own `answer`, on an injected
|
||||
/// clock: a chunk served inside the TTL slides the expiry, a body untouched
|
||||
/// for the TTL is gone and its chunk is answered `found: false` by name,
|
||||
/// and the client asks again and fetches the same answer.
|
||||
#[test]
|
||||
fn an_expired_body_answers_found_false_and_the_client_can_ask_again() {
|
||||
let (_root, mut daemon) = seeded_wide();
|
||||
let registry = schema_registry().unwrap();
|
||||
let wide = spec_ref(&mut daemon, WIDE_CUT);
|
||||
let read = HuginnMindRequest::View { instance: slug(INSTANCE), id: wide };
|
||||
let expected = rmp_serde::to_vec_named(&daemon.handle(read.clone(), now())).unwrap();
|
||||
let t0 = now();
|
||||
let after = |seconds: i64| t0 + chrono::Duration::seconds(seconds);
|
||||
let mut bodies = fresh_bodies();
|
||||
|
||||
let manifest = ask_deferred(&mut daemon, &mut bodies, "m-v", &read, t0);
|
||||
assert!(manifest.chunks.len() >= 2, "the fixture needs a second chunk to serve");
|
||||
let mut chunk = |index: usize, at: i64| {
|
||||
answer(&mut daemon, ®istry, &mut bodies, chunk_request(&manifest, index), after(at))
|
||||
};
|
||||
assert!(found(&chunk(0, 59)), "inside the TTL");
|
||||
assert!(found(&chunk(1, 118)), "serving a chunk slid the expiry past the first TTL");
|
||||
let gone = chunk(0, 178);
|
||||
assert!(!found(&gone), "untouched for the TTL since t=118");
|
||||
let CultNetMessage::ContentChunkResponse { error, payload, .. } = &gone else { unreachable!() };
|
||||
assert!(error.starts_with("FileNotFoundException"), "{error}");
|
||||
assert!(payload.is_empty());
|
||||
|
||||
let again = ask_deferred(&mut daemon, &mut bodies, "m-v2", &read, after(179));
|
||||
assert_eq!(again.content_hash, manifest.content_hash, "the same answer is the same body");
|
||||
let bytes = fetch_content(&again, MAX_DEFERRED_BODY_BYTES, |request| {
|
||||
Ok(answer(&mut daemon, ®istry, &mut bodies, request, after(180)))
|
||||
})
|
||||
.unwrap();
|
||||
assert_eq!(bytes, expected);
|
||||
}
|
||||
|
||||
/// Past the budget the body touched longest ago is evicted, its chunks
|
||||
/// answer `found: false`, and asking again evicts the other in turn. Two
|
||||
/// identical answers are one body.
|
||||
#[test]
|
||||
fn eviction_and_sharing_through_the_daemons_answer() {
|
||||
let (_root, mut daemon) = seeded_wide();
|
||||
let wide = spec_ref(&mut daemon, WIDE_CUT);
|
||||
let view = HuginnMindRequest::View { instance: slug(INSTANCE), id: wide };
|
||||
let query = document_query();
|
||||
let size = |daemon: &mut BodyDaemon, read: &HuginnMindRequest| {
|
||||
rmp_serde::to_vec_named(&daemon.handle(read.clone(), now())).unwrap().len() as u64
|
||||
};
|
||||
let (view_bytes, query_bytes) = (size(&mut daemon, &view), size(&mut daemon, &query));
|
||||
// Room for the larger body alone, and not for both.
|
||||
let budget = view_bytes.max(query_bytes) + 1;
|
||||
let t0 = now();
|
||||
let after = |seconds: i64| t0 + chrono::Duration::seconds(seconds);
|
||||
let mut bodies = DeferredBodies::new(budget, TTL);
|
||||
|
||||
let first = ask_deferred(&mut daemon, &mut bodies, "m-1", &view, after(0));
|
||||
let same = ask_deferred(&mut daemon, &mut bodies, "m-2", &view, after(1));
|
||||
assert_eq!(first.content_hash, same.content_hash, "two identical answers are one body");
|
||||
assert_eq!((bodies.body_count(), bodies.stored_bytes()), (1, view_bytes));
|
||||
|
||||
let second = ask_deferred(&mut daemon, &mut bodies, "m-3", &query, after(2));
|
||||
assert_eq!(bodies.body_count(), 1, "the least recently touched body was evicted");
|
||||
assert!(bodies.stored_bytes() <= budget);
|
||||
assert!(!first_chunk_found(&mut daemon, &mut bodies, &first, after(3)), "the evicted body answers found: false");
|
||||
assert!(first_chunk_found(&mut daemon, &mut bodies, &second, after(3)), "the newer body stays");
|
||||
|
||||
let again = ask_deferred(&mut daemon, &mut bodies, "m-4", &view, after(4));
|
||||
assert!(first_chunk_found(&mut daemon, &mut bodies, &again, after(5)));
|
||||
assert!(
|
||||
refusal_bytes <= MAX_FRAGMENT_BYTES as u64,
|
||||
"the refusal is {refusal_bytes} bytes and must fit one packet"
|
||||
!first_chunk_found(&mut daemon, &mut bodies, &second, after(5)),
|
||||
"the client's re-ask evicted the one touched longest ago"
|
||||
);
|
||||
|
||||
let mut hub = bind("127.0.0.1:0".parse().unwrap(), "huginn-yggdrasil").unwrap();
|
||||
let (_client, session) = settled_session(&mut hub);
|
||||
let error = hub.send_schema_message(&session, &over).unwrap_err();
|
||||
assert!(format!("{error:#}").contains("queue is full"), "limit + 1 was accepted: {error:#}");
|
||||
hub.send_schema_message(&session, &at).expect("exactly the limit is accepted on an empty window");
|
||||
hub.send_schema_message(&session, &synthetic(1))
|
||||
.expect_err("the accepted answer filled the window, so nothing follows it");
|
||||
}
|
||||
|
||||
/// The boundary above is pinned for one envelope shape: every reply in it
|
||||
/// carries `message_id` `m-0`, three bytes, so a gate that measures
|
||||
/// `payload.len() + 231` — exactly the encoded envelope's overhead for
|
||||
/// that one id — passes it unnoticed. The reply echoes the client's own
|
||||
/// `message_id`, so that overhead is under the client's control: at a
|
||||
/// longer id, such a gate is wrong, and wrong on the side that lets an
|
||||
/// oversize answer through the size check to a send that then fails
|
||||
/// silently. Forty bytes is used because it is far from `m-0`'s length in
|
||||
/// either direction, not chosen to sit near some other boundary.
|
||||
#[test]
|
||||
fn the_gates_boundary_holds_for_a_second_message_id_length() {
|
||||
let long_id = format!("m-{}", "0".repeat(38));
|
||||
assert_eq!(long_id.len(), 40, "a message_id far from m-0's own length");
|
||||
|
||||
let at = sized_exactly_with_id(&long_id, MAX_RESPONSE_BYTES);
|
||||
let over = sized_exactly_with_id(&long_id, MAX_RESPONSE_BYTES + 1);
|
||||
assert_eq!(
|
||||
within_window(at.clone(), &long_id, "view", "huginn-yggdrasil"),
|
||||
at,
|
||||
"exactly the limit passes at this id's length too"
|
||||
);
|
||||
|
||||
let refused = within_window(over.clone(), &long_id, "view", "huginn-yggdrasil");
|
||||
let (correlation, answered) = decode_response(&refused).unwrap();
|
||||
assert_eq!(correlation, long_id);
|
||||
let HuginnMindResponse::Refused(MindRefusal::ResponseTooLarge { bytes, limit }) =
|
||||
answered.expect("a refusal is an answer, not an envelope failure")
|
||||
else {
|
||||
panic!("one byte over the limit was not refused: {refused:?}");
|
||||
};
|
||||
assert_eq!((bytes, limit), (MAX_RESPONSE_BYTES + 1, MAX_RESPONSE_BYTES));
|
||||
}
|
||||
|
||||
/// N4: the boundary above is pinned at one `message_id` length, but
|
||||
/// `operation` and `source_runtime_id` are in the encoded envelope the
|
||||
/// same way `message_id` is, and every fixture before this one held both
|
||||
/// fixed at `"view"` and `"huginn-yggdrasil"`. A gate whose measurement is
|
||||
/// a formula of `message_id`'s length alone -- shaped to pass at exactly
|
||||
/// `m-0`'s three bytes and the other fixture's forty -- has no way to be
|
||||
/// wrong there and every way to be wrong once the operation or the
|
||||
/// runtime id it never looked at changes length too. This holds the
|
||||
/// boundary at `whoami` (six bytes, not `view`'s four) over a second
|
||||
/// runtime id, so such a formula is measured somewhere it was never
|
||||
/// fitted.
|
||||
#[test]
|
||||
fn the_gates_boundary_holds_across_operation_and_runtime_id() {
|
||||
for (operation, runtime_id) in [("view", "huginn-yggdrasil"), ("whoami", "huginn-thought-cage")] {
|
||||
let at = sized_exactly_full("m-g", operation, runtime_id, MAX_RESPONSE_BYTES);
|
||||
let over = sized_exactly_full("m-g", operation, runtime_id, MAX_RESPONSE_BYTES + 1);
|
||||
assert_eq!(
|
||||
within_window(at.clone(), "m-g", operation, runtime_id),
|
||||
at,
|
||||
"{operation}/{runtime_id}: exactly the limit passes"
|
||||
);
|
||||
|
||||
let refused = within_window(over.clone(), "m-g", operation, runtime_id);
|
||||
let (correlation, answered) = decode_response(&refused).unwrap();
|
||||
assert_eq!(correlation, "m-g");
|
||||
let HuginnMindResponse::Refused(MindRefusal::ResponseTooLarge { bytes, limit }) =
|
||||
answered.expect("a refusal is an answer, not an envelope failure")
|
||||
else {
|
||||
panic!("{operation}/{runtime_id}: one byte over the limit was not refused: {refused:?}");
|
||||
};
|
||||
assert_eq!((bytes, limit), (MAX_RESPONSE_BYTES + 1, MAX_RESPONSE_BYTES), "{operation}/{runtime_id}");
|
||||
}
|
||||
}
|
||||
|
||||
/// Every fixture up to this one, including the pair above,
|
||||
/// moves `operation` and the runtime id together -- both short (`view`,
|
||||
/// 4 bytes / `huginn-yggdrasil`, 16) or both long (`whoami`, 6 /
|
||||
/// `huginn-thought-cage`, 19) -- so a formula that tracks only one of the
|
||||
/// two and ignores the other still lands on the right answer at both
|
||||
/// points by coincidence. This pins a point where they move apart: a long
|
||||
/// operation with the short runtime id, and a short operation with the
|
||||
/// long one, so a gate that dropped either term is wrong at one of them.
|
||||
#[test]
|
||||
fn the_gates_boundary_holds_when_operation_and_runtime_id_vary_independently() {
|
||||
for (operation, runtime_id) in [("whoami", "huginn-yggdrasil"), ("view", "huginn-thought-cage")] {
|
||||
let at = sized_exactly_full("m-g", operation, runtime_id, MAX_RESPONSE_BYTES);
|
||||
let over = sized_exactly_full("m-g", operation, runtime_id, MAX_RESPONSE_BYTES + 1);
|
||||
assert_eq!(
|
||||
within_window(at.clone(), "m-g", operation, runtime_id),
|
||||
at,
|
||||
"{operation}/{runtime_id}: exactly the limit passes"
|
||||
);
|
||||
|
||||
let refused = within_window(over.clone(), "m-g", operation, runtime_id);
|
||||
let (correlation, answered) = decode_response(&refused).unwrap();
|
||||
assert_eq!(correlation, "m-g");
|
||||
let HuginnMindResponse::Refused(MindRefusal::ResponseTooLarge { bytes, limit }) =
|
||||
answered.expect("a refusal is an answer, not an envelope failure")
|
||||
else {
|
||||
panic!("{operation}/{runtime_id}: one byte over the limit was not refused: {refused:?}");
|
||||
};
|
||||
assert_eq!((bytes, limit), (MAX_RESPONSE_BYTES + 1, MAX_RESPONSE_BYTES), "{operation}/{runtime_id}");
|
||||
}
|
||||
}
|
||||
|
||||
/// One cut spec's own reference, by its cut label.
|
||||
|
|
|
|||
|
|
@ -48,7 +48,7 @@ pub use receipt::{DocumentVersion, Faculty, HuginnCommitReceipt, PipelineProvena
|
|||
pub use refusal::MindRefusal;
|
||||
pub use store::{MindStore, OwnedRedbMessagePackBackingStore};
|
||||
pub use wire::{
|
||||
HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA,
|
||||
DeferredAnswer, HuginnMindRequest, HuginnMindResponse, MIND_REQUEST_SCHEMA, MIND_REQUEST_SCHEMA_JSON, MIND_RESPONSE_SCHEMA,
|
||||
MIND_RESPONSE_SCHEMA_JSON, MIND_SERVICE_ID, MindStatus,
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -55,14 +55,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.
|
||||
/// The one refusal a mind never raises: the answer exceeds the largest
|
||||
/// body the organ will deliver by any plane, one send or a deferred body.
|
||||
/// 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
|
||||
/// (the payload, not its envelope) and `limit` is the largest body the
|
||||
/// daemon defers, both so a caller can narrow its own request; nothing is
|
||||
/// truncated or paginated on its behalf. The daemon owns it and its
|
||||
/// deferral bound decides it; no rule of the mind is involved.
|
||||
ResponseTooLarge { bytes: u64, limit: u64 },
|
||||
/// A selection the substrate or the organ's value door refused: `field`
|
||||
/// is the selection field the client sent, `value` the offending part.
|
||||
|
|
|
|||
|
|
@ -5,7 +5,7 @@
|
|||
//! names a transport.
|
||||
|
||||
use cultcache_rs::DatabaseEntry;
|
||||
use cultnet_rs::Selection;
|
||||
use cultnet_rs::{CultMeshCdnArtifactManifest, Selection};
|
||||
use epiphany_pipeline::{PIPELINE_SCHEMA_EPOCH, PipelineKind, PipelineRef, Slug};
|
||||
use schemars::JsonSchema;
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
|
@ -41,6 +41,18 @@ fn selection_ref(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
|
|||
})
|
||||
}
|
||||
|
||||
/// The deferred answer's manifest is CultMesh's own type, a positional array
|
||||
/// the C# reference defines (`CultMeshCdnArtifactManifest`, schema version
|
||||
/// `gamecult.mesh.cdn_artifact_manifest.v1`). CultLib publishes no JSON schema
|
||||
/// for it, so the schema says what the reference says: an array of its ten
|
||||
/// fields in order, and nothing here copies their meaning.
|
||||
fn manifest_array(_: &mut schemars::SchemaGenerator) -> schemars::Schema {
|
||||
schemars::json_schema!({
|
||||
"description": "gamecult.mesh.cdn_artifact_manifest.v1, the C# reference's CultMeshCdnArtifactManifest as its positional array: [artifactId, kind, version, contentHash, sizeBytes, mimeType, createdAtUtc, chunks[[chunkHash, offset, sizeBytes, recordKey]], tags, metadata].",
|
||||
"type": "array"
|
||||
})
|
||||
}
|
||||
|
||||
/// Every operation a mind answers. `Admit` carries the batch whole because the
|
||||
/// batch already names the instance and the asker; a second `instance` beside
|
||||
/// it would be two declarations. Every read names the instance because ruling
|
||||
|
|
@ -91,6 +103,22 @@ pub enum HuginnMindResponse {
|
|||
View(Option<PipelineDocumentView>),
|
||||
Query(PipelineSelectionPage),
|
||||
Refused(MindRefusal),
|
||||
/// The answer did not fit one send and is being held for the client to
|
||||
/// fetch. Nothing in a mind produces this: the daemon's serve module does,
|
||||
/// in place of an answer, and the answer it stands for is the named
|
||||
/// MessagePack of the `HuginnMindResponse` the mind gave.
|
||||
Deferred(DeferredAnswer),
|
||||
}
|
||||
|
||||
/// Where the body of an answer that did not fit one send can be fetched: the
|
||||
/// reference's own manifest, whose `contentHash` is the SHA-256 of the answer's
|
||||
/// encoded bytes and whose chunks are asked for, one at a time, on the session
|
||||
/// this arrived on. Row-level facts come from asking again with a header
|
||||
/// projection; there is no summary.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
|
||||
pub struct DeferredAnswer {
|
||||
#[schemars(schema_with = "manifest_array")]
|
||||
pub manifest: CultMeshCdnArtifactManifest,
|
||||
}
|
||||
|
||||
impl HuginnMindResponse {
|
||||
|
|
@ -176,6 +204,11 @@ mod tests {
|
|||
edges: None,
|
||||
}),
|
||||
HuginnMindResponse::Refused(refusal),
|
||||
HuginnMindResponse::Deferred(DeferredAnswer {
|
||||
manifest: cultnet_rs::pack_content("huginn.mind_response", "package", "", "", "", b"body", 4)
|
||||
.unwrap()
|
||||
.0,
|
||||
}),
|
||||
]
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -69,6 +69,19 @@
|
|||
"required": [
|
||||
"Refused"
|
||||
]
|
||||
},
|
||||
{
|
||||
"description": "The answer did not fit one send and is being held for the client to\nfetch. Nothing in a mind produces this: the daemon's serve module does,\nin place of an answer, and the answer it stands for is the named\nMessagePack of the `HuginnMindResponse` the mind gave.",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"Deferred": {
|
||||
"$ref": "#/$defs/DeferredAnswer"
|
||||
}
|
||||
},
|
||||
"additionalProperties": false,
|
||||
"required": [
|
||||
"Deferred"
|
||||
]
|
||||
}
|
||||
],
|
||||
"$defs": {
|
||||
|
|
@ -287,6 +300,19 @@
|
|||
"type": "string",
|
||||
"maxLength": 10
|
||||
},
|
||||
"DeferredAnswer": {
|
||||
"description": "Where the body of an answer that did not fit one send can be fetched: the\nreference's own manifest, whose `contentHash` is the SHA-256 of the answer's\nencoded bytes and whose chunks are asked for, one at a time, on the session\nthis arrived on. Row-level facts come from asking again with a header\nprojection; there is no summary.",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"manifest": {
|
||||
"description": "gamecult.mesh.cdn_artifact_manifest.v1, the C# reference's CultMeshCdnArtifactManifest as its positional array: [artifactId, kind, version, contentHash, sizeBytes, mimeType, createdAtUtc, chunks[[chunkHash, offset, sizeBytes, recordKey]], tags, metadata].",
|
||||
"type": "array"
|
||||
}
|
||||
},
|
||||
"required": [
|
||||
"manifest"
|
||||
]
|
||||
},
|
||||
"Deviation": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
|
|
@ -1202,7 +1228,7 @@
|
|||
]
|
||||
},
|
||||
{
|
||||
"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.",
|
||||
"description": "The one refusal a mind never raises: the answer exceeds the largest\nbody the organ will deliver by any plane, one send or a deferred body.\nIt lives here because a refusal rides the response schema and there is\none of those, so a client reads it the way it reads every other refusal\nrather than parsing a second vocabulary. `bytes` is the encoded answer\n(the payload, not its envelope) and `limit` is the largest body the\ndaemon defers, both so a caller can narrow its own request; nothing is\ntruncated or paginated on its behalf. The daemon owns it and its\ndeferral bound decides it; no rule of the mind is involved.",
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"ResponseTooLarge": {
|
||||
|
|
|
|||
Loading…
Reference in a new issue