huginn import: the receiver's Self replays a moved campaign from the giver's mind
The verb walks the giver (Query only), plans with huginn_mind::plan, walks her own root to resume by what she holds, admits one batch per source receipt with imported_from, and ends with a verification walk. Settings::source builds the giver's client from explicit flags. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
f2cfdcbeb6
commit
5f67eb1188
5 changed files with 436 additions and 27 deletions
|
|
@ -39,6 +39,20 @@ impl Settings {
|
|||
self.client.as_ref().map_err(|detail| Trouble::Misconfigured(detail.clone()))
|
||||
}
|
||||
|
||||
/// A client for another mind, read from: `instance` at `endpoint` (`rudp://<ip>:<port>`), both
|
||||
/// given as arguments. It refuses this process's own instance before any call, so the one
|
||||
/// mind a verb writes is never also its source. Every request through it declares `instance`,
|
||||
/// so an endpoint that holds a different mind answers ForeignInstance.
|
||||
pub fn source(&self, instance: &str, endpoint: &str) -> Result<HuginnClient, Trouble> {
|
||||
let slug = Slug(instance.to_string());
|
||||
slug.validate_slug().map_err(|refusal| Trouble::invalid(format!("--from is not a valid instance name: {refusal:?}")))?;
|
||||
if self.instance.as_deref() == Some(instance) {
|
||||
return Err(Trouble::invalid(format!("--from {instance} is this process's own instance (EUREKA_INSTANCE)")));
|
||||
}
|
||||
let address = address(endpoint).ok_or_else(|| Trouble::invalid(format!("--from-endpoint is not rudp://<ip>:<port>: {endpoint}")))?;
|
||||
Ok(HuginnClient::new(address, slug, TIMEOUT))
|
||||
}
|
||||
|
||||
/// One wire call for the configured instance. The request is built from
|
||||
/// that instance, never from anything the caller sent. No retry.
|
||||
pub fn ask(&self, request: impl FnOnce(&Slug) -> HuginnMindRequest) -> Result<HuginnMindResponse, Trouble> {
|
||||
|
|
@ -78,9 +92,11 @@ fn client(instance: Result<Option<String>, String>, endpoint: Result<Option<Stri
|
|||
let slug = Slug(instance);
|
||||
slug.validate_slug().map_err(|refusal| format!("EUREKA_INSTANCE is not a valid instance name: {refusal:?}"))?;
|
||||
let endpoint = endpoint?.ok_or("HUGINN_ENDPOINT is not set")?;
|
||||
let address = endpoint
|
||||
.strip_prefix("rudp://")
|
||||
.and_then(|address| address.parse::<SocketAddr>().ok())
|
||||
.ok_or_else(|| format!("HUGINN_ENDPOINT is not rudp://<ip>:<port>: {endpoint}"))?;
|
||||
let address = address(&endpoint).ok_or_else(|| format!("HUGINN_ENDPOINT is not rudp://<ip>:<port>: {endpoint}"))?;
|
||||
Ok(HuginnClient::new(address, slug, TIMEOUT))
|
||||
}
|
||||
|
||||
/// The socket address of a `rudp://<ip>:<port>` endpoint.
|
||||
fn address(endpoint: &str) -> Option<SocketAddr> {
|
||||
endpoint.strip_prefix("rudp://").and_then(|address| address.parse::<SocketAddr>().ok())
|
||||
}
|
||||
|
|
|
|||
379
crates/eureka-state/src/bin/huginn/import.rs
Normal file
379
crates/eureka-state/src/bin/huginn/import.rs
Normal file
|
|
@ -0,0 +1,379 @@
|
|||
//! `huginn import <campaign> --from <instance> --from-endpoint <rudp>`: the receiver's Self replays
|
||||
//! a moved campaign's receipts from the giver's mind into her own.
|
||||
//!
|
||||
//! The verb decides only what to send and in what order. `huginn_mind::plan` lays the replay out
|
||||
//! from one cursor walk of the giver; the receiver's walk of her own root says what she already
|
||||
//! holds, so a rerun after any stop resumes at the first unheld batch. Every write is the
|
||||
//! receiver daemon's `Admit`, one per source receipt, and her admission decides it. The giver is
|
||||
//! sent `Query` alone. Nothing is stored here: progress is what the receiver holds, and one
|
||||
//! verification walk of her root at the end compares every planned document.
|
||||
|
||||
use std::collections::{BTreeMap, BTreeSet};
|
||||
|
||||
use cultnet_rs::{LIMIT_MAX, PROJECTION_DOCUMENT, Selection};
|
||||
use eureka_state::hold;
|
||||
use huginn_mind::eureka_pipeline::{PipelineDocument, PipelineRef, Short, Slug};
|
||||
use huginn_mind::{
|
||||
Faculty, HuginnMindRequest, HuginnMindResponse, ImportBatch, ImportFault, ImportPlan, PipelineAdmissionBatch, PipelineAdmissionOutcome,
|
||||
PipelineDocumentSummary, PipelineDocumentView, PipelinePageItems, PipelineProvenance, PipelineSelectionPage, plan,
|
||||
};
|
||||
|
||||
use crate::config::{Settings, TOOL};
|
||||
use crate::trouble::Trouble;
|
||||
use crate::{differences, say, to_value, unexpected};
|
||||
|
||||
/// The arguments of the verb, as given.
|
||||
pub struct Flags {
|
||||
pub campaign: String,
|
||||
pub from: String,
|
||||
pub from_endpoint: String,
|
||||
pub faculty: Faculty,
|
||||
pub agent: String,
|
||||
pub session: String,
|
||||
}
|
||||
|
||||
/// Why the verb stops: the mind said no (exit 1, the text printed) or a trouble (its own exit).
|
||||
enum Stop {
|
||||
No(String),
|
||||
Trouble(Trouble),
|
||||
}
|
||||
|
||||
impl From<Trouble> for Stop {
|
||||
fn from(trouble: Trouble) -> Self {
|
||||
Self::Trouble(trouble)
|
||||
}
|
||||
}
|
||||
|
||||
type Step<T> = Result<T, Stop>;
|
||||
|
||||
pub fn run(settings: &Settings, flags: Flags) -> Result<u8, Trouble> {
|
||||
match import(settings, flags) {
|
||||
Ok(()) => Ok(0),
|
||||
Err(Stop::No(text)) => {
|
||||
say(text);
|
||||
Ok(1)
|
||||
}
|
||||
Err(Stop::Trouble(trouble)) => Err(trouble),
|
||||
}
|
||||
}
|
||||
|
||||
fn import(settings: &Settings, flags: Flags) -> Step<()> {
|
||||
let campaign = Slug(flags.campaign.clone());
|
||||
campaign.validate_slug().map_err(|refusal| Trouble::invalid(format!("the campaign is not a valid name: {refusal:?}")))?;
|
||||
let giver = settings.source(&flags.from, &flags.from_endpoint)?;
|
||||
let receiver = settings.client()?.instance().clone();
|
||||
|
||||
let views = documents(&|selection| {
|
||||
giver.call(HuginnMindRequest::Query { instance: giver.instance().clone(), selection, semantic: None }).map_err(Trouble::from)
|
||||
}, &campaign)?;
|
||||
let plan = plan(&campaign, giver.instance(), &receiver, views).map_err(fault)?;
|
||||
|
||||
let ask_receiver = |selection: Selection| {
|
||||
settings.ask(|instance| HuginnMindRequest::Query { instance: instance.clone(), selection, semantic: None })
|
||||
};
|
||||
let held = resume(&plan, &headers(&ask_receiver, &campaign)?, giver.instance())?;
|
||||
let (receipts, total) = (plan.batches.len(), plan.batches.iter().map(|batch| batch.documents.len()).sum::<usize>());
|
||||
say(format_args!(
|
||||
"plan {campaign} from {giver}: {receipts} receipts {total} documents, {held} held, moved by {moved}",
|
||||
campaign = flags.campaign,
|
||||
giver = flags.from,
|
||||
held = held.iter().filter(|held| **held).count(),
|
||||
moved = plan.moved.id.0,
|
||||
));
|
||||
|
||||
for (index, (batch, held)) in plan.batches.iter().zip(&held).enumerate() {
|
||||
if *held {
|
||||
continue;
|
||||
}
|
||||
let external = unheld(&ask_receiver, &batch.foreign)?;
|
||||
let count = external.len();
|
||||
send(settings, &flags, batch, external)?;
|
||||
say(format_args!("imported ordinal {} {} documents {count} external ({}/{receipts})", batch.source.ordinal, batch.documents.len(), index + 1));
|
||||
}
|
||||
|
||||
verify(&documents(&|selection| ask_receiver(selection), &campaign)?, &plan, giver.instance())?;
|
||||
say(format_args!("imported {} from {}: {receipts} receipts {total} documents, verified", flags.campaign, flags.from));
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// A planner fault as the verb ends: the mind's no for who the campaign moved to, an integrity
|
||||
/// trouble for a history that cannot be replayed whole.
|
||||
fn fault(fault: ImportFault) -> Stop {
|
||||
match fault {
|
||||
ImportFault::NotMoved | ImportFault::MovedElsewhere { .. } | ImportFault::SourceIsReceiver => Stop::No(format!("Refused {fault:?}")),
|
||||
ImportFault::BatchTooLarge { .. } | ImportFault::ExternalsTooMany { .. } | ImportFault::Unordered { .. } => {
|
||||
Stop::Trouble(Trouble::Integrity(format!("{fault:?}; nothing was written")))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One batch, admitted as the importer's own with the source's receipt in `imported_from`.
|
||||
fn send(settings: &Settings, flags: &Flags, batch: &ImportBatch, external: Vec<PipelineRef>) -> Step<()> {
|
||||
let mut source = batch.source.clone();
|
||||
source.external = external;
|
||||
let provenance = PipelineProvenance {
|
||||
faculty: flags.faculty,
|
||||
agent: Short(flags.agent.clone()),
|
||||
session: Short(flags.session.clone()),
|
||||
tool: Short(TOOL.to_string()),
|
||||
imported_from: Some(source),
|
||||
};
|
||||
let documents = batch.documents.iter().map(|(_, document)| document.clone()).collect();
|
||||
let ordinal = batch.source.ordinal;
|
||||
match settings.ask(|instance| HuginnMindRequest::Admit(PipelineAdmissionBatch { instance: instance.clone(), provenance, documents }))? {
|
||||
HuginnMindResponse::Admit(PipelineAdmissionOutcome::Committed { .. } | PipelineAdmissionOutcome::AlreadyAdmitted { .. }) => Ok(()),
|
||||
HuginnMindResponse::Admit(PipelineAdmissionOutcome::Refused(refusal)) | HuginnMindResponse::Refused(refusal) => {
|
||||
Err(Stop::No(format!("Refused {refusal:?} at source ordinal {ordinal}")))
|
||||
}
|
||||
HuginnMindResponse::Admit(PipelineAdmissionOutcome::Conflict { identities }) => Err(Stop::No(format!(
|
||||
"Conflict at source ordinal {ordinal}: {}",
|
||||
identities.iter().map(|identity| identity.id.0.as_str()).collect::<Vec<_>>().join(", ")
|
||||
))),
|
||||
other => Err(unexpected(&other).into()),
|
||||
}
|
||||
}
|
||||
|
||||
/// The foreign referents the receiver does not hold: exactly the ones the batch declares external,
|
||||
/// since declaring one she holds is refused.
|
||||
fn unheld(ask: &dyn Fn(Selection) -> Result<HuginnMindResponse, Trouble>, foreign: &[PipelineRef]) -> Step<Vec<PipelineRef>> {
|
||||
let mut held = BTreeSet::new();
|
||||
for chunk in foreign.chunks(LIMIT_MAX as usize) {
|
||||
let selection = Selection {
|
||||
keys: Some(chunk.iter().map(|referent| referent.id.0.clone()).collect()),
|
||||
limit: Some(chunk.len() as u32),
|
||||
..Selection::default()
|
||||
};
|
||||
for page in pages(ask, selection)? {
|
||||
let PipelinePageItems::Headers(found) = page.items else { return Err(Trouble::Internal("a header query answered documents".into()).into()) };
|
||||
held.extend(found.into_iter().map(|header| header.id.id.0));
|
||||
}
|
||||
}
|
||||
Ok(foreign.iter().filter(|referent| !held.contains(&referent.id.0)).cloned().collect())
|
||||
}
|
||||
|
||||
/// Every page of a selection, to the end, each as of the first page's snapshot.
|
||||
fn pages(ask: &dyn Fn(Selection) -> Result<HuginnMindResponse, Trouble>, mut selection: Selection) -> Step<Vec<PipelineSelectionPage>> {
|
||||
let mut found = Vec::new();
|
||||
loop {
|
||||
match ask(selection.clone())? {
|
||||
HuginnMindResponse::Query(page) => {
|
||||
let next = page.next.clone();
|
||||
found.push(page);
|
||||
match next {
|
||||
Some(next) => selection.cursor = Some(next),
|
||||
None => return Ok(found),
|
||||
}
|
||||
}
|
||||
HuginnMindResponse::Refused(refusal) => return Err(Stop::No(format!("Refused {refusal:?}"))),
|
||||
other => return Err(unexpected(&other).into()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn root_walk(campaign: &Slug, projection: &str) -> Selection {
|
||||
Selection {
|
||||
projection: projection.to_string(),
|
||||
limit: Some(LIMIT_MAX),
|
||||
..hold::any_of(Selection::default(), "root", &[&campaign.0])
|
||||
}
|
||||
}
|
||||
|
||||
/// Every document under the campaign's root, whole.
|
||||
fn documents(ask: &dyn Fn(Selection) -> Result<HuginnMindResponse, Trouble>, campaign: &Slug) -> Step<Vec<PipelineDocumentView>> {
|
||||
let mut views = Vec::new();
|
||||
for page in pages(ask, root_walk(campaign, PROJECTION_DOCUMENT))? {
|
||||
let PipelinePageItems::Documents(found) = page.items else { return Err(Trouble::Internal("a document query answered headers".into()).into()) };
|
||||
views.extend(found);
|
||||
}
|
||||
Ok(views)
|
||||
}
|
||||
|
||||
/// Every document under the campaign's root, as a header.
|
||||
fn headers(ask: &dyn Fn(Selection) -> Result<HuginnMindResponse, Trouble>, campaign: &Slug) -> Step<Vec<PipelineDocumentSummary>> {
|
||||
let mut headers = Vec::new();
|
||||
for page in pages(ask, root_walk(campaign, &Selection::default().projection))? {
|
||||
let PipelinePageItems::Headers(found) = page.items else { return Err(Trouble::Internal("a header query answered documents".into()).into()) };
|
||||
headers.extend(found);
|
||||
}
|
||||
Ok(headers)
|
||||
}
|
||||
|
||||
/// Which batches the receiver already holds whole. A document under the root that is not one of
|
||||
/// the plan's, imported from the plan's receipt of the source, is her own record and refuses the
|
||||
/// import; a batch held in part cannot be resumed, only repaired.
|
||||
fn resume(plan: &ImportPlan, root: &[PipelineDocumentSummary], source: &Slug) -> Step<Vec<bool>> {
|
||||
let planned: BTreeMap<&str, (usize, &str)> = plan
|
||||
.batches
|
||||
.iter()
|
||||
.enumerate()
|
||||
.flat_map(|(index, batch)| batch.documents.iter().map(move |(id, _)| (id.id.0.as_str(), (index, batch.source.receipt_id.0.as_str()))))
|
||||
.collect();
|
||||
let mut held: Vec<BTreeSet<&str>> = vec![BTreeSet::new(); plan.batches.len()];
|
||||
let mut own = Vec::new();
|
||||
for header in root {
|
||||
let key = header.id.id.0.as_str();
|
||||
let imported = header.admission.provenance.imported_from.as_ref();
|
||||
match planned.get(key) {
|
||||
Some((index, receipt)) if imported.is_some_and(|from| from.instance == *source && from.receipt_id.0 == *receipt) => {
|
||||
held[*index].insert(key);
|
||||
}
|
||||
_ => own.push(key),
|
||||
}
|
||||
}
|
||||
if !own.is_empty() {
|
||||
return Err(Stop::No(format!("Refused the receiver already holds records under the root that are not imported from this source: {}; nothing was written", own.join(", "))));
|
||||
}
|
||||
plan.batches
|
||||
.iter()
|
||||
.zip(&held)
|
||||
.map(|(batch, held)| match held.len() {
|
||||
0 => Ok(false),
|
||||
count if count == batch.documents.len() => Ok(true),
|
||||
count => Err(Stop::Trouble(Trouble::Integrity(format!(
|
||||
"source ordinal {} is held in part ({count} of {} documents); repair it by hand, a rerun cannot",
|
||||
batch.source.ordinal,
|
||||
batch.documents.len()
|
||||
)))),
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The receiver's root against the plan: every planned document equal and imported from its
|
||||
/// source receipt, and nothing else under the root.
|
||||
fn verify(root: &[PipelineDocumentView], plan: &ImportPlan, source: &Slug) -> Step<()> {
|
||||
let held: BTreeMap<&str, &PipelineDocumentView> = root.iter().map(|view| (view.id.id.0.as_str(), view)).collect();
|
||||
let mut differing = Vec::new();
|
||||
for batch in &plan.batches {
|
||||
for (id, document) in &batch.documents {
|
||||
let key = id.id.0.as_str();
|
||||
let Some(view) = held.get(key) else {
|
||||
differing.push(format!("{key}: missing from the receiver"));
|
||||
continue;
|
||||
};
|
||||
let mut what = Vec::new();
|
||||
differences(&to_value(document), &to_value(&view.document), "", &mut what);
|
||||
let imported = view.admission.provenance.imported_from.as_ref();
|
||||
if imported.is_none_or(|from| from.instance != *source || from.receipt_id != batch.source.receipt_id) {
|
||||
what.push("imported_from".to_string());
|
||||
}
|
||||
if !what.is_empty() {
|
||||
differing.push(format!("{key}: {}", if what.iter().all(|path| path.is_empty()) { "the document".to_string() } else { what.join(", ") }));
|
||||
}
|
||||
}
|
||||
}
|
||||
let planned: BTreeSet<&str> = plan.batches.iter().flat_map(|batch| batch.documents.iter().map(|(id, _)| id.id.0.as_str())).collect();
|
||||
differing.extend(held.keys().filter(|key| !planned.contains(*key)).map(|key| format!("{key}: not in the plan")));
|
||||
if differing.is_empty() {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(Stop::Trouble(Trouble::ReadBackDiffers(format!("the receiver's root differs from the giver's: {}", differing.join("; ")))))
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use huginn_mind::eureka_pipeline::{Date, PipelineKind, PipelineRuling, RulingAuthority};
|
||||
use huginn_mind::{AdmissionFacts, ImportSource, PipelineFacts, PipelineStatusSummary};
|
||||
|
||||
use super::*;
|
||||
|
||||
fn slug(text: &str) -> Slug {
|
||||
Slug(text.to_string())
|
||||
}
|
||||
|
||||
fn id(local: &str) -> PipelineRef {
|
||||
PipelineRef { kind: PipelineKind::Ruling, id: Short(format!("c:ruling:{local}")) }
|
||||
}
|
||||
|
||||
fn batch(ordinal: u64, locals: &[&str]) -> ImportBatch {
|
||||
let source = ImportSource {
|
||||
instance: slug("eureka"),
|
||||
receipt_id: Short(format!("receipt-{ordinal}")),
|
||||
ordinal,
|
||||
committed_at: Short("2026-11-08T16:00:00Z".into()),
|
||||
faculty: Faculty::Hands,
|
||||
agent: Short("a".into()),
|
||||
session: Short("s".into()),
|
||||
tool: Short("huginn".into()),
|
||||
external: vec![],
|
||||
};
|
||||
ImportBatch { source, documents: locals.iter().map(|local| (id(local), question(local))).collect(), foreign: vec![] }
|
||||
}
|
||||
|
||||
fn question(local: &str) -> PipelineDocument {
|
||||
PipelineDocument::Ruling(PipelineRuling {
|
||||
campaign: slug("c"),
|
||||
label: local.into(),
|
||||
title: "A ruling".into(),
|
||||
answers: None,
|
||||
choice: None,
|
||||
ruling: "It stands.".into(),
|
||||
operator_quote: None,
|
||||
ruled_on: Date("2026-11-08".into()),
|
||||
precedents: vec![],
|
||||
authority: RulingAuthority::Standing,
|
||||
})
|
||||
}
|
||||
|
||||
fn plan_of(batches: Vec<ImportBatch>) -> ImportPlan {
|
||||
ImportPlan { moved: id("moved"), batches }
|
||||
}
|
||||
|
||||
fn header(local: &str, from: Option<(&str, u64)>) -> PipelineDocumentSummary {
|
||||
let imported_from = from.map(|(instance, ordinal)| ImportSource { instance: slug(instance), ..batch(ordinal, &[]).source });
|
||||
PipelineDocumentSummary {
|
||||
id: id(local),
|
||||
admission: AdmissionFacts {
|
||||
receipt_id: "mine".into(),
|
||||
admitted_at: "2026-11-08T17:00:00Z".into(),
|
||||
provenance: PipelineProvenance { faculty: Faculty::Hands, agent: Short("me".into()), session: Short("s".into()), tool: Short("huginn".into()), imported_from },
|
||||
ordinal: 1,
|
||||
},
|
||||
status: PipelineStatusSummary::InForce,
|
||||
facts: PipelineFacts::Campaign { title: "t".into(), repos: vec![] },
|
||||
}
|
||||
}
|
||||
|
||||
fn two() -> ImportPlan {
|
||||
plan_of(vec![batch(3, &["a", "b"]), batch(5, &["c"])])
|
||||
}
|
||||
|
||||
fn stopped(step: Step<Vec<bool>>) -> String {
|
||||
match step {
|
||||
Err(Stop::No(text)) => format!("no: {text}"),
|
||||
Err(Stop::Trouble(trouble)) => format!("{trouble}"),
|
||||
Ok(held) => panic!("{held:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_rerun_resumes_at_the_first_unheld_batch() {
|
||||
let source = slug("eureka");
|
||||
assert_eq!(resume(&two(), &[], &source).ok().unwrap(), vec![false, false]);
|
||||
let first = [header("a", Some(("eureka", 3))), header("b", Some(("eureka", 3)))];
|
||||
assert_eq!(resume(&two(), &first, &source).ok().unwrap(), vec![true, false], "the held receipt is skipped");
|
||||
let both = [first.as_slice(), &[header("c", Some(("eureka", 5)))]].concat();
|
||||
assert_eq!(resume(&two(), &both, &source).ok().unwrap(), vec![true, true]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_receipt_held_in_part_is_an_integrity_fault() {
|
||||
let one_of_two = [header("a", Some(("eureka", 3)))];
|
||||
let said = stopped(resume(&two(), &one_of_two, &slug("eureka")));
|
||||
assert!(said.starts_with("Integrity: source ordinal 3 is held in part (1 of 2"), "{said}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_receivers_own_records_under_the_root_refuse_the_import() {
|
||||
let source = slug("eureka");
|
||||
for own in [
|
||||
header("zz", None),
|
||||
header("a", None),
|
||||
header("a", Some(("kiko", 3))),
|
||||
header("a", Some(("eureka", 5))),
|
||||
] {
|
||||
let said = stopped(resume(&two(), &[own], &source));
|
||||
assert!(said.starts_with("no: Refused the receiver already holds records under the root"), "{said}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -13,11 +13,13 @@
|
|||
//! 1 the mind said no (a refusal, a conflict, an absent document). Troubles go
|
||||
//! to stderr as `<Code>: <detail>`: 2 the call is wrong, 3 Unavailable, 4 no
|
||||
//! answer and a retry cannot help, 5 the read-back after a committed admission
|
||||
//! differs from what was sent, 6 `admit` held the batch once (`admit_hold`).
|
||||
//! differs from what was sent or the verification walk of `import` finds a difference, 6 `admit` held the batch once (`admit_hold`). `import` also
|
||||
//! exits 1 when the campaign is not moved to this mind or this mind holds records of its own under it, and 4 when the history cannot be replayed whole.
|
||||
|
||||
mod admit_hold;
|
||||
mod config;
|
||||
mod door;
|
||||
mod import;
|
||||
mod near;
|
||||
mod recall;
|
||||
mod recipes;
|
||||
|
|
@ -60,8 +62,10 @@ The file is a JSON array of 1 to 64 documents, each {\"kind\": <kind>, \"value\"
|
|||
Every key the schema does not know is refused (exit 2) with its path; nothing is sent. \
|
||||
After Committed or AlreadyAdmitted the batch is read back and compared with what was sent: `read back <n> equal`, or exit 5 with the ids and field paths that differ. A batch whose rulings or questions have in-force neighbours is held once: exit 6, the neighbours printed with a token, nothing sent. Run the same command with --seen <token> to admit it; a batch or --agent that changed needs a new token (a wrong one is exit 2).";
|
||||
|
||||
const IMPORT_HELP: &str = "Reads the giver's mind (it sends Query alone) and admits its receipts for the campaign into this one, one batch per source receipt, in order, each naming the giver's receipt in imported_from. It refuses unless the campaign is moved to this mind (exit 1), and when this mind already holds records under the campaign that it did not import from the giver (exit 1). A rerun after any stop resumes at the first batch this mind does not hold; it prints `imported <campaign> from <instance>: ... verified` only after comparing every document it holds with the giver's (exit 5 if one differs). Until it prints imported, this mind writes nothing else under the campaign; a write that collides stops the import with the refusal (exit 1). Exit 3 Unavailable (rerun), 4 the history cannot be replayed whole (a receipt over 64 documents, too many external referents, a receipt held in part). Only a Self run can run it: a Persona's traffic to a peer's huginn port is fenced.";
|
||||
|
||||
#[derive(Parser, Debug)]
|
||||
#[command(name = "huginn", about = "Huginn's mind: whoami, view, query, recipes, persona, admit, recall, schema", arg_required_else_help = true)]
|
||||
#[command(name = "huginn", about = "Huginn's mind: whoami, view, query, recipes, persona, admit, import, recall, schema", arg_required_else_help = true)]
|
||||
struct Cli {
|
||||
#[command(subcommand)]
|
||||
command: Command,
|
||||
|
|
@ -128,6 +132,26 @@ enum Command {
|
|||
},
|
||||
/// The JSON Schema of a document kind or of a selection; no name lists them.
|
||||
Schema { name: Option<String> },
|
||||
/// Replay a moved campaign's history from the giver's mind into this one.
|
||||
#[command(after_help = IMPORT_HELP)]
|
||||
Import {
|
||||
/// The campaign's name: the root of its ids.
|
||||
campaign: String,
|
||||
/// The giver's instance.
|
||||
#[arg(long = "from")]
|
||||
from: String,
|
||||
/// The giver's daemon, rudp://<ip>:<port>.
|
||||
#[arg(long)]
|
||||
from_endpoint: String,
|
||||
#[arg(long, value_parser = faculty)]
|
||||
faculty: Faculty,
|
||||
/// The importing agent's name.
|
||||
#[arg(long)]
|
||||
agent: String,
|
||||
/// The importing session.
|
||||
#[arg(long)]
|
||||
session: String,
|
||||
},
|
||||
/// Recall into a dispatched brief.
|
||||
Recall {
|
||||
#[command(subcommand)]
|
||||
|
|
@ -260,6 +284,9 @@ fn run(settings: &Settings, command: Command) -> Result<u8, Trouble> {
|
|||
query(settings, door::selection_value(&ask.selection)?, ask.semantic, ask.page.cursor, ask.page.documents)
|
||||
}
|
||||
Command::Admit { faculty, agent, session, seen, file } => admit(settings, faculty, agent, session, seen, &file),
|
||||
Command::Import { campaign, from, from_endpoint, faculty, agent, session } => {
|
||||
import::run(settings, import::Flags { campaign, from, from_endpoint, faculty, agent, session })
|
||||
}
|
||||
Command::Persona { command: PersonaCommand::Get } => persona_get(settings),
|
||||
Command::Persona { command: PersonaCommand::Put { faculty, agent, session, expected_updated_at, file } } => {
|
||||
persona_put(settings, faculty, agent, session, expected_updated_at, &file)
|
||||
|
|
|
|||
|
|
@ -2,7 +2,8 @@
|
|||
//!
|
||||
//! Exit codes: 2 the call is wrong (usage, InvalidInput, TooLarge,
|
||||
//! Misconfigured); 3 Unavailable (retry once); 4 no answer and a retry cannot
|
||||
//! help (Rejected, Unencodable, Internal); 5 ReadBackDiffers. Exit 0 and 1 are
|
||||
//! help (Rejected, Unencodable, Internal, Integrity); 5 ReadBackDiffers, the read-back after an
|
||||
//! admission or the verification walk of an import. Exit 0 and 1 are
|
||||
//! answers (the mind said yes, the mind said no) and are not troubles.
|
||||
|
||||
use std::fmt;
|
||||
|
|
@ -19,6 +20,8 @@ pub enum Trouble {
|
|||
Rejected(String),
|
||||
Unencodable(String),
|
||||
Internal(String),
|
||||
/// What an import found cannot be replayed whole, however often it is rerun.
|
||||
Integrity(String),
|
||||
ReadBackDiffers(String),
|
||||
}
|
||||
|
||||
|
|
@ -32,6 +35,7 @@ impl Trouble {
|
|||
Self::Rejected(_) => "Rejected",
|
||||
Self::Unencodable(_) => "Unencodable",
|
||||
Self::Internal(_) => "Internal",
|
||||
Self::Integrity(_) => "Integrity",
|
||||
Self::ReadBackDiffers(_) => "ReadBackDiffers",
|
||||
}
|
||||
}
|
||||
|
|
@ -40,7 +44,7 @@ impl Trouble {
|
|||
match self {
|
||||
Self::Misconfigured(_) | Self::InvalidInput(_) | Self::TooLarge { .. } => 2,
|
||||
Self::Unavailable(_) => 3,
|
||||
Self::Rejected(_) | Self::Unencodable(_) | Self::Internal(_) => 4,
|
||||
Self::Rejected(_) | Self::Unencodable(_) | Self::Internal(_) | Self::Integrity(_) => 4,
|
||||
Self::ReadBackDiffers(_) => 5,
|
||||
}
|
||||
}
|
||||
|
|
@ -60,6 +64,7 @@ impl fmt::Display for Trouble {
|
|||
| Self::Rejected(detail)
|
||||
| Self::Unencodable(detail)
|
||||
| Self::Internal(detail)
|
||||
| Self::Integrity(detail)
|
||||
| Self::ReadBackDiffers(detail) => f.write_str(detail),
|
||||
Self::TooLarge { bytes, limit } => {
|
||||
write!(f, "the request encodes to {bytes} bytes and one send carries at most {limit}")
|
||||
|
|
|
|||
|
|
@ -89,7 +89,7 @@ pub fn plan(campaign: &Slug, source: &Slug, receiver: &Slug, views: Vec<Pipeline
|
|||
let in_batch: BTreeSet<&str> = group.iter().map(|view| view.id.id.0.as_str()).collect();
|
||||
let mut foreign: BTreeMap<String, PipelineRef> = BTreeMap::new();
|
||||
for view in &group {
|
||||
for (_, target) in cited(&view.document, receiver) {
|
||||
for (_, target) in citations(&view.document) {
|
||||
let key = target.id.0.as_str();
|
||||
if in_batch.contains(key) || written.contains(key) {
|
||||
continue;
|
||||
|
|
@ -124,15 +124,6 @@ pub fn plan(campaign: &Slug, source: &Slug, receiver: &Slug, views: Vec<Pipeline
|
|||
Ok(ImportPlan { moved, batches })
|
||||
}
|
||||
|
||||
/// What admission requires a document's referents to be. A hand-off cites its documents on its
|
||||
/// source side only (`admission::references`), so one from another mind cites nothing here.
|
||||
fn cited(document: &PipelineDocument, receiver: &Slug) -> Vec<(crate::CitationRole, PipelineRef)> {
|
||||
match document {
|
||||
PipelineDocument::HandOff(hand_off) if hand_off.from_instance != *receiver => Vec::new(),
|
||||
_ => citations(document),
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use eureka_pipeline::{PipelineKind as K, pipeline_key};
|
||||
|
|
@ -272,13 +263,4 @@ mod tests {
|
|||
views[1] = view(PipelineDocument::CutSpec(spec), 6);
|
||||
assert_eq!(plan_of(views).unwrap().batches[0].foreign.len(), IMPORT_EXTERNAL_MAX, "the limit itself is declarable");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn plan_ignores_a_hand_off_cited_from_another_mind() {
|
||||
let q9 = id("question", "Q9");
|
||||
let from_libby = view(hand_off("libby", RECEIVER, REPO, &[&q9]), 3);
|
||||
assert!(plan_of(vec![moved(RECEIVER, 20), from_libby]).unwrap().batches[0].foreign.is_empty(), "the receiving side cites nothing");
|
||||
let from_receiver = view(hand_off(RECEIVER, "libby", REPO, &[&q9]), 3);
|
||||
assert_eq!(plan_of(vec![moved(RECEIVER, 20), from_receiver]), Err(ImportFault::Unordered { ordinal: 3, id: s(&q9) }));
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue