From a2dbcf670bd2fece6ad7d91ba602d1bc031d5f6f Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:43:39 +0100 Subject: [PATCH] =?UTF-8?q?Bound=20a=20proposal's=20answers=20and=20ask=20?= =?UTF-8?q?by=20pulse=20keys=20whose=20expiry=20the=20holder=20hears,=20an?= =?UTF-8?q?d=20name=20a=20hand-back=20proposal,=20so=20no=20timer=20waits?= =?UTF-8?q?=20(ADR=200567=20=C2=A74,=20=C2=A77,=20issue=201126)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- go/compose/judge/judge.go | 43 +++++++++++++++++-------------- go/compose/judge/judge_test.go | 20 +++++++------- go/delivery/placement_proposal.go | 33 ++++++++++++++++++++++-- go/delivery/placement_test.go | 32 ++++++++++++++++------- 4 files changed, 87 insertions(+), 41 deletions(-) diff --git a/go/compose/judge/judge.go b/go/compose/judge/judge.go index 289f6f4..a6780db 100644 --- a/go/compose/judge/judge.go +++ b/go/compose/judge/judge.go @@ -15,8 +15,9 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" ) -// AnswerBound is how long the holder waits for every answer after the newest candidate was written (ADR 0567 §4): a node -// asleep or away is named past it, never passed. A variable so a test on the bus can shorten it. +// AnswerBound is how long the holder waits for a node's answer after its candidate was written (ADR 0567 §4): the time to +// live of the node's pulse key (delivery.PulseAnswerKey), whose expiry names it silent, never passed. A variable so a test +// on the bus can shorten it. var AnswerBound = 10 * time.Minute // Intended is, by node, the modules whose composition the proposal changes, and the nodes that must answer it (§3). @@ -113,12 +114,13 @@ func Intended(current, laidInput *compose.Input, currentPlaced map[string]delive // JudgeAnswers is the verdict of a proposal's answers (§4). answers is by node its HaveNode.Candidates entry for the // proposal (nil when it has none) and revisions by node the revision of its candidate key: an answer counts when it -// names the proposal and that revision. It fails on a refusal by a node that must answer, and, from the proposal's -// Written plus AnswerBound on, on every node of Must with no answer that counts, named; before that it waits for them. +// names the proposal and that revision. It fails on a refusal by a node that must answer, and on every node of Must with +// no answer that counts once its pulse key has ended (pulsing false: the bound's expiry, heard on the bus); while it +// stands it waits for them. // A difference fails nothing: one Intended computed is listed under Intended, one it did not under Unintended, both // shown in the ask's brief. The seed alone fails on any difference. func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]bool, answers map[string]*delivery.CandidateAnswer, - revisions map[string]uint64, online map[string]bool, now time.Time) delivery.PlacementVerdict { + revisions map[string]uint64, online map[string]bool, pulsing map[string]bool, now time.Time) delivery.PlacementVerdict { v := delivery.PlacementVerdict{At: now.UTC(), Intended: map[string][]string{}, Unintended: map[string][]string{}, Answers: map[string]delivery.CandidateAnswer{}} for node, set := range intended { @@ -175,27 +177,30 @@ func JudgeAnswers(p delivery.PlacementProposal, intended map[string]map[string]b if len(v.Answers) == 0 { v.Answers = nil } - if len(v.Silent) > 0 { - v.Until = p.Written.Add(AnswerBound).UTC() - named := make([]string, len(v.Silent)) - for i, node := range v.Silent { - named[i] = node - if online != nil && !online[node] { - named[i] = node + " (offline)" - } + var waiting, gone []string + for _, node := range v.Silent { + named := node + if online != nil && !online[node] { + named += " (offline)" } - if now.Before(v.Until) { - v.Notes = append(v.Notes, fmt.Sprintf("waits for the answers of %s until %s", strings.Join(named, ", "), - v.Until.Format(time.RFC3339))) + if pulsing[node] { + waiting = append(waiting, named) } else { - v.Fails = append(v.Fails, fmt.Sprintf("%s did not answer within %s of the newest candidate's write at %s; judged "+ - "again when it answers", strings.Join(named, ", "), AnswerBound, p.Written.UTC().Format(time.RFC3339))) + gone = append(gone, named) } } + if len(waiting) > 0 { + v.Notes = append(v.Notes, fmt.Sprintf("waits for the answers of %s, each for %s from its candidate's write", + strings.Join(waiting, ", "), AnswerBound)) + } + if len(gone) > 0 { + v.Fails = append(v.Fails, fmt.Sprintf("%s did not answer within %s of its candidate's write; judged again when it "+ + "answers", strings.Join(gone, ", "), AnswerBound)) + } switch { case len(v.Fails) > 0: v.State = delivery.VerdictFail - case len(v.Silent) > 0: + case len(waiting) > 0: v.State = delivery.VerdictWaits default: v.State = delivery.VerdictPass diff --git a/go/compose/judge/judge_test.go b/go/compose/judge/judge_test.go index f9b9bbd..caf5111 100644 --- a/go/compose/judge/judge_test.go +++ b/go/compose/judge/judge_test.go @@ -38,25 +38,25 @@ func TestJudge(t *testing.T) { name string seed bool answers map[string]*delivery.CandidateAnswer - now time.Time + pulsing map[string]bool state string words string marked string }{ - {"all the same", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": same(8)}, at, delivery.VerdictPass, "", ""}, + {"all the same", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": same(8)}, nil, delivery.VerdictPass, "", ""}, {"a module added where intended", false, map[string]*delivery.CandidateAnswer{"anchor": {Proposal: "p-one", Revision: 7, Verdict: delivery.CandidateDiffers, - Added: []delivery.ModuleAdded{{Module: "web", Unissued: []string{"its own secret"}}}}, "laptop": same(8)}, at, delivery.VerdictPass, "", ""}, - {"a difference not computed passes, marked", false, map[string]*delivery.CandidateAnswer{"anchor": differs(7, "mail"), "laptop": same(8)}, at, delivery.VerdictPass, "", "mail"}, - {"the seed with a difference", true, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": differs(8, "web")}, at, delivery.VerdictFail, "laptop composes web differently", ""}, + Added: []delivery.ModuleAdded{{Module: "web", Unissued: []string{"its own secret"}}}}, "laptop": same(8)}, nil, delivery.VerdictPass, "", ""}, + {"a difference not computed passes, marked", false, map[string]*delivery.CandidateAnswer{"anchor": differs(7, "mail"), "laptop": same(8)}, nil, delivery.VerdictPass, "", "mail"}, + {"the seed with a difference", true, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": differs(8, "web")}, nil, delivery.VerdictFail, "laptop composes web differently", ""}, {"a refusal", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": {Proposal: "p-one", Revision: 8, Verdict: delivery.CandidateRefused, - Refusals: []delivery.ModuleRefused{{Module: "web", Why: "no own secret"}}}}, at, delivery.VerdictFail, "laptop refuses", ""}, - {"a node not answering, within its bound", false, map[string]*delivery.CandidateAnswer{"anchor": same(7)}, at.Add(AnswerBound - time.Second), delivery.VerdictWaits, "laptop (offline)", ""}, - {"an answer of an earlier revision, past its bound", false, map[string]*delivery.CandidateAnswer{"anchor": same(6), "laptop": same(8)}, at.Add(AnswerBound), delivery.VerdictFail, "anchor did not answer", ""}, - {"its later answer", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": same(8)}, at.Add(time.Hour), delivery.VerdictPass, "", ""}, + Refusals: []delivery.ModuleRefused{{Module: "web", Why: "no own secret"}}}}, nil, delivery.VerdictFail, "laptop refuses", ""}, + {"a node not answering, within its bound", false, map[string]*delivery.CandidateAnswer{"anchor": same(7)}, map[string]bool{"laptop": true}, delivery.VerdictWaits, "laptop (offline)", ""}, + {"an answer of an earlier revision, past its bound", false, map[string]*delivery.CandidateAnswer{"anchor": same(6), "laptop": same(8)}, nil, delivery.VerdictFail, "anchor did not answer", ""}, + {"its later answer", false, map[string]*delivery.CandidateAnswer{"anchor": same(7), "laptop": same(8)}, nil, delivery.VerdictPass, "", ""}, } { q := p q.Seed = row.seed - v := JudgeAnswers(q, intended, row.answers, revisions, map[string]bool{"anchor": true}, row.now) + v := JudgeAnswers(q, intended, row.answers, revisions, map[string]bool{"anchor": true}, row.pulsing, at) all := strings.Join(append(append([]string{}, v.Fails...), v.Notes...), "; ") if v.State != row.state || !strings.Contains(all, row.words) { t.Errorf("%s: %s, %q", row.name, v.State, all) diff --git a/go/delivery/placement_proposal.go b/go/delivery/placement_proposal.go index 80656ec..1a99de2 100644 --- a/go/delivery/placement_proposal.go +++ b/go/delivery/placement_proposal.go @@ -26,8 +26,34 @@ const ( PlacementProposalsRead = AssignerModule + "." + PlacementProposalsState // PlacementProposalsHistory is the revisions the bucket keeps per key. PlacementProposalsHistory = 5 + + // PlacementPulsesState is the holder's state of bounds (§4): a key per node a proposal waits on, written with a time + // to live of the answer bound, and a key per raised ask, with its ask's; the server's delete marker when one ends is + // the event that fails a silent node or ends an unanswered ask, so the holder keeps no timer. Its bucket allows a + // time to live per key (catalogue.StateDeclaration.KeyTTL). + PlacementPulsesState = "pulses" + PlacementPulsesBucket = AssignerModule + "_" + PlacementPulsesState ) +// PulseAnswerKey is the key of the bound on one node's answer to a proposal: `answer..`. +func PulseAnswerKey(proposal, node string) (string, error) { + if err := CheckProposalID(proposal); err != nil { + return "", err + } + if err := checkName("node", node); err != nil { + return "", err + } + return "answer." + proposal + "." + node, nil +} + +// PulseAskKey is the key of the bound on a proposal's raised ask: `ask.`. +func PulseAskKey(proposal string) (string, error) { + if err := CheckProposalID(proposal); err != nil { + return "", err + } + return "ask." + proposal, nil +} + // The operations a proposal's change names (ADR 0567 §2). const ( OpPlace = "place" // place a module on a node, with its host ports and bindings @@ -360,6 +386,9 @@ type PlacementProposal struct { // Must is every node it must hear from (§3): each whose key it changes and each whose composition it changes // through another's, sorted; set by the holder once it composed, before then the nodes of Base. Must []string `json:"must,omitempty"` + // HandBack says it is the operator's hand-back (§7): every key of Base deleted on its warrant, with no change and no + // node's answer. + HandBack bool `json:"hand-back,omitempty"` // State is ProposalAnswering, ProposalFailed or ProposalAsked; Verdict is the latest judgement, Ask the authorising // ask raised on a passing one. State string `json:"state"` @@ -385,8 +414,8 @@ func (p PlacementProposal) check() error { if strings.TrimSpace(p.By) == "" || strings.TrimSpace(p.Why) == "" { return fmt.Errorf("delivery: the proposal %s names no proposer or no why", p.ID) } - if len(p.Changes) == 0 || len(p.Base) == 0 { - return fmt.Errorf("delivery: the proposal %s changes nothing", p.ID) + if len(p.Base) == 0 || (len(p.Changes) == 0) != p.HandBack || (p.HandBack && p.Seed) { + return fmt.Errorf("delivery: the proposal %s changes nothing, or is a hand-back with changes", p.ID) } for i, c := range p.Changes { if err := c.check(); err != nil { diff --git a/go/delivery/placement_test.go b/go/delivery/placement_test.go index bca020c..94cb097 100644 --- a/go/delivery/placement_test.go +++ b/go/delivery/placement_test.go @@ -535,20 +535,32 @@ func TestPlacement(t *testing.T) { if ExportDigest(current) != ExportDigest(map[string]Placement{"novox": novox, "g14": g14}) || ExportDigest(current) == ExportDigest(nil) { t.Error("the export digest depends on order or not on the placements") } + back := p + back.HandBack, back.Changes = true, nil + if _, err := EncodePlacementProposal(back); err != nil { + t.Errorf("a hand-back: %v", err) + } + if k, err := PulseAnswerKey(id, "g14"); err != nil || k != "answer."+id+".g14" { + t.Errorf("PulseAnswerKey = %q, %v", k, err) + } + if k, err := PulseAskKey(id); err != nil || k != "ask."+id || PlacementPulsesBucket != "mesh-delivery-assigner_pulses" { + t.Errorf("PulseAskKey = %q, %v", k, err) + } if ChangesDigest(p) == ChangesDigest(seed) { t.Error("two proposals' changes digest alike") } for name, mut := range map[string]func(*PlacementProposal){ - "no id": func(p *PlacementProposal) { p.ID = "" }, - "no proposer": func(p *PlacementProposal) { p.By = "" }, - "no why": func(p *PlacementProposal) { p.Why = " " }, - "no change": func(p *PlacementProposal) { p.Changes = nil }, - "no base": func(p *PlacementProposal) { p.Base = map[string]uint64{"novox": 1} }, - "no state": func(p *PlacementProposal) { p.State = "" }, - "asked, no ask": func(p *PlacementProposal) { p.State = ProposalAsked }, - "seed mixed": func(p *PlacementProposal) { p.Seed, p.ExportDigest = true, "d" }, - "bad change": func(p *PlacementProposal) { p.Changes[0].Op = "swap" }, - "seed, no digest": func(p *PlacementProposal) { *p = seed; p.ExportDigest = "" }, + "no id": func(p *PlacementProposal) { p.ID = "" }, + "no proposer": func(p *PlacementProposal) { p.By = "" }, + "no why": func(p *PlacementProposal) { p.Why = " " }, + "no change": func(p *PlacementProposal) { p.Changes = nil }, + "no base": func(p *PlacementProposal) { p.Base = map[string]uint64{"novox": 1} }, + "no state": func(p *PlacementProposal) { p.State = "" }, + "asked, no ask": func(p *PlacementProposal) { p.State = ProposalAsked }, + "seed mixed": func(p *PlacementProposal) { p.Seed, p.ExportDigest = true, "d" }, + "bad change": func(p *PlacementProposal) { p.Changes[0].Op = "swap" }, + "seed, no digest": func(p *PlacementProposal) { *p = seed; p.ExportDigest = "" }, + "hand-back with changes": func(p *PlacementProposal) { p.HandBack = true }, } { q := p q.Changes = append([]PlacementProposalChange(nil), p.Changes...)