Merge pull request 'Import: resolution of a window follow-up or idea is inert' (#23) from eureka/import-inert-resolutions into main
This commit is contained in:
commit
563b6f3bf4
4 changed files with 122 additions and 18 deletions
|
|
@ -15,7 +15,7 @@ use eureka_state::hold;
|
|||
use huginn_mind::eureka_pipeline::{PipelineDocument, PipelineKind, PipelineRef, Short, Slug};
|
||||
use huginn_mind::{
|
||||
AdmissionFacts, Faculty, HuginnMindRequest, HuginnMindResponse, ImportBatch, ImportFault, ImportPlan, PipelineAdmissionBatch, PipelineAdmissionOutcome,
|
||||
PipelineDocumentSummary, PipelineDocumentView, PipelinePageItems, PipelineProvenance, PipelineSelectionPage, inert_in_window, plan,
|
||||
PipelineDocumentSummary, PipelineDocumentView, PipelineFacts, PipelinePageItems, PipelineProvenance, PipelineSelectionPage, inert_in_window, plan,
|
||||
};
|
||||
|
||||
use crate::config::{Settings, TOOL};
|
||||
|
|
@ -73,7 +73,7 @@ fn import(settings: &Settings, flags: Flags) -> Step<()> {
|
|||
};
|
||||
let root = headers(&ask_receiver, &campaign)?;
|
||||
let held = resume(&plan, &root, giver.instance())?;
|
||||
let window = window_writes(root.iter().map(|header| (header.id.kind, header.id.id.0.as_str(), &header.admission)), &planned_keys(&plan), giver.instance());
|
||||
let window = window_writes(root.iter().map(|header| (header.id.kind, header.id.id.0.as_str(), &header.admission, resolved_by_facts(&header.facts))), &planned_keys(&plan), giver.instance());
|
||||
if !window.is_empty() {
|
||||
say(format_args!("left in place, admitted by the receiver after the import began: {}", window.iter().copied().collect::<Vec<_>>().join(", ")));
|
||||
}
|
||||
|
|
@ -207,14 +207,32 @@ fn headers(ask: &dyn Fn(Selection) -> Result<HuginnMindResponse, Trouble>, campa
|
|||
/// from this source and are no planned document: writes in the import window (ruling
|
||||
/// `import-window-writes-ruled`: the import stops at a collision, and only a collision). They are
|
||||
/// left where they are; they cannot be a planned document's key, and the admission of each later
|
||||
/// batch still refuses any replay that meets one.
|
||||
fn window_writes<'a>(root: impl Iterator<Item = (PipelineKind, &'a str, &'a AdmissionFacts)> + Clone, planned: &BTreeSet<&str>, source: &Slug) -> BTreeSet<&'a str> {
|
||||
let began = root.clone().filter(|(_, _, admission)| admission.provenance.imported_from.as_ref().is_some_and(|from| from.instance == *source)).map(|(_, _, admission)| admission.ordinal).min();
|
||||
root.filter(|(kind, key, admission)| inert_in_window(*kind) && admission.provenance.imported_from.is_none() && !planned.contains(key) && began.is_some_and(|began| admission.ordinal > began))
|
||||
.map(|(_, key, _)| key)
|
||||
/// batch still refuses any replay that meets one. A record whose key a later batch carries is no window write: the
|
||||
/// tool keeps it planned and refuses the resume, and the daemon, which cannot see the later batch,
|
||||
/// refuses that batch as an identity collision when it arrives.
|
||||
fn window_writes<'a>(root: impl Iterator<Item = (PipelineKind, &'a str, &'a AdmissionFacts, Option<&'a PipelineRef>)> + Clone, planned: &BTreeSet<&str>, source: &Slug) -> BTreeSet<&'a str> {
|
||||
let began = root.clone().filter(|(_, _, admission, _)| admission.provenance.imported_from.as_ref().is_some_and(|from| from.instance == *source)).map(|(_, _, admission, _)| admission.ordinal).min();
|
||||
root.filter(|(kind, key, admission, subject)| inert_in_window(*kind, subject.map(|subject| (subject.kind, subject.id.0.as_str())), |id| planned.contains(id)) && admission.provenance.imported_from.is_none() && !planned.contains(key) && began.is_some_and(|began| admission.ordinal > began))
|
||||
.map(|(_, key, _, _)| key)
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// What a resolution resolves, from a header.
|
||||
fn resolved_by_facts(facts: &PipelineFacts) -> Option<&PipelineRef> {
|
||||
match facts {
|
||||
PipelineFacts::Resolution { subject, .. } => Some(subject),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// What a resolution resolves, from a whole document.
|
||||
fn resolved_by_document(document: &PipelineDocument) -> Option<&PipelineRef> {
|
||||
match document {
|
||||
PipelineDocument::Resolution(resolution) => Some(&resolution.subject),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// The plan's document keys.
|
||||
fn planned_keys(plan: &ImportPlan) -> BTreeSet<&str> {
|
||||
plan.batches.iter().flat_map(|batch| batch.documents.iter().map(|(id, _)| id.id.0.as_str())).collect()
|
||||
|
|
@ -233,7 +251,7 @@ fn resume(plan: &ImportPlan, root: &[PipelineDocumentSummary], source: &Slug) ->
|
|||
.collect();
|
||||
let mut held: Vec<BTreeSet<&str>> = vec![BTreeSet::new(); plan.batches.len()];
|
||||
let mut own = Vec::new();
|
||||
let window = window_writes(root.iter().map(|header| (header.id.kind, header.id.id.0.as_str(), &header.admission)), &planned_keys(plan), source);
|
||||
let window = window_writes(root.iter().map(|header| (header.id.kind, header.id.id.0.as_str(), &header.admission, resolved_by_facts(&header.facts))), &planned_keys(plan), source);
|
||||
for header in root {
|
||||
let key = header.id.id.0.as_str();
|
||||
if window.contains(key) {
|
||||
|
|
@ -289,7 +307,7 @@ fn verify(root: &[PipelineDocumentView], plan: &ImportPlan, source: &Slug) -> St
|
|||
}
|
||||
}
|
||||
let planned = planned_keys(plan);
|
||||
let window = window_writes(root.iter().map(|view| (view.id.kind, view.id.id.0.as_str(), &view.admission)), &planned, source);
|
||||
let window = window_writes(root.iter().map(|view| (view.id.kind, view.id.id.0.as_str(), &view.admission, resolved_by_document(&view.document))), &planned, source);
|
||||
differing.extend(held.keys().filter(|key| !planned.contains(*key) && !window.contains(*key)).map(|key| format!("{key}: not in the plan")));
|
||||
if differing.is_empty() {
|
||||
Ok(())
|
||||
|
|
@ -300,8 +318,8 @@ fn verify(root: &[PipelineDocumentView], plan: &ImportPlan, source: &Slug) -> St
|
|||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use huginn_mind::eureka_pipeline::{Date, PipelineKind, PipelineRuling, RulingAuthority};
|
||||
use huginn_mind::{AdmissionFacts, ImportSource, PipelineFacts, PipelineStatus, PipelineStatusSummary};
|
||||
use huginn_mind::eureka_pipeline::{Date, PipelineKind, PipelineRuling, ResolutionOutcome, RulingAuthority};
|
||||
use huginn_mind::{AdmissionFacts, ImportSource, PipelineStatus, PipelineStatusSummary};
|
||||
|
||||
use super::*;
|
||||
|
||||
|
|
@ -422,7 +440,7 @@ mod tests {
|
|||
let source = slug("eureka");
|
||||
let root = [at(header("a", Some(("eureka", 3))), 10), at(header("b", Some(("eureka", 3))), 10), at(note(header("zz", None), PipelineKind::FollowUp), 20)];
|
||||
assert_eq!(resume(&two(), &root, &source).ok().unwrap(), vec![true, false]);
|
||||
let window = window_writes(root.iter().map(|held| (held.id.kind, held.id.id.0.as_str(), &held.admission)), &planned_keys(&two()), &source);
|
||||
let window = window_writes(root.iter().map(|held| (held.id.kind, held.id.id.0.as_str(), &held.admission, resolved_by_facts(&held.facts))), &planned_keys(&two()), &source);
|
||||
assert_eq!(window.into_iter().collect::<Vec<_>>(), vec!["c:ruling:zz"]);
|
||||
}
|
||||
|
||||
|
|
@ -450,6 +468,39 @@ mod tests {
|
|||
assert_eq!(resume(&two(), &between, &source).ok().unwrap(), vec![true, false]);
|
||||
}
|
||||
|
||||
/// Void's case: the import stopped after its first receipt, her Persona then admitted
|
||||
/// follow-ups under the root and resolved them. A resolution of a window follow-up the plan
|
||||
/// lacks touches nothing planned, so it stays in place like the follow-up.
|
||||
#[test]
|
||||
fn a_resolution_of_a_window_follow_up_does_not_block_the_resume() {
|
||||
let source = slug("eureka");
|
||||
let began = [at(header("a", Some(("eureka", 3))), 10), at(header("b", Some(("eureka", 3))), 10)];
|
||||
let resolving = |subject_kind: PipelineKind, subject: &str, ordinal| {
|
||||
let mut held = note(header("zz.n1", None), PipelineKind::Resolution);
|
||||
held.facts = PipelineFacts::Resolution {
|
||||
subject: PipelineRef { kind: subject_kind, id: Short(subject.into()) },
|
||||
sequence: 1,
|
||||
outcome: ResolutionOutcome::Recorded { reason: "done".into() },
|
||||
resolved_on: Date("2026-11-09".into()),
|
||||
};
|
||||
at(held, ordinal)
|
||||
};
|
||||
for (kind, subject) in [(PipelineKind::FollowUp, "c:follow_up:zz"), (PipelineKind::Idea, "c:idea:zz")] {
|
||||
let root = [began.as_slice(), &[resolving(kind, subject, 20)]].concat();
|
||||
assert_eq!(resume(&two(), &root, &source).ok().unwrap(), vec![true, false], "{kind:?}");
|
||||
}
|
||||
for refused in [
|
||||
// a resolution of a planned document, or of a kind that derives status
|
||||
resolving(PipelineKind::FollowUp, "c:ruling:a", 20),
|
||||
resolving(PipelineKind::Ruling, "c:ruling:zz", 20),
|
||||
// admitted before the import began
|
||||
resolving(PipelineKind::FollowUp, "c:follow_up:zz", 9),
|
||||
] {
|
||||
let said = stopped(resume(&two(), &[began.as_slice(), &[refused]].concat(), &source));
|
||||
assert!(said.starts_with("no: Refused the receiver already holds records under the root"), "{said}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_verification_walk_leaves_a_window_write_out_and_still_names_an_earlier_stranger() {
|
||||
let source = slug("eureka");
|
||||
|
|
|
|||
|
|
@ -151,9 +151,17 @@ impl Image {
|
|||
.flat_map(|kind| self.posting(*kind, Alias::Root.name(), root).into_iter().flatten().map(move |id| (*kind, self.row(*id))))
|
||||
.collect();
|
||||
let began = rows.iter().filter(|(_, row)| row.view.admission.provenance.imported_from.is_some()).map(|(_, row)| row.view.admission.ordinal).min();
|
||||
// What the import replays is what this mind holds from it already or will be handed; a
|
||||
// subject this mind wrote itself is never in the plan.
|
||||
let replayed = |kind: PipelineKind, id: &str| self.id_of(kind, id).is_some_and(|row| self.row(row).view.admission.provenance.imported_from.is_some());
|
||||
rows.iter().any(|(kind, row)| {
|
||||
let resolves = match &row.view.document {
|
||||
PipelineDocument::Resolution(resolution) => Some((resolution.subject.kind, resolution.subject.id.0.as_str())),
|
||||
_ => None,
|
||||
};
|
||||
row.view.admission.provenance.imported_from.is_none()
|
||||
&& !(crate::import::inert_in_window(*kind) && began.is_some_and(|began| row.view.admission.ordinal > began))
|
||||
&& !(crate::import::inert_in_window(*kind, resolves, |id| resolves.is_some_and(|(subject, _)| replayed(subject, id)))
|
||||
&& began.is_some_and(|began| row.view.admission.ordinal > began))
|
||||
})
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -53,11 +53,14 @@ pub enum ImportFault {
|
|||
|
||||
/// Whether a native write of this kind, admitted after an import under its root began, cannot touch
|
||||
/// the status or derivation of any document the import will still replay: a follow-up or an idea
|
||||
/// has no referent a replay resolves and derives nothing. Ruling `import-window-writes-ruled`
|
||||
/// stops an import at a collision; this is the writes that cannot collide. Every other kind may
|
||||
/// answer, supersede or resolve a planned document, and still refuses the import.
|
||||
pub fn inert_in_window(kind: PipelineKind) -> bool {
|
||||
matches!(kind, PipelineKind::FollowUp | PipelineKind::Idea)
|
||||
/// has no referent a replay resolves and derives nothing, and neither has the resolution of one
|
||||
/// (`resolves`: the subject's kind and id) that the import does not replay (`planned`). Ruling
|
||||
/// `import-window-writes-ruled` stops an import at a collision; this is the writes that cannot
|
||||
/// collide. Every other kind may answer, supersede or resolve a planned document, and still refuses
|
||||
/// the import.
|
||||
pub fn inert_in_window(kind: PipelineKind, resolves: Option<(PipelineKind, &str)>, planned: impl Fn(&str) -> bool) -> bool {
|
||||
let harmless = |kind| matches!(kind, PipelineKind::FollowUp | PipelineKind::Idea);
|
||||
harmless(kind) || (kind == PipelineKind::Resolution && resolves.is_some_and(|(subject, id)| harmless(subject) && !planned(id)))
|
||||
}
|
||||
|
||||
/// The plan for `receiver` to import `campaign` from `source`, given every document the source
|
||||
|
|
|
|||
|
|
@ -613,6 +613,48 @@ fn an_import_resumes_past_a_follow_up_or_idea_the_receiver_wrote_after_it_began(
|
|||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_import_resumes_past_the_resolution_of_a_follow_up_the_receiver_wrote_after_it_began() {
|
||||
let q1 = id("question", "Q1");
|
||||
let begun = || {
|
||||
let mut recv = receiver();
|
||||
committed(import_from_kiko(&mut recv, vec![campaign(&[REPO]), question("Q1", &["A", "B"], "A")]));
|
||||
recv
|
||||
};
|
||||
// Her own follow-up, closed natively: nothing the replay reaches.
|
||||
let mut recv = begun();
|
||||
let native = follow_up("FU-N", r(K::Question, &q1));
|
||||
let native_ref = r(K::FollowUp, &pipeline_key(&native).unwrap());
|
||||
committed(admit(&mut recv, vec![native]));
|
||||
committed(admit(&mut recv, vec![resolution(native_ref, ResolutionOutcome::Recorded { reason: "done".into() })]));
|
||||
committed(import_from_kiko(&mut recv, vec![question("Q2", &["A", "B"], "A")]));
|
||||
|
||||
// A native closing of a replayed follow-up is the receiver reaching into the plan: still refused.
|
||||
let mut recv = begun();
|
||||
let imported = follow_up("FU-I", r(K::Question, &q1));
|
||||
let imported_ref = r(K::FollowUp, &pipeline_key(&imported).unwrap());
|
||||
committed(import_from_kiko(&mut recv, vec![imported]));
|
||||
committed(admit(&mut recv, vec![resolution(imported_ref, ResolutionOutcome::Recorded { reason: "done".into() })]));
|
||||
assert_eq!(refusal(import_from_kiko(&mut recv, vec![question("Q2", &["A", "B"], "A")])), MindRefusal::ImportOutsideCampaign { root: CAMPAIGN.into() });
|
||||
}
|
||||
|
||||
/// The daemon cannot see a later batch: a follow-up the receiver wrote natively and closed, whose
|
||||
/// key a later batch replays, is inert to it now. The batch then meets the key and is refused as a
|
||||
/// collision, so the window never reaches the receiver's own record.
|
||||
#[test]
|
||||
fn a_native_follow_up_a_later_batch_replays_is_refused_when_that_batch_arrives() {
|
||||
let q1 = id("question", "Q1");
|
||||
let mut recv = receiver();
|
||||
committed(import_from_kiko(&mut recv, vec![campaign(&[REPO]), question("Q1", &["A", "B"], "A")]));
|
||||
let native = follow_up("FU-L", r(K::Question, &q1));
|
||||
let native_ref = r(K::FollowUp, &pipeline_key(&native).unwrap());
|
||||
committed(admit(&mut recv, vec![native.clone()]));
|
||||
committed(admit(&mut recv, vec![resolution(native_ref, ResolutionOutcome::Recorded { reason: "done".into() })]));
|
||||
let before = head(&recv);
|
||||
assert!(matches!(refusal(import_from_kiko(&mut recv, vec![native])), MindRefusal::IdentityCollision { .. }));
|
||||
assert_eq!(head(&recv), before);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_import_never_cites_or_resolves_a_follow_up_the_receiver_wrote() {
|
||||
let q1 = id("question", "Q1");
|
||||
|
|
|
|||
Loading…
Reference in a new issue