Merge pull request 'Desk door reads: Forks and Fork on the peer-checked socket (whole record), unique order rank, the leftover Lows' (#38) from eureka-substrate/desk-door-reads-r2 into main

This commit is contained in:
Eureka 2026-11-11 00:46:59 +00:00
commit 2662a34a41
15 changed files with 3295 additions and 196 deletions

View file

@ -119,9 +119,6 @@ pub fn runtime_id(instance: &Slug) -> String {
format!("huginn-{}", instance.0)
}
/// What a desk read is told until follow-up desk-door-reads builds the reads and deletes this.
const DESK_READS_UNBUILT: &str = "desk reads not built";
/// One mind and whatever indexes what lands in it.
pub struct Daemon<S: MindStore, I: IndexSink<S>> {
mind: Mind<S>,
@ -238,16 +235,18 @@ impl<S: MindStore, I: IndexSink<S>> Daemon<S, I> {
/// One desk request, answered by the Mind method it names. The Mind is the instance's own
/// (`mind` is checked against it); `deciders` is this call's list, handed to the Mind and
/// dropped. A desk answer is not indexed (`index_text` gives it no text), so nothing here tells the index.
/// The reads wait on follow-up desk-door-reads and are refused until they exist.
/// The reads answer with the whole record and nothing here gates them (ruling
/// `desk-read-visibility-whole-record`): they are handed `deciders` and not `subject`.
pub fn handle_desk(&mut self, request: DeskRequest, now: DateTime<Utc>) -> DeskResponse {
if let Err(refusal) = self.mind.require_instance(request.mind()) {
return DeskResponse::Refused(refusal);
}
match request {
DeskRequest::Answer { subject, authority, deciders, request, .. } => {
self.mind.desk_answer(&deciders, subject, authority, request, now).0
self.mind.desk_answer(&deciders, subject, authority, request, now)
}
DeskRequest::Forks { .. } | DeskRequest::Fork { .. } => DeskResponse::Refused(MindRefusal::Unavailable { detail: DESK_READS_UNBUILT.into() }),
DeskRequest::Forks { deciders, status, cursor, limit, .. } => self.mind.desk_forks(&deciders, status, cursor, limit),
DeskRequest::Fork { deciders, fork_id, .. } => self.mind.desk_fork(&deciders, &fork_id),
}
}

View file

@ -9,6 +9,17 @@
//! read and write spends what is left of it, so a peer that trickles bytes is cut off at the
//! deadline like one that is silent), a reply past `DESK_MAX_RESPONSE_BYTES` refused by name.
//!
//! The trust boundary of the reads (ruling `desk-read-visibility-whole-record`). The peer-uid check
//! IS the whole authentication: Huginn must be reachable only by Bifrost's peer-checked socket, so
//! the configured uid is Bifrost's and no other process's. `Forks` and `Fork` return the WHOLE record
//! (cards, outcomes, every decider's note) to whoever passes that check; nothing in Huginn strips a
//! note or gates a read on `subject` or `deciders` (`deciders` only fills the card and the
//! showability gate). Bifrost applies visibility at its own door (a stripped read serves `note: null`,
//! CultLib `fork-card.md`). A second reader of this socket is therefore a note leak, and a deployment
//! that gives the socket to any uid but Bifrost's has no reader check left. The reads carry no
//! `authority` (a payload that names one is refused whole), and the door has no other way to ask who
//! is reading.
//!
//! It owns no rule. Which Mind method answers is `Daemon::handle_desk`'s; when the door is
//! served is `serve::run`'s, once per pass in the thread that owns the Mind. A peer that is
//! slow costs the loop at most `DESK_ACCEPTS_PER_PASS` connections of `DESK_CONNECTION_DEADLINE`

View file

@ -20,7 +20,8 @@ use cultnet_rs::{
decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, encode_frame,
};
use huginn_mind::desk_wire::{
DESK_REQUEST_SCHEMA, DESK_SERVICE_ID, DeskRequest, DeskResponse, ForkAnswer, ForkAnswerCode, ForkAnswerRequest, RequestSchema,
DESK_MAX_RESPONSE_BYTES, DESK_REQUEST_SCHEMA, DESK_SERVICE_ID, DeskRequest, DeskResponse, ForkAnswer, ForkAnswerCode, ForkAnswerRequest, ForkCursor,
ForkEntry, ForkLimit, ForkList, ListStatus, RequestSchema,
};
use huginn_mind::envelope::{OperationFailure, decode_desk_response, decode_response, encode_desk_request, encode_request};
use huginn_mind::eureka_pipeline::{DeskAuthority, PipelineDocument, Short};
@ -92,6 +93,45 @@ fn answer(who: &str, authority: DeskAuthority, deciders: &[&str], request: ForkA
DeskRequest::Answer { mind: slug(INSTANCE), subject: short(who), authority, deciders: deciders.iter().map(|d| short(d)).collect(), request }
}
/// A list read by `who` against `deciders`.
fn forks(who: &str, deciders: &[&str], status: Option<ListStatus>, cursor: Option<ForkCursor>, limit: u8) -> DeskRequest {
DeskRequest::Forks {
mind: slug(INSTANCE),
subject: short(who),
deciders: deciders.iter().map(|d| short(d)).collect(),
status,
cursor,
limit: ForkLimit::new(limit).unwrap(),
}
}
fn forks_all(deciders: &[&str]) -> DeskRequest {
forks("x", deciders, None, None, 50)
}
fn fork_read(who: &str, deciders: &[&str], fork_id: &str) -> DeskRequest {
DeskRequest::Fork { mind: slug(INSTANCE), subject: short(who), deciders: deciders.iter().map(|d| short(d)).collect(), fork_id: fork_id.into() }
}
fn page_of(response: DeskResponse) -> (ForkList, u64) {
match response {
DeskResponse::Forks { list, as_of } => (list, as_of),
other => panic!("expected a page, got {other:?}"),
}
}
fn entry_of(response: DeskResponse) -> (ForkEntry, u64) {
match response {
DeskResponse::Fork { entry, as_of } => (entry, as_of),
other => panic!("expected an entry, got {other:?}"),
}
}
/// A plain answer to the seeded fork, with a note.
fn noted(hash: &str, request_id: &str, note: &str) -> ForkAnswerRequest {
ForkAnswerRequest { answer: Some(ForkAnswer { choice: Some("A".into()), note: Some(note.into()) }), ..wire(hash, request_id) }
}
fn message(id: &str, request: &DeskRequest) -> CultNetMessage {
encode_desk_request(id, request, None).unwrap()
}
@ -750,18 +790,205 @@ mod door_smoke {
}
#[test]
fn forks_and_fork_are_refused_unavailable_while_reads_are_unbuilt() {
fn forks_and_fork_read_the_stored_fork_and_store_nothing() {
let mut rig = rig(my_uid());
let (before, hash) = (rig.state(), rig.hash.clone());
let (forks, fork) = rig.serving(|path| (reply_of(path, "m-1", &forks_all(&["x", "y"])), reply_of(path, "m-2", &fork_read("x", &["x", "y"], FORK))));
let (list, listed_at) = page_of(forks);
let (entry, read_at) = entry_of(fork);
assert_eq!((list.open_count, list.recorded_count, list.forks.len(), list.next_cursor), (1, 0, 1, None));
assert_eq!(list.mind, INSTANCE);
assert_eq!((listed_at, read_at), (before.1 as u64, before.1 as u64), "both read at the head the mind held");
assert_eq!(list.stamp, format!("s{listed_at}"));
assert_eq!(list.forks[0], entry);
assert_eq!((entry.card.fork_id.as_str(), entry.card.options_hash.as_str(), entry.card.deciders.as_slice()), (FORK, hash.as_str(), ["x".to_string(), "y".to_string()].as_slice()));
assert_eq!(rig.state(), before, "a read stores nothing");
assert!(rig.index.seen.lock().unwrap().is_empty());
// An unknown fork is the answer path's refusal.
let unknown = rig.serving(|path| reply_of(path, "m-3", &fork_read("x", &["x"], "eureka-state:question:Q404")));
assert!(matches!(unknown, DeskResponse::NotStored(ref refusal) if refusal.code == huginn_mind::desk_wire::ForkRefusalCode::ForkUnknown), "{unknown:?}");
// The mind is the instance's own.
let mut foreign = forks_all(&["x"]);
let DeskRequest::Forks { mind, .. } = &mut foreign else { unreachable!() };
*mind = slug(OTHER);
let refused = rig.serving(|path| reply_of(path, "m-4", &foreign));
assert_eq!(refused, DeskResponse::Refused(MindRefusal::ForeignInstance { declared: OTHER.into(), mind: INSTANCE.into() }));
assert_eq!(rig.state(), before);
}
/// The daemon hands a list's `status` and `cursor` to the Mind, whole: a recorded-only list is the
/// one closed fork, an open-only list the open ones, and a cursor page is the forks below it; read
/// both in the daemon and over the socket.
#[test]
fn the_status_and_the_cursor_reach_the_mind() {
let mut rig = rig(my_uid());
let (question, ruling) = question_and_ruling();
let PipelineDocument::Question(held) = &question else { unreachable!() };
let more: Vec<PipelineDocument> = (2..=3)
.map(|n| {
let mut next = held.clone();
next.label = format!("Q{n}").as_str().into();
PipelineDocument::Question(next)
})
.collect();
for documents in [more, vec![ruling]] {
let admitted = rig.daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents)), now()).answered();
assert!(matches!(admitted, HuginnMindResponse::Admit(huginn_mind::PipelineAdmissionOutcome::Committed { .. })), "{admitted:?}");
}
let ids = |list: &ForkList| list.forks.iter().map(|entry| entry.card.fork_id.clone()).collect::<Vec<_>>();
let q = |n: u32| format!("eureka-state:question:Q{n}");
let cases = [
(Some(ListStatus::Recorded), None, vec![q(1)]),
(Some(ListStatus::Open), None, vec![q(3), q(2)]),
(None, None, vec![q(3), q(2), q(1)]),
(None, ForkCursor::new(2), vec![q(2), q(1)]),
(Some(ListStatus::Open), ForkCursor::new(2), vec![q(2)]),
];
let now = now();
for (status, cursor, expected) in cases {
let request = forks("x", &["x"], status, cursor, 50);
let (list, _) = page_of(rig.daemon.handle_desk(request.clone(), now));
assert_eq!(ids(&list), expected, "{status:?} {cursor:?} in the daemon");
let (list, _) = page_of(rig.serving(|path| reply_of(path, "m-1", &request)));
assert_eq!(ids(&list), expected, "{status:?} {cursor:?} over the socket");
assert_eq!((list.open_count, list.recorded_count), (2, 1));
}
}
/// Ruling `desk-read-visibility-whole-record`: Huginn returns the whole record, notes included, to
/// whoever the socket carries. The subject is not in the deciders; the decider's note is there.
#[test]
fn a_subject_not_in_deciders_reads_the_whole_record_with_the_notes() {
let mut rig = rig(my_uid());
let hash = rig.hash.clone();
let note = "Keep it local: the author's call.";
rig.serving(|path| counted(reply_of(path, "m-1", &DeskRequest::Answer {
mind: slug(INSTANCE),
subject: short("y"),
authority: DeskAuthority::Member,
deciders: vec![short("y")],
request: noted(&hash, R1, note),
})));
let (list, read) = rig.serving(|path| (reply_of(path, "m-2", &forks("outsider", &["x"], None, None, 50)), reply_of(path, "m-3", &fork_read("outsider", &["x"], FORK))));
let (list, _) = page_of(list);
let (read, _) = entry_of(read);
for entry in [&list.forks[0], &read] {
assert_eq!(entry.record.answers.len(), 1);
assert_eq!(entry.record.answers[0].note.as_deref(), Some(note));
assert_eq!(entry.record.answers[0].answerer, "y");
assert_eq!(entry.card.deciders, ["x"], "the list is the call's, and only fills the card");
}
}
#[test]
fn forks_and_fork_deciders_64_accepted_65_refused() {
let mut rig = rig(my_uid());
let before = rig.state();
let forks = DeskRequest::Forks { mind: slug(INSTANCE), subject: short("x"), deciders: vec![short("x")] };
let fork = DeskRequest::Fork { mind: slug(INSTANCE), subject: short("x"), deciders: vec![short("x")], fork_id: FORK.into() };
let (forks, fork) = rig.serving(|path| (reply_of(path, "m-1", &forks), reply_of(path, "m-2", &fork)));
for response in [forks, fork] {
let DeskResponse::Refused(MindRefusal::Unavailable { detail }) = response else { panic!("{response:?}") };
assert!(detail.contains("desk reads not built"), "{detail}");
let names = |n: usize| (0..n).map(|d| format!("d{d}")).collect::<Vec<_>>();
let (sixty_four, sixty_five) = (names(64), names(65));
let (sixty_four, sixty_five): (Vec<&str>, Vec<&str>) =
(sixty_four.iter().map(String::as_str).collect(), sixty_five.iter().map(String::as_str).collect());
let accepted = rig.serving(|path| (reply_of(path, "m-1", &forks("d0", &sixty_four, None, None, 50)), reply_of(path, "m-2", &fork_read("d0", &sixty_four, FORK))));
assert_eq!(page_of(accepted.0).0.forks[0].card.deciders.len(), 64);
assert_eq!(entry_of(accepted.1).0.card.deciders.len(), 64);
for (what, request) in [("forks", forks("d0", &sixty_five, None, None, 50)), ("fork", fork_read("d0", &sixty_five, FORK))] {
let bytes = framed(&message("m-3", &request));
let refused = rig.serving(|path| failure_of(path, &bytes));
assert_eq!(refused.code, "payload-not-a-request", "{what}");
assert!(refused.message.contains("deciders"), "{what}: {}", refused.message);
assert!(!refused.message.contains("d64"), "{what} echoes a decider: {}", refused.message);
}
assert_eq!(rig.state(), before, "nothing stored, head unchanged");
}
#[test]
fn forks_and_fork_deciders_empty_or_duplicate_refused() {
let mut rig = rig(my_uid());
let before = rig.state();
for (what, list) in [("empty", vec![]), ("duplicate", vec!["x", "x"])] {
for (name, request) in [("forks", forks("x", &list, None, None, 50)), ("fork", fork_read("x", &list, FORK))] {
let bytes = framed(&message("m-1", &request));
let refused = rig.serving(|path| failure_of(path, &bytes));
assert_eq!(refused.code, "payload-not-a-request", "{what} {name}");
assert!(refused.message.contains("deciders"), "{what} {name}: {}", refused.message);
}
}
// One decider is a list.
let ok = rig.serving(|path| reply_of(path, "m-2", &forks_all(&["x"])));
assert_eq!(page_of(ok).0.forks.len(), 1);
assert_eq!(rig.state(), before);
}
/// Chat's reads carry no `authority`: CultLib's list and read requests are closed and have none, so a
/// payload that names it is no request. Nothing is answered and the head does not move.
#[test]
fn a_read_that_names_an_authority_is_refused_at_the_door() {
let mut rig = rig(my_uid());
let before = rig.state();
for (operation, request) in [("forks", forks_all(&["x"])), ("fork", fork_read("x", &["x"], FORK))] {
let mut value = serde_json::to_value(&request).unwrap();
let variant = if operation == "forks" { "Forks" } else { "Fork" };
value[variant]["authority"] = serde_json::json!("Operator");
let payload = rmp_serde::to_vec_named(&value).unwrap();
let sent = framed(&raw(DESK_SERVICE_ID, DESK_REQUEST_SCHEMA, operation, "m-1", payload));
let refused = rig.serving(|path| failure_of(path, &sent));
assert_eq!(refused.code, "payload-not-a-request", "{operation}");
assert!(refused.message.contains(variant), "{operation}: {}", refused.message);
}
assert_eq!(rig.state(), before);
assert!(rig.index.seen.lock().unwrap().is_empty());
// The same reads without it are answered.
let (list, entry) = rig.serving(|path| (reply_of(path, "m-2", &forks_all(&["x"])), reply_of(path, "m-3", &fork_read("x", &["x"], FORK))));
assert_eq!((page_of(list).0.forks.len(), entry_of(entry).0.card.fork_id.as_str().to_string()), (1, FORK.to_string()));
}
/// Fifty wordy forks, one commit each. A limit-50 page of them would pass the response bound: the Mind
/// stops the page short at its budget and names the cursor, so the door sends it whole; the door's own
/// measure stays exact at the bound. Fifty of them admitted in one commit page like any other: the
/// ranks are unique, so the Mind stops the page at its budget inside the commit.
#[test]
fn a_page_is_short_of_the_response_bound_and_the_door_still_measures_it_exactly() {
let mut rig = rig(my_uid());
let (question, _) = question_and_ruling();
let PipelineDocument::Question(held) = question else { unreachable!() };
let documents: Vec<PipelineDocument> = (2..=51)
.map(|n| {
let mut wordy = held.clone();
wordy.label = format!("Q{n}").as_str().into();
wordy.question = "Should a slot mark be shared between authors? ".repeat(18).as_str().into();
PipelineDocument::Question(wordy)
})
.collect();
let admitted = rig.daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, documents.clone())), now()).answered();
assert!(matches!(admitted, HuginnMindResponse::Admit(huginn_mind::PipelineAdmissionOutcome::Committed { .. })), "{admitted:?}");
let before = rig.state();
let grouped = rig.serving(|path| reply_of(path, "m-0", &forks_all(&["x"])));
let (grouped, _) = page_of(grouped);
assert!(grouped.forks.len() > 1 && grouped.forks.len() < 50 && grouped.next_cursor.is_some(), "{} entries", grouped.forks.len());
assert_eq!(rig.state(), before);
// Then fifty more, each in its own commit, newer than the group.
for n in 52..=101 {
let PipelineDocument::Question(mut wordy) = documents[0].clone() else { unreachable!() };
wordy.label = format!("Q{n}").as_str().into();
let admitted = rig.daemon.handle(HuginnMindRequest::Admit(batch(INSTANCE, vec![PipelineDocument::Question(wordy)])), now()).answered();
assert!(matches!(admitted, HuginnMindResponse::Admit(huginn_mind::PipelineAdmissionOutcome::Committed { .. })), "{admitted:?}");
}
let before = rig.state();
let (big, small) = rig.serving(|path| (reply_of(path, "m-1", &forks_all(&["x"])), reply_of(path, "m-2", &forks("x", &["x"], None, None, 10))));
let (big, _) = page_of(big);
assert!(big.forks.len() > 10 && big.forks.len() < 50 && big.next_cursor.is_some(), "{} entries", big.forks.len());
let (small, _) = page_of(small);
assert_eq!((small.forks.len(), small.open_count), (10, 101));
assert!(small.next_cursor.is_some());
assert_eq!(rig.state(), before);
// The bound is exact on a read: a response of N bytes passes at N and is refused at N - 1.
let response = DeskResponse::Forks { list: small, as_of: 1 };
let size = {
let message = huginn_mind::envelope::encode_desk_response("m-9", "forks", &response, "huginn-yggdrasil").unwrap();
encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap().len()
};
let at = |limit: usize| decode_desk_response(&crate::desk_door::respond(&response, "m-9", "forks", "huginn-yggdrasil", limit)).unwrap().1.unwrap();
assert_eq!(at(size), response);
assert!(matches!(at(size - 1), DeskResponse::Refused(MindRefusal::ResponseTooLarge { .. })));
}
}
@ -817,17 +1044,26 @@ mod bounds {
fn a_peer_that_never_reads_is_cut_off_at_one_second() {
let (_peer, server) = UnixStream::pair().unwrap();
let mut stream = Deadlined::new(server, DESK_CONNECTION_DEADLINE);
let (chunk, started) = (vec![0_u8; 64 * 1024], Instant::now());
let error = loop {
if let Err(error) = stream.write(&chunk) {
break error;
}
};
let took = started.elapsed();
// The writer runs on its own thread and the test waits for it with a clock of its own: without
// the deadline the write blocks for good, and that must fail this test, not hang the suite.
let (sent, heard) = std::sync::mpsc::channel();
let writer = std::thread::spawn(move || {
let (chunk, started) = (vec![0_u8; 64 * 1024], Instant::now());
let error = loop {
if let Err(error) = stream.write(&chunk) {
break error;
}
};
let took = started.elapsed();
let again = stream.write(&chunk).unwrap_err().kind();
let _ = sent.send((error, took, again));
});
let (error, took, again) = heard.recv_timeout(Duration::from_secs(5)).expect("a peer that never reads is cut off within seconds, not waited on");
writer.join().unwrap();
// The kernel's send timeout reads as WouldBlock; the deadline already spent reads as TimedOut.
assert!(matches!(error.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut), "{error:?}");
assert!(took >= Duration::from_millis(950) && took < Duration::from_millis(1250), "the write was cut after {took:?}");
assert_eq!(stream.write(&chunk).unwrap_err().kind(), std::io::ErrorKind::TimedOut, "once spent, nothing more is written");
assert_eq!(again, std::io::ErrorKind::TimedOut, "once spent, nothing more is written");
}
/// A catalog is a reply like any other and is measured against the response bound: one that

View file

@ -210,6 +210,17 @@ fn answer(who: &str, hash: &str, request_id: &str) -> DeskRequest {
}
}
fn forks_request(who: &str, deciders: &[&str]) -> DeskRequest {
DeskRequest::Forks {
mind: INSTANCE.into(),
subject: short(who),
deciders: deciders.iter().map(|d| short(d)).collect(),
status: None,
cursor: None,
limit: huginn_mind::desk_wire::ForkLimit::new(50).unwrap(),
}
}
fn framed(message: &CultNetMessage) -> Vec<u8> {
encode_frame(&encode_cultnet_message_to_vec(message, CultNetWireContract::CultNetSchemaV0).unwrap()).unwrap()
}
@ -266,14 +277,37 @@ fn the_binary_serves_both_doors_and_survives_restart() {
let catalog = format!("{:?}", exchange(&desk, &framed(&catalog_request())).expect("a catalog"));
assert!(catalog.contains("huginn.desk_request.v1") && catalog.contains("huginn.desk_response.v1"), "{catalog}");
assert!(!catalog.contains("huginn.mind_request.v1"), "{catalog}");
// Reads are not built: Unavailable, nothing stored.
let forks = DeskRequest::Forks { mind: INSTANCE.into(), subject: short("x"), deciders: vec![short("x")] };
let fork = DeskRequest::Fork { mind: INSTANCE.into(), subject: short("x"), deciders: vec![short("x")], fork_id: FORK.into() };
for request in [forks, fork] {
let DeskResponse::Refused(huginn_mind::MindRefusal::Unavailable { detail }) = ask(&desk, "m-r", &request) else {
panic!("a read is refused unavailable")
// The reads answer, store nothing, and carry the whole record to a subject the deciders do not name.
let (list, listed_at) = match ask(&desk, "m-r1", &forks_request("x", &["x"])) {
DeskResponse::Forks { list, as_of } => (list, as_of),
other => panic!("{other:?}"),
};
assert_eq!((list.open_count, list.recorded_count, list.forks.len(), list.stamp.clone()), (1, 0, 1, format!("s{listed_at}")));
assert_eq!(list.forks[0].card.options_hash, hash);
let DeskResponse::Fork { entry, as_of } = ask(&desk, "m-r2", &DeskRequest::Fork { mind: INSTANCE.into(), subject: short("x"), deciders: vec![short("x")], fork_id: FORK.into() })
else {
panic!("a read of the seeded fork")
};
assert_eq!((entry, as_of), (list.forks[0].clone(), listed_at));
assert_eq!(daemon.documents(), seeded);
// `authority` is no field of a read: the payload is refused whole and nothing is answered or stored.
for (operation, request) in [("forks", forks_request("x", &["x"])), ("fork", DeskRequest::Fork { mind: INSTANCE.into(), subject: short("x"), deciders: vec![short("x")], fork_id: FORK.into() })] {
let mut value = serde_json::to_value(&request).unwrap();
let variant = if operation == "forks" { "Forks" } else { "Fork" };
value[variant]["authority"] = serde_json::json!("Operator");
let message = CultNetMessage::OperationRequest {
message_id: "m-ra".into(),
service_id: huginn_mind::desk_wire::DESK_SERVICE_ID.into(),
operation: operation.into(),
payload_schema: huginn_mind::desk_wire::DESK_REQUEST_SCHEMA.into(),
payload_encoding: "messagepack-base64".into(),
payload: base64::Engine::encode(&base64::engine::general_purpose::STANDARD, rmp_serde::to_vec_named(&value).unwrap()),
source_runtime_id: None,
target_runtime_id: None,
};
assert!(detail.contains("desk reads not built"), "{detail}");
let reply = exchange(&desk, &framed(&message)).expect("the door replied");
let failure = decode_desk_response(&reply).unwrap().1.unwrap_err();
assert_eq!(failure.code, "payload-not-a-request", "{operation}");
}
assert_eq!(daemon.documents(), seeded);
// An answer, then a stale one.
@ -285,6 +319,16 @@ fn the_binary_serves_both_doors_and_survives_restart() {
assert_eq!((stale.code, counted_answer_id.as_deref()), (Some(ForkAnswerCode::StaleFork), Some("n1")));
let stored = daemon.documents();
assert_eq!(stored, seeded + 2);
// A subject the call's deciders do not name reads both answers, the first with its choice and the stale one with its code.
let DeskResponse::Fork { entry, .. } =
ask(&desk, "m-r3", &DeskRequest::Fork { mind: INSTANCE.into(), subject: short("outsider"), deciders: vec![short("y")], fork_id: FORK.into() })
else {
panic!("a read")
};
let seen: Vec<_> = entry.record.answers.iter().map(|answer| (answer.answer_id.as_str(), answer.answerer.as_str(), answer.code)).collect();
assert_eq!(seen, [("n1", "x", None), ("n2", "x", Some(ForkAnswerCode::StaleFork))]);
assert_eq!((entry.record.counted_answer_id.as_deref(), entry.card.deciders.as_slice()), (Some("n1"), ["y".to_string()].as_slice()));
assert_eq!(daemon.documents(), stored);
// A stop takes the socket with it; the stored answers are the retry's memory.
daemon.signal("TERM");

View file

@ -7,8 +7,7 @@
//! request id). It builds one `desk_answer` and commits it through
//! `admit_steps` with `Door::Desk`, the same commit primitive every other write
//! uses; no other door may write the kind. `answer_desk` and `DeskStamp` are
//! crate-private: only code inside this crate can stamp, so the verified door
//! that will call them (follow-up desk-answer-door) widens them in its own cut.
//! crate-private: only code inside this crate can stamp.
//!
//! The decider list is an input (`DeskStamp.deciders`, from the door); the Mind
//! never reads one. Whether an answerer decides is membership in that list and
@ -125,7 +124,7 @@ pub fn options_hash(question: &PipelineQuestion) -> Sha256Hex {
}
/// Every desk answer of one fork, with its key.
fn answers_of<'a>(docs: &'a Docs<'_>, fork: &'a str) -> impl Iterator<Item = (&'a str, &'a PipelineDeskAnswer)> + 'a {
pub(crate) fn answers_of<'a>(docs: &'a Docs<'_>, fork: &'a str) -> impl Iterator<Item = (&'a str, &'a PipelineDeskAnswer)> + 'a {
docs.of_kind(PipelineKind::DeskAnswer).filter_map(move |(key, document)| match document {
PipelineDocument::DeskAnswer(answer) if answer.fork.0 == fork => Some((key, answer)),
_ => None,

View file

@ -1,8 +1,10 @@
//! The fork desk's answer wire, Huginn's half: Libby's answer-side documents
//! (CultLib `contracts/cultnet/governance/`, the contract file and its vectors at 6231ecad)
//! (CultLib `contracts/cultnet/governance/`, the contract file and its vectors at e57fbaee)
//! as Rust types, the one conversion between her `n<seq>` answer ids and the
//! record keys, the projection of a stored answer, the refusal mapping and the
//! showability predicate, and `Mind::desk_answer`, the one call that runs them in order.
//! The read side (card, record, entry and list, `Mind::desk_forks` and `Mind::desk_fork`)
//! is the child module `reads`.
//!
//! `Mind::answer_desk` stays the only owner of an answer's outcome, sequence,
//! request id lookup, replay and conflict; this module converts, projects and
@ -11,7 +13,10 @@
//! the response is read at, and nothing here stores it. Huginn writes
//! `gamecult.desk_answer.v0` and its own refusals; the stale-fork display
//! and the answer result are Bifrost's to attach and write. The request type, its
//! constants and the published schemas are here; the reads and the socket are other cuts.
//! constants and the published schemas are here; the socket is the daemon's.
mod reads;
pub use reads::{CardSchema, CardStatus, EntrySchema, ForkCard, ForkCursor, ForkEntry, ForkLimit, ForkList, ForkOption, ForkRecord, ForkRef, ListSchema, ListStatus, RecordSchema};
use chrono::{DateTime, Utc};
use eureka_pipeline::{
@ -169,13 +174,24 @@ pub struct ForkRefusal {
/// What the door answers one desk request with. `Answered` is a stored answer, refused ones
/// included (a stale fork, a non-decider and the rest are stored), fresh or replayed alike.
/// `NotStored` is a request that stored nothing. `Refused` is any other Mind refusal; an
/// `Unavailable` is Bifrost's to turn into its 503, and Huginn writes no document for it.
/// `Forks` and `Fork` are the reads: one page of entries, or one entry, with the head they were
/// all read at. `NotStored` is a request that stored nothing (an unknown fork included).
/// `Refused` is any other Mind refusal; an `Unavailable` is Bifrost's to turn into its 503, and
/// Huginn writes no document for it.
///
/// The trust boundary of the reads (ruling `desk-read-visibility-whole-record`): `Forks` and `Fork`
/// carry the WHOLE record, every answer's note included, to whatever the peer-checked socket
/// carries. Huginn strips nothing and gates nothing; Bifrost applies visibility at its door (a
/// stripped read serves `note: null`, CultLib `fork-card.md`). Huginn must therefore be reachable
/// only by Bifrost's peer-checked socket: the peer-uid check IS the whole authentication, and a
/// second reader of that socket is a note leak.
// One value per call, built and encoded once; a box would only add a deref to every reader.
#[allow(clippy::large_enum_variant)]
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub enum DeskResponse {
Answered { answer: ForkDeskAnswer, counted_answer_id: Option<String>, as_of: u64 },
Forks { list: ForkList, as_of: u64 },
Fork { entry: ForkEntry, as_of: u64 },
NotStored(ForkRefusal),
Refused(MindRefusal),
}
@ -184,7 +200,7 @@ impl DeskResponse {
/// The envelope's status: `rejected` exactly when nothing was stored or the Mind refused.
pub fn status(&self) -> &'static str {
match self {
Self::Answered { .. } => "accepted",
Self::Answered { .. } | Self::Forks { .. } | Self::Fork { .. } => "accepted",
Self::NotStored(_) | Self::Refused(_) => "rejected",
}
}
@ -203,6 +219,11 @@ pub const DESK_RESPONSE_SCHEMA_JSON: &str = include_str!("../../../schemas/cultn
/// The frame bounds the door enforces, on the encoded CultNet message.
pub const DESK_MAX_REQUEST_BYTES: usize = 65536;
pub const DESK_MAX_RESPONSE_BYTES: usize = 65536;
/// What a page of forks leaves for the envelope around it: the echoed message id (at most 256
/// bytes), the operation, the runtime id and the frame's own keys. `Mind::desk_forks` fills a page
/// to `DESK_PAGE_BUDGET`, the response bound less this, measured on the bare message.
pub const DESK_ENVELOPE_MARGIN: usize = 1024;
pub const DESK_PAGE_BUDGET: usize = DESK_MAX_RESPONSE_BYTES - DESK_ENVELOPE_MARGIN;
/// Most deciders one call carries (CultLib `card.deciders` `maxItems`).
pub const DESK_DECIDERS_MAX: usize = 64;
@ -249,8 +270,17 @@ fn deciders_schema(generator: &mut schemars::SchemaGenerator) -> schemars::Schem
/// Every request the desk door answers. The fields are the wire names: the `mind` addressed
/// (the persona and pipeline requests say `instance`; the sets are separate), the verified
/// `subject` and `authority` Bifrost vouches for (door context, never body fields), and the
/// `deciders` of this call (1 to 64, distinct; the Mind never stores them). Closed: a key outside the set refuses
/// the whole payload. The reads answer in follow-up desk-wire-reads.
/// `deciders` of this call (1 to 64, distinct; the Mind never stores them). Closed: a key outside
/// the set refuses the whole payload.
///
/// The reads, `Forks` and `Fork`, carry NO `authority` (CultLib `list_request` and
/// `read_request` are closed and have none; one that names it is refused whole). Their `subject`
/// and `deciders` gate nothing: `deciders` only fills the card's `deciders` and the showability
/// gate, `subject` is read by no read. `Forks` takes the contract's `list_request` fields flat
/// beside them (`status`, `cursor`, `limit`), each key written, `null` where absent. Huginn
/// answers every read with the whole record and Bifrost applies visibility; see `DeskResponse`
/// for the trust boundary. These docs and the doc of `Mind::desk_forks`/`Mind::desk_fork` are the
/// reads spec.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub enum DeskRequest {
@ -269,6 +299,13 @@ pub enum DeskRequest {
#[serde(deserialize_with = "deciders")]
#[schemars(schema_with = "deciders_schema")]
deciders: Vec<Short>,
#[serde(deserialize_with = "required")]
#[schemars(required)]
status: Option<ListStatus>,
#[serde(deserialize_with = "required")]
#[schemars(required)]
cursor: Option<ForkCursor>,
limit: ForkLimit,
},
Fork {
mind: Slug,
@ -446,9 +483,8 @@ fn showable(question: &PipelineQuestion, deciders: &[Short]) -> bool {
impl<S: MindStore> Mind<S> {
/// One desk answer from the door: the wire request converted, the question gated, the answer
/// decided and stored (or replayed) by `answer_desk`, and the stored answer projected with the
/// derived count at the head it was read at. The flag is `true` when nothing was committed (a
/// replay); it is for the door and is never encoded. `subject` and `authority` are the verified
/// decided and stored (or committed) by `answer_desk`, and the stored answer projected with the
/// derived count at the head it was read at. `subject` and `authority` are the verified
/// caller's, `deciders` the door's list.
pub fn desk_answer(
&mut self,
@ -457,10 +493,10 @@ impl<S: MindStore> Mind<S> {
authority: DeskAuthority,
request: ForkAnswerRequest,
now: DateTime<Utc>,
) -> (DeskResponse, bool) {
) -> DeskResponse {
match self.desk_response(deciders, subject, authority, &request, now) {
Ok(done) => done,
Err(refused) => (door_refusal(refused), false),
Err(refused) => door_refusal(refused),
}
}
@ -471,10 +507,10 @@ impl<S: MindStore> Mind<S> {
authority: DeskAuthority,
request: &ForkAnswerRequest,
now: DateTime<Utc>,
) -> Result<(DeskResponse, bool), MindRefusal> {
) -> Result<DeskResponse, MindRefusal> {
let sent = match request_of(request) {
Ok(sent) => sent,
Err(refused) => return Ok((DeskResponse::NotStored(refused), false)),
Err(refused) => return Ok(DeskResponse::NotStored(refused)),
};
let image = self.image()?;
let docs = Docs::at(image, image.head);
@ -483,12 +519,11 @@ impl<S: MindStore> Mind<S> {
let unshowable = bound_answer(&docs, &subject, &sent.request_id).is_none()
&& matches!(docs.find(PipelineKind::Question, &sent.fork.0), Some(PipelineDocument::Question(question)) if !showable(question, deciders));
if unshowable {
return Ok((DeskResponse::NotStored(fork_unknown()), false));
return Ok(DeskResponse::NotStored(fork_unknown()));
}
let stamp = DeskStamp { answerer: subject, authority, deciders: deciders.to_vec() };
let (key, replayed) = match self.answer_desk(stamp, sent, now) {
DeskAnswerOutcome::Stored { answer, .. } => (answer.0, false),
DeskAnswerOutcome::Replayed { answer, .. } => (answer.0, true),
let key = match self.answer_desk(stamp, sent, now) {
DeskAnswerOutcome::Stored { answer, .. } | DeskAnswerOutcome::Replayed { answer, .. } => answer.0,
DeskAnswerOutcome::Refused(refused) => return Err(refused),
};
let image = self.image()?;
@ -496,7 +531,7 @@ impl<S: MindStore> Mind<S> {
let Some(PipelineDocument::DeskAnswer(stored)) = docs.find(PipelineKind::DeskAnswer, &key) else { return Err(unreadable_key()) };
let fork = stored.fork.0.as_str();
let counted_answer_id = counted_answer(&docs, fork).map(|counted| answer_seq(fork, &counted).map(n_text).ok_or_else(unreadable_key)).transpose()?;
Ok((DeskResponse::Answered { answer: project(&key, stored)?, counted_answer_id, as_of: image.head }, replayed))
Ok((DeskResponse::Answered { answer: project(&key, stored)?, counted_answer_id, as_of: image.head }))
}
}
@ -584,8 +619,12 @@ mod tests {
list.iter().map(|who| s(who)).collect()
}
/// The response and whether the call committed anything: a replay, a refused body and a gate
/// refusal store nothing, so the head does not move.
fn ask_as(mind: &mut Mind<MemoryStore>, who: &str, list: &[&str], request: ForkAnswerRequest) -> (DeskResponse, bool) {
mind.desk_answer(&deciders(list), s(who), DeskAuthority::Member, request, now())
let before = head(mind);
let response = mind.desk_answer(&deciders(list), s(who), DeskAuthority::Member, request, now());
(response, head(mind) != before)
}
fn ask(mind: &mut Mind<MemoryStore>, who: &str, request: ForkAnswerRequest) -> (DeskResponse, bool) {
@ -846,9 +885,9 @@ mod tests {
assert_eq!((answer.outcome.as_str(), answer.retracts.as_deref(), counted.as_deref(), as_of), ("Retraction", Some("n1"), Some("n2"), head(&mind)));
// The first request replayed: its stored word is still Counted, the count is read now.
let before = head(&mind);
let (response, replayed) = ask(&mut mind, "x", first);
let (response, committed) = ask(&mut mind, "x", first);
let (answer, counted, as_of) = answered(response);
assert!(replayed);
assert!(!committed);
assert_eq!((answer.outcome.as_str(), answer.answer_id.as_str(), counted.as_deref(), as_of), ("Counted", "n1", Some("n2"), before));
assert_eq!(head(&mind), before);
// The second request replayed now counts, and its stored word is still Revision.
@ -984,14 +1023,14 @@ mod tests {
];
for (form, request) in forms.into_iter().enumerate() {
let stored = stored_count(&mind, FORK);
let (response, replayed) = ask(&mut mind, "x", request.clone());
assert!(!replayed);
let (response, committed) = ask(&mut mind, "x", request.clone());
assert!(committed);
let (answer, _, _) = answered(response.clone());
assert_eq!((answer.outcome.as_str(), answer.code), ("Refused", Some(ForkAnswerCode::BadRetraction)), "form {form}");
assert_eq!(stored_count(&mind, FORK), stored + 1, "form {form}");
// The refusal is stored and bound: the same bytes replay, with no write.
let before = head(&mind);
assert_eq!(ask(&mut mind, "x", request), (response, true), "form {form}");
assert_eq!(ask(&mut mind, "x", request), (response, false), "form {form}");
assert_eq!(head(&mind), before);
}
// (i) and (iii) with equal answer and target differ only in `retracted`: one id, two bodies.
@ -1117,14 +1156,14 @@ mod tests {
fn a_stale_answer_carries_no_card() {
let mut mind = desk();
let stale = wire(FORK, &"e".repeat(64));
let (fresh, replayed) = ask(&mut mind, "x", stale.clone());
assert!(!replayed);
let (fresh, committed) = ask(&mut mind, "x", stale.clone());
assert!(committed);
let (answer, _, _) = answered(fresh.clone());
assert_eq!((answer.outcome.as_str(), answer.code), ("Refused", Some(ForkAnswerCode::StaleFork)));
// After the fork closed, the retry still answers the stored refusal, with nothing added.
close(&mut mind, FORK, "RQ1", "A");
let (again, replayed) = ask(&mut mind, "x", stale);
assert!(replayed);
let (again, committed) = ask(&mut mind, "x", stale);
assert!(!committed);
let (again_answer, _, _) = answered(again.clone());
assert_eq!(again_answer, answer);
for response in [&fresh, &again] {
@ -1139,7 +1178,7 @@ mod tests {
let request = wire(FORK, &hash);
let (fresh, first) = ask(&mut mind, "x", request.clone());
let (replay, second) = ask(&mut mind, "x", request);
assert_eq!((first, second), (false, true));
assert_eq!((first, second), (true, false));
assert_eq!(encode(&fresh), encode(&replay));
assert_eq!(stored_count(&mind, FORK), 1);
}
@ -1188,17 +1227,17 @@ mod tests {
let mut mind = desk();
let hash = hash_of(&mind, FORK);
let request = wire(FORK, &hash);
let (stored, replayed) = ask(&mut mind, "x", request.clone());
assert!(!replayed);
let (stored, committed) = ask(&mut mind, "x", request.clone());
assert!(committed);
let before = head(&mind);
// The list Bifrost passes is empty now; the stored request still gets its stored answer.
let (again, replayed) = ask_as(&mut mind, "x", &[], request.clone());
assert!(replayed);
let (again, committed) = ask_as(&mut mind, "x", &[], request.clone());
assert!(!committed);
assert_eq!(encode(&again), encode(&stored));
assert_eq!(head(&mind), before);
// Another subject with the same bytes has stored nothing: the gate refuses it.
let (other, replayed) = ask_as(&mut mind, "y", &[], request);
assert_eq!((not_stored(other).code, replayed), (ForkRefusalCode::ForkUnknown, false));
let (other, committed) = ask_as(&mut mind, "y", &[], request);
assert_eq!((not_stored(other).code, committed), (ForkRefusalCode::ForkUnknown, false));
// And a request nobody stored is fork_unknown while the list is empty.
let fresh = wire(FORK, &hash);
assert_eq!(not_stored(ask_as(&mut mind, "x", &[], fresh).0).code, ForkRefusalCode::ForkUnknown);
@ -1206,22 +1245,22 @@ mod tests {
// A stored refusal and a stored revision replay the same way: a stale hash is stored as a
// refusal, a second answer by the same subject as a revision.
let stale = wire(FORK, "0".repeat(hash.len()).as_str());
let (refused, replayed) = ask(&mut mind, "x", stale.clone());
assert!(!replayed);
let (refused, committed) = ask(&mut mind, "x", stale.clone());
assert!(committed);
assert_eq!(answered(refused.clone()).0.outcome, "Refused");
let revision = wire(FORK, &hash);
let (revised, replayed) = ask(&mut mind, "x", revision.clone());
assert!(!replayed);
let (revised, committed) = ask(&mut mind, "x", revision.clone());
assert!(committed);
assert_eq!(answered(revised.clone()).0.outcome, "Revision");
let before = head(&mind);
for (request, stored) in [(stale, refused), (revision, revised)] {
let (again, replayed) = ask_as(&mut mind, "x", &[], request.clone());
assert!(replayed);
let (again, committed) = ask_as(&mut mind, "x", &[], request.clone());
assert!(!committed);
// `as_of` is the head the replay was read at, so the stored answer is what is compared.
assert_eq!(encode(&answered(again).0), encode(&answered(stored).0));
assert_eq!(head(&mind), before);
let (other, replayed) = ask_as(&mut mind, "w", &[], request);
assert_eq!((not_stored(other).code, replayed), (ForkRefusalCode::ForkUnknown, false));
let (other, committed) = ask_as(&mut mind, "w", &[], request);
assert_eq!((not_stored(other).code, committed), (ForkRefusalCode::ForkUnknown, false));
assert_eq!(head(&mind), before);
}
}
@ -1285,16 +1324,16 @@ mod tests {
let (first, _) = ask(&mut mind, "x", request.clone());
// V12: the same bytes replay.
let before = head(&mind);
assert_eq!(ask(&mut mind, "x", request.clone()), (first, true));
assert_eq!(ask(&mut mind, "x", request.clone()), (first, false));
assert_eq!(head(&mind), before);
// V13: only answered_at changed is a conflict, and nothing is stored.
let moved = ForkAnswerRequest { answered_at: "2026-11-10T12:00:01Z".into(), ..request.clone() };
assert_eq!(not_stored(ask(&mut mind, "x", moved).0).code, ForkRefusalCode::RequestIdConflict);
assert_eq!((head(&mind), stored_count(&mind, FORK)), (before, 1));
// V14: another subject with identical bytes is its own request: its own answer id, stamped with it.
let (other, replayed) = ask(&mut mind, "y", request.clone());
let (other, committed) = ask(&mut mind, "y", request.clone());
let (other, _, _) = answered(other);
assert!(!replayed);
assert!(committed);
assert_eq!((other.answer_id.as_str(), other.answerer.as_str(), other.outcome.as_str(), other.request_id.as_str()), ("n2", "y", "Revision", request.request_id.as_str()));
let (original, _, _) = answered(ask(&mut mind, "x", request.clone()).0);
assert_eq!((original.answer_id.as_str(), original.answerer.as_str()), ("n1", "x"));
@ -1452,7 +1491,7 @@ mod tests {
let authority = serde_json::from_value(server["authority"].clone()).unwrap();
let at = DateTime::parse_from_rfc3339(server["received_at"].as_str().unwrap()).unwrap().to_utc();
let answerer = s(server["answerer"].as_str().unwrap());
let (mine, _) = mind.desk_answer(&deciders(&list), answerer, authority, request, at);
let mine = mind.desk_answer(&deciders(&list), answerer, authority, request, at);
if code == Some("unavailable") {
continue; // the door went down after the Mind stored: its answer was never seen
}
@ -1508,9 +1547,9 @@ mod tests {
let mut mind = desk();
let hash = hash_of(&mind, FORK);
// The operator's own account, off the list, is no decider; a member on the list is.
let off = mind.desk_answer(&deciders(&["x"]), s("op"), DeskAuthority::Operator, wire(FORK, &hash), now()).0;
let off = mind.desk_answer(&deciders(&["x"]), s("op"), DeskAuthority::Operator, wire(FORK, &hash), now());
assert_eq!(answered(off).0.code, Some(ForkAnswerCode::NotADecider));
let on = mind.desk_answer(&deciders(&["m"]), s("m"), DeskAuthority::Member, wire(FORK, &hash), now()).0;
let on = mind.desk_answer(&deciders(&["m"]), s("m"), DeskAuthority::Member, wire(FORK, &hash), now());
let (answer, _, _) = answered(on);
assert_eq!((answer.outcome.as_str(), answer.authority, answer.answerer.as_str()), ("Counted", DeskAuthority::Member, "m"));
}
@ -1527,7 +1566,7 @@ mod tests {
fn desk_requests(mind: &str, list: &[Short], hash: &str) -> [DeskRequest; 3] {
[
DeskRequest::Answer { mind: slug(mind), subject: s("d0"), authority: DeskAuthority::Member, deciders: list.to_vec(), request: wire(FORK, hash) },
DeskRequest::Forks { mind: slug(mind), subject: s("d0"), deciders: list.to_vec() },
DeskRequest::Forks { mind: slug(mind), subject: s("d0"), deciders: list.to_vec(), status: None, cursor: None, limit: ForkLimit::new(10).unwrap() },
DeskRequest::Fork { mind: slug(mind), subject: s("d0"), deciders: list.to_vec(), fork_id: FORK.into() },
]
}
@ -1550,8 +1589,9 @@ mod tests {
}
// 64 deciders through to the Mind: the subject is listed and counts.
let DeskRequest::Answer { subject, authority, deciders, request, .. } = through_the_door(&desk_requests(INSTANCE, &list_of(64), &hash)[0]).unwrap() else { panic!("not an answer") };
let (response, replayed) = mind.desk_answer(&deciders, subject, authority, request, now());
assert!(!replayed);
let before = head(&mind);
let response = mind.desk_answer(&deciders, subject, authority, request, now());
assert!(head(&mind) > before);
assert_eq!(answered(response).0.outcome, "Counted");
let (head_before, stored_before) = (head(&mind), stored_count(&mind, FORK));
@ -1585,7 +1625,7 @@ mod tests {
assert_eq!((head(&mind), stored_count(&mind, FORK)), before);
// The Mind's showability gate is the second line: an empty list handed to it directly
// still refuses the fork and stores nothing.
let refused = not_stored(mind.desk_answer(&[], s("d0"), DeskAuthority::Member, wire(FORK, &hash), now()).0);
let refused = not_stored(mind.desk_answer(&[], s("d0"), DeskAuthority::Member, wire(FORK, &hash), now()));
assert_eq!(refused.code, ForkRefusalCode::ForkUnknown);
assert_eq!((head(&mind), stored_count(&mind, FORK)), before);
}
@ -1606,6 +1646,6 @@ mod tests {
// The same request to the mind it names is answered.
let DeskRequest::Answer { mind: named, subject, authority, deciders, request } = through_the_door(&desk_requests(INSTANCE, &list_of(4), &hash)[0]).unwrap() else { panic!("not an answer") };
mind.require_instance(&named).unwrap();
assert_eq!(answered(mind.desk_answer(&deciders, subject, authority, request, now()).0).0.outcome, "Counted");
assert_eq!(answered(mind.desk_answer(&deciders, subject, authority, request, now())).0.outcome, "Counted");
}
}

File diff suppressed because it is too large Load diff

View file

@ -69,6 +69,11 @@ impl<'i> Docs<'i> {
self.image.id_of(kind, id).filter(|row| self.visible(*row)).map(|row| &self.image.row(row).view.document)
}
/// The ordinal of the receipt that wrote an image document: its admission order.
pub(crate) fn ordinal_of(&self, kind: PipelineKind, id: &str) -> Option<u64> {
self.image.id_of(kind, id).filter(|row| self.visible(*row)).map(|row| self.image.row(row).ordinal)
}
pub(crate) fn find(&self, kind: PipelineKind, id: &str) -> Option<&PipelineDocument> {
self.in_batch(kind, id).or_else(|| self.in_image(kind, id))
}

View file

@ -428,7 +428,7 @@ mod tests {
let (mind, subject, deciders) = (slug(INSTANCE), crate::fixtures::s("user-7"), vec![crate::fixtures::s("user-7")]);
[
(desk_answer(), "Answer"),
(DeskRequest::Forks { mind: mind.clone(), subject: subject.clone(), deciders: deciders.clone() }, "Forks"),
(DeskRequest::Forks { mind: mind.clone(), subject: subject.clone(), deciders: deciders.clone(), status: Some(crate::desk_wire::ListStatus::Open), cursor: crate::desk_wire::ForkCursor::new(7), limit: crate::desk_wire::ForkLimit::new(50).unwrap() }, "Forks"),
(DeskRequest::Fork { mind, subject, deciders, fork_id: "eureka-state:question:Q1".into() }, "Fork"),
]
}
@ -446,7 +446,7 @@ mod tests {
};
let expected = [
("Answer", vec!["authority", "deciders", "mind", "request", "subject"]),
("Forks", vec!["deciders", "mind", "subject"]),
("Forks", vec!["cursor", "deciders", "limit", "mind", "status", "subject"]),
("Fork", vec!["deciders", "fork_id", "mind", "subject"]),
];
for ((request, variant), (named, names)) in desk_requests().iter().zip(expected) {
@ -595,7 +595,9 @@ mod tests {
..request
},
},
DeskRequest::Forks { mind, .. } => DeskRequest::Forks { mind, subject: crate::fixtures::s(&long(200)), deciders: deciders.clone() },
DeskRequest::Forks { mind, status, cursor, limit, .. } => {
DeskRequest::Forks { mind, subject: crate::fixtures::s(&long(200)), deciders: deciders.clone(), status, cursor, limit }
}
DeskRequest::Fork { mind, .. } => {
DeskRequest::Fork { mind, subject: crate::fixtures::s(&long(200)), deciders: deciders.clone(), fork_id: long(200) }
}

View file

@ -55,7 +55,7 @@ mod read_tests;
mod move_tests;
pub use desk::{DeskAnswerOutcome, DeskAnswerRequest, DeskBody, options_hash};
pub use desk_wire::{DeskResponse, ForkAnswer, ForkAnswerCode, ForkAnswerRequest, ForkDeskAnswer, ForkRefusal, ForkRefusalCode};
pub use desk_wire::{DeskRequest, DeskResponse, ForkAnswer, ForkAnswerCode, ForkAnswerRequest, ForkDeskAnswer, ForkRefusal, ForkRefusalCode};
pub use docs::CitationRole;
pub use index::{INDEX_TEXT_MAX_BYTES, INDEX_TEXT_VERSION, IndexEntry, index_text};
pub use admission::{BATCH_MAX, PipelineAdmissionBatch, PipelineAdmissionOutcome};

View file

@ -109,6 +109,11 @@ pub enum MindRefusal {
/// 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 desk page whose first entry alone is past the page budget: `fork` is that fork, `order` its
/// rank (a caller goes on past it with the cursor `c<order>`, which returns the forks below it),
/// `bytes` what the page holding it would encode to (without the envelope) and `limit` the page
/// budget it exceeds. Raised by the mind's page fill, which is why it is not `ResponseTooLarge`.
DeskForkTooLarge { fork: String, order: u64, 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.
SelectionInvalid { field: String, value: Option<String>, message: String },

File diff suppressed because one or more lines are too long

View file

@ -1,7 +1,7 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "DeskRequest",
"description": "Every request the desk door answers. The fields are the wire names: the `mind` addressed\n(the persona and pipeline requests say `instance`; the sets are separate), the verified\n`subject` and `authority` Bifrost vouches for (door context, never body fields), and the\n`deciders` of this call (1 to 64, distinct; the Mind never stores them). Closed: a key outside the set refuses\nthe whole payload. The reads answer in follow-up desk-wire-reads.",
"description": "Every request the desk door answers. The fields are the wire names: the `mind` addressed\n(the persona and pipeline requests say `instance`; the sets are separate), the verified\n`subject` and `authority` Bifrost vouches for (door context, never body fields), and the\n`deciders` of this call (1 to 64, distinct; the Mind never stores them). Closed: a key outside\nthe set refuses the whole payload.\n\nThe reads, `Forks` and `Fork`, carry NO `authority` (CultLib `list_request` and\n`read_request` are closed and have none; one that names it is refused whole). Their `subject`\nand `deciders` gate nothing: `deciders` only fills the card's `deciders` and the showability\ngate, `subject` is read by no read. `Forks` takes the contract's `list_request` fields flat\nbeside them (`status`, `cursor`, `limit`), each key written, `null` where absent. Huginn\nanswers every read with the whole record and Bifrost applies visibility; see `DeskResponse`\nfor the trust boundary. These docs and the doc of `Mind::desk_forks`/`Mind::desk_fork` are the\nreads spec.",
"oneOf": [
{
"type": "object",
@ -52,6 +52,10 @@
"Forks": {
"type": "object",
"properties": {
"cursor": {
"type": "string",
"pattern": "^c[1-9][0-9]*$"
},
"deciders": {
"type": "array",
"items": {
@ -61,9 +65,20 @@
"minItems": 1,
"uniqueItems": true
},
"limit": {
"$ref": "#/$defs/ForkLimit"
},
"mind": {
"$ref": "#/$defs/Slug"
},
"status": {
"description": "Which forks a list asks for; absent is both. `recorded` is a closed card.",
"type": "string",
"enum": [
"open",
"recorded"
]
},
"subject": {
"$ref": "#/$defs/Short"
}
@ -72,7 +87,10 @@
"required": [
"mind",
"subject",
"deciders"
"deciders",
"status",
"cursor",
"limit"
]
}
},
@ -185,6 +203,11 @@
"request_id"
]
},
"ForkLimit": {
"type": "integer",
"maximum": 50,
"minimum": 0
},
"RequestSchema": {
"description": "The request document's schema id; any other value fails to decode.",
"type": "string",

View file

@ -1,7 +1,7 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"title": "DeskResponse",
"description": "What the door answers one desk request with. `Answered` is a stored answer, refused ones\nincluded (a stale fork, a non-decider and the rest are stored), fresh or replayed alike.\n`NotStored` is a request that stored nothing. `Refused` is any other Mind refusal; an\n`Unavailable` is Bifrost's to turn into its 503, and Huginn writes no document for it.",
"description": "What the door answers one desk request with. `Answered` is a stored answer, refused ones\nincluded (a stale fork, a non-decider and the rest are stored), fresh or replayed alike.\n`Forks` and `Fork` are the reads: one page of entries, or one entry, with the head they were\nall read at. `NotStored` is a request that stored nothing (an unknown fork included).\n`Refused` is any other Mind refusal; an `Unavailable` is Bifrost's to turn into its 503, and\nHuginn writes no document for it.\n\nThe trust boundary of the reads (ruling `desk-read-visibility-whole-record`): `Forks` and `Fork`\ncarry the WHOLE record, every answer's note included, to whatever the peer-checked socket\ncarries. Huginn strips nothing and gates nothing; Bifrost applies visibility at its door (a\nstripped read serves `note: null`, CultLib `fork-card.md`). Huginn must therefore be reachable\nonly by Bifrost's peer-checked socket: the peer-uid check IS the whole authentication, and a\nsecond reader of that socket is a note leak.",
"oneOf": [
{
"type": "object",
@ -35,6 +35,58 @@
"Answered"
]
},
{
"type": "object",
"properties": {
"Forks": {
"type": "object",
"properties": {
"as_of": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"list": {
"$ref": "#/$defs/ForkList"
}
},
"required": [
"list",
"as_of"
]
}
},
"additionalProperties": false,
"required": [
"Forks"
]
},
{
"type": "object",
"properties": {
"Fork": {
"type": "object",
"properties": {
"as_of": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"entry": {
"$ref": "#/$defs/ForkEntry"
}
},
"required": [
"entry",
"as_of"
]
}
},
"additionalProperties": false,
"required": [
"Fork"
]
},
{
"type": "object",
"properties": {
@ -68,6 +120,21 @@
"gamecult.desk_answer.v0"
]
},
"CardSchema": {
"description": "The card document's schema id.",
"type": "string",
"enum": [
"gamecult.fork.card.v0"
]
},
"CardStatus": {
"description": "Whether a ruling closed the fork.",
"type": "string",
"enum": [
"open",
"closed"
]
},
"DeskAuthority": {
"type": "string",
"enum": [
@ -76,6 +143,115 @@
"Member"
]
},
"EntrySchema": {
"description": "The entry document's schema id.",
"type": "string",
"enum": [
"gamecult.fork.entry.v0"
]
},
"ForkCard": {
"description": "`gamecult.fork.card.v0`: closed, her field order.",
"type": "object",
"properties": {
"context": {
"type": "string"
},
"deciders": {
"type": "array",
"items": {
"type": "string"
}
},
"depends": {
"type": "array",
"items": {
"type": "string"
}
},
"flag": {
"type": "string"
},
"fork_id": {
"type": "string"
},
"leans": {
"type": "string"
},
"options": {
"type": "array",
"items": {
"$ref": "#/$defs/ForkOption"
}
},
"options_hash": {
"type": "string"
},
"order": {
"description": "The question's rank: unique, 0-based, in (admission ordinal, key) order among all the Mind's\nquestions (see `Mind::desk_forks`; it deviates from \"the admission ordinal\" only in being unique).",
"type": "integer",
"format": "uint64",
"minimum": 0
},
"raised_in": {
"description": "A typed reference, as the question holds it (`raised_in`).",
"type": "object",
"properties": {
"id": {
"type": "string"
},
"kind": {
"type": "string"
}
},
"additionalProperties": false,
"required": [
"kind",
"id"
]
},
"recommendation": {
"type": "string"
},
"ruled_from": {
"type": "array",
"items": {
"type": "string"
}
},
"ruling": {
"type": "string"
},
"schemaVersion": {
"$ref": "#/$defs/CardSchema"
},
"status": {
"$ref": "#/$defs/CardStatus"
},
"title": {
"type": "string"
}
},
"additionalProperties": false,
"required": [
"schemaVersion",
"fork_id",
"title",
"context",
"options",
"recommendation",
"depends",
"leans",
"flag",
"raised_in",
"deciders",
"status",
"ruling",
"ruled_from",
"order",
"options_hash"
]
},
"ForkDeskAnswer": {
"description": "`gamecult.desk_answer.v0`: one stored answer as the desk shows it, her field order,\n`request_id` last. Every value is a stored fact; `answer_id`, `of` and `retracts` are the\nrecord keys in her `n<seq>` spelling.",
"type": "object",
@ -158,6 +334,121 @@
"request_id"
]
},
"ForkEntry": {
"description": "`gamecult.fork.entry.v0`: a card and its record.",
"type": "object",
"properties": {
"card": {
"$ref": "#/$defs/ForkCard"
},
"record": {
"$ref": "#/$defs/ForkRecord"
},
"schemaVersion": {
"$ref": "#/$defs/EntrySchema"
}
},
"additionalProperties": false,
"required": [
"schemaVersion",
"card",
"record"
]
},
"ForkList": {
"description": "`gamecult.fork.list.v0`: a page of entries, the counts over the showable set and the stamp.",
"type": "object",
"properties": {
"forks": {
"type": "array",
"items": {
"$ref": "#/$defs/ForkEntry"
}
},
"mind": {
"type": "string"
},
"next_cursor": {
"type": "string",
"pattern": "^c[1-9][0-9]*$"
},
"open_count": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"recorded_count": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"schemaVersion": {
"$ref": "#/$defs/ListSchema"
},
"stamp": {
"type": "string"
}
},
"additionalProperties": false,
"required": [
"schemaVersion",
"mind",
"stamp",
"open_count",
"recorded_count",
"forks",
"next_cursor"
]
},
"ForkOption": {
"description": "One shown option.",
"type": "object",
"properties": {
"id": {
"type": "string"
},
"text": {
"type": "string"
}
},
"additionalProperties": false,
"required": [
"id",
"text"
]
},
"ForkRecord": {
"description": "`gamecult.fork.record.v0`: the answers in admission order and the derived counted answer.",
"type": "object",
"properties": {
"answers": {
"type": "array",
"items": {
"$ref": "#/$defs/ForkDeskAnswer"
}
},
"counted_answer_id": {
"type": "string"
},
"fork_id": {
"type": "string"
},
"options_hash": {
"type": "string"
},
"schemaVersion": {
"$ref": "#/$defs/RecordSchema"
}
},
"additionalProperties": false,
"required": [
"schemaVersion",
"fork_id",
"options_hash",
"counted_answer_id",
"answers"
]
},
"ForkRefusal": {
"description": "`gamecult.fork.refusal.v0`. `error` is a fixed sentence, never an echo of what was sent;\nfor `body_invalid` it names the field.",
"type": "object",
@ -188,6 +479,13 @@
"request_id_conflict"
]
},
"ListSchema": {
"description": "The list document's schema id.",
"type": "string",
"enum": [
"gamecult.fork.list.v0"
]
},
"MindRefusal": {
"description": "Every way a mind refuses. `Document` wraps the leaf's four (bounds,\nformats, key identity, foreign type); the rest are the organ's: store\nidentity, epoch, ownership, batch shape, references, and every cross-field\nand cross-document rule of admission.\n\nThe two sequenced kinds refuse in the same pair of shapes: an\n`OutOfSequence` when the writer's `sequence` is not the previous plus one,\nand `AlreadyResolved`/`AlreadyStewarded` when a record of that scope is\nstill in force.",
"oneOf": [
@ -1264,6 +1562,45 @@
"ResponseTooLarge"
]
},
{
"description": "A desk page whose first entry alone is past the page budget: `fork` is that fork, `order` its\nrank (a caller goes on past it with the cursor `c<order>`, which returns the forks below it),\n`bytes` what the page holding it would encode to (without the envelope) and `limit` the page\nbudget it exceeds. Raised by the mind's page fill, which is why it is not `ResponseTooLarge`.",
"type": "object",
"properties": {
"DeskForkTooLarge": {
"type": "object",
"properties": {
"bytes": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"fork": {
"type": "string"
},
"limit": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"order": {
"type": "integer",
"format": "uint64",
"minimum": 0
}
},
"required": [
"fork",
"order",
"bytes",
"limit"
]
}
},
"additionalProperties": false,
"required": [
"DeskForkTooLarge"
]
},
{
"description": "A selection the substrate or the organ's value door refused: `field`\nis the selection field the client sent, `value` the offending part.",
"type": "object",
@ -1497,6 +1834,13 @@
}
]
},
"RecordSchema": {
"description": "The record document's schema id.",
"type": "string",
"enum": [
"gamecult.fork.record.v0"
]
},
"RefusalSchema": {
"description": "The refusal document's schema id.",
"type": "string",

View file

@ -2059,6 +2059,45 @@
"ResponseTooLarge"
]
},
{
"description": "A desk page whose first entry alone is past the page budget: `fork` is that fork, `order` its\nrank (a caller goes on past it with the cursor `c<order>`, which returns the forks below it),\n`bytes` what the page holding it would encode to (without the envelope) and `limit` the page\nbudget it exceeds. Raised by the mind's page fill, which is why it is not `ResponseTooLarge`.",
"type": "object",
"properties": {
"DeskForkTooLarge": {
"type": "object",
"properties": {
"bytes": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"fork": {
"type": "string"
},
"limit": {
"type": "integer",
"format": "uint64",
"minimum": 0
},
"order": {
"type": "integer",
"format": "uint64",
"minimum": 0
}
},
"required": [
"fork",
"order",
"bytes",
"limit"
]
}
},
"additionalProperties": false,
"required": [
"DeskForkTooLarge"
]
},
{
"description": "A selection the substrate or the organ's value door refused: `field`\nis the selection field the client sent, `value` the offending part.",
"type": "object",