From 06e668cb90035f0cdac3a22858c9b91732cc639e Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:46:39 +0100 Subject: [PATCH] =?UTF-8?q?Assigner:=20bound=20answers=20and=20asks=20by?= =?UTF-8?q?=20pulse=20keys=20instead=20of=20a=20timer,=20ask=20the=20opera?= =?UTF-8?q?tor=20for=20a=20hand-back,=20follow=20a=20partial=20write=20thr?= =?UTF-8?q?ough,=20and=20call=20no=20controller=20verb=20for=20the=20seed?= =?UTF-8?q?=20(ADR=200567=20=C2=A74,=20=C2=A75,=20=C2=A77,=20=C2=A79,=20re?= =?UTF-8?q?view=20round=202)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../integration_test.go | 22 +- .../cmd/mesh-delivery-assigner/main.go | 34 +-- .../cmd/mesh-delivery-assigner/pass.go | 8 +- .../cmd/mesh-delivery-assigner/seat.go | 241 ++++++++++++------ .../cmd/mesh-delivery-assigner/seat_tools.go | 36 +-- .../mesh-delivery-assigner/seat_tools_test.go | 14 +- .../internal/assign/seat.go | 37 ++- .../internal/assign/seat_test.go | 30 +++ .../mesh-delivery-assigner/manifest_test.go | 2 +- modules/mesh-delivery-assigner/module.json | 1 + 10 files changed, 280 insertions(+), 145 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 5cd0d15..1323409 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -22,6 +22,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose/judge" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/provider" @@ -71,7 +72,11 @@ func startBed(t *testing.T, leaveOut string) *bed { if name == leaveOut { continue } - kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5}) + cfg := jetstream.KeyValueConfig{Bucket: name, History: 5} + if name == delivery.PlacementPulsesBucket { + cfg.History, cfg.LimitMarkerTTL = 1, time.Minute // a time to live per key, as the controller makes it (key-ttl) + } + kv, err := js.CreateKeyValue(ctx, cfg) if err != nil { t.Fatal(err) } @@ -1082,6 +1087,9 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) { return rev } anchorRev, laptopRev := place("anchor", "web"), place("laptop", "shell") + bound := judge.AnswerBound + t.Cleanup(func() { judge.AnswerBound = bound }) + judge.AnswerBound = 2 * time.Second saysWhatItKeeps(t, b, b.start(t)) verbs := newSeatTools(bedStore{b}, func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"ok":true}`), nil }) propose := func() delivery.PlacementProposal { @@ -1177,4 +1185,16 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) { if e, _ := b.buckets[delivery.PlacementBucket].Get(b.ctx, key); e == nil || !strings.Contains(string(e.Value()), "web") { t.Error("a declined proposal wrote its key") } + + // ADR 0567 §4: a node that does not answer fails the verdict once its pulse key's time to live ends, heard on the bus + // with no timer of the assigner's, naming it. + q, err = verbs.doPropose(map[string]any{"why": "again", "changes": `[{"op":"take-off","node":"laptop","module":"web"}]`}, "op") + if err != nil { + t.Fatal(err) + } + silent := q.(proposed).Proposal + failed := proposalNow(silent.ID, "failed for a silent node", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalFailed }) + if failed.Verdict == nil || !strings.Contains(strings.Join(failed.Verdict.Fails, "; "), "laptop (offline) did not answer") { + t.Errorf("the silent node's verdict = %+v", failed.Verdict) + } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index d2fca5e..5311d3a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -14,7 +14,6 @@ import ( "os" "os/signal" "slices" - "sync" "syscall" "time" @@ -74,11 +73,12 @@ const goBuildModule = "mesh-build-go" // (ADR 0567 §1), named after the assigner as every module state is. var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket, - delivery.PlacementCandidatesBucket} + delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket} // seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants // them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind). -var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposalsBucket, delivery.PlacementCandidatesBucket} +var seatBuckets = []string{delivery.PlacementBucket, delivery.PlacementProposalsBucket, delivery.PlacementCandidatesBucket, + delivery.PlacementPulsesBucket} // keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view // (and the node agent from release 6), its differences by the view, its compositions by the checker, its asks by the view @@ -88,6 +88,7 @@ var keptStates = []announce.State{ {Name: delivery.PlacementState, Readable: true}, {Name: delivery.PlacementProposalsState, Readable: true}, {Name: delivery.PlacementCandidatesState, Readable: true}, + {Name: delivery.PlacementPulsesState, Readable: true}, {Name: "asks", Readable: true}, {Name: "compositions", Readable: true}, {Name: "differences", Readable: true}, @@ -177,6 +178,7 @@ var watched = []string{ delivery.PlacementBucket, delivery.PlacementProposalsBucket, delivery.PlacementCandidatesBucket, + delivery.PlacementPulsesBucket, delivery.PresenceBucket, delivery.HaveBucket, delivery.AssignerAsksBucket, @@ -317,7 +319,6 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } } p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, said: map[string]bool{}, answeredAt: map[string]uint64{}} - p.wakeAt = wakeAtVia(ctx, wake) p.asks = newComposeAsks(ctx, buildrun.Asker{Publisher: buildrun.JetStreamAsks{JS: js}, Waiter: buildrun.JetStreamAsks{JS: js}}, wake, logger.Printf) if bus != nil && kvs[delivery.AssignerPutsBucket] != nil { p.acts = newPutActs(bus, wake, logger.Printf) @@ -347,31 +348,6 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } } -// wakeAtVia asks wake once at each time it is given, as a context's deadline (ADR 0437 §8: no timer), each time once. -func wakeAtVia(ctx context.Context, wake func()) func(time.Time) { - var mu sync.Mutex - asked := map[int64]bool{} - return func(at time.Time) { - mu.Lock() - defer mu.Unlock() - if asked[at.UnixNano()] { - return - } - asked[at.UnixNano()] = true - go func() { - bound, cancel := context.WithDeadline(ctx, at) - defer cancel() - <-bound.Done() - mu.Lock() - delete(asked, at.UnixNano()) - mu.Unlock() - if ctx.Err() == nil { - wake() - } - }() - } -} - // PutsReopenEvery is how often, at most, a pass opens the puts' bucket again while it is not open. var PutsReopenEvery = time.Minute diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index e584912..651b73a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -43,16 +43,10 @@ type passer struct { // nodeFailed is, by node and module, the words of the condition last raised for a failure that is the node's own (ADR 0557 // §1, issue 1496): raised once per words, cleared once when it ends. nodeFailed map[string]string - // wakeAt asks a pass at a time: the end of a proposal's answer bound (ADR 0567 §4), as a context's deadline (ADR 0437 - // §8: no timer). Nil asks none. - wakeAt func(at time.Time) // The seat mesh-placement's holder (seat.go, ADR 0567): its calls of the controller (nil: none is made), how it says - // the seat's events (nil: none is said), the placement keys it saw last (a key gone that it did not delete is a - // hand-back), the proposals it said, and what Intended computed by proposal and base. + // the seat's events (nil: none is said), the proposals it said, and what Intended computed by proposal and base. seat *seatActs publish publisher - seen map[string]delivery.Placement - seenRev map[string]uint64 saidProposed map[string]bool intended map[string]map[string]map[string]bool } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go index a9a9cc1..79b00f3 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "reflect" "sort" "strings" @@ -126,6 +127,16 @@ func (a *seatActs) record(warrant string) { }() } +// handAct records a hand-back in the controller's hand-act log (§7). +func (a *seatActs) handAct(prop delivery.PlacementProposal) { + go func() { + if _, err := a.bus.Call(handActCall, map[string]any{"what": "mesh-placement hand-back of " + strings.Join(prop.Nodes(), ", "), + "why": prop.Why, "cause": "asked by " + prop.By + ", approved by the operator"}); err != nil { + a.logf("the hand-back %s is not recorded in the hand-act log: %v", prop.ID, err) + } + }() +} + // callAll makes the controller's calls in order in the background, each said in the journal and kept by its pair. func (a *seatActs) callAll(calls []assign.ControllerCall) { if len(calls) == 0 { @@ -225,16 +236,17 @@ func (p *passer) answersTo(id string) map[string]*delivery.CandidateAnswer { return out } -// placementSeat takes every open proposal one step further, and says the keys a hand-back deleted. +// placementSeat takes every open proposal one step further. func (p *passer) placementSeat(ctx context.Context) { - if p.kvs[delivery.PlacementProposalsBucket] == nil || p.kvs[delivery.PlacementBucket] == nil || p.kvs[delivery.PlacementCandidatesBucket] == nil { - return + for _, b := range seatBuckets { + if p.kvs[b] == nil { + return + } } if p.saidProposed == nil { p.saidProposed, p.intended = map[string]bool{}, map[string]map[string]map[string]bool{} } current, revs := p.seatPlacements() - p.handedBack(current, revs) ids := make([]string, 0) raws := p.w.read(delivery.PlacementProposalsBucket) for id := range raws { @@ -251,39 +263,46 @@ func (p *passer) placementSeat(ctx context.Context) { } } -// handedBack says placement-changed for every key that went without this process deleting it: a hand-back (§7). -func (p *passer) handedBack(current map[string]delivery.Placement, revs map[string]uint64) { - if p.seen == nil { - p.seen = map[string]delivery.Placement{} - p.seenRev = map[string]uint64{} - for n, pl := range current { - p.seen[n], p.seenRev[n] = pl, revs[n] - } +// pulse writes a bound's key with its time to live (§4): its expiry, heard on the watch as the server's delete marker, +// is the event that ends the wait. Written again over one that stands, so its time starts anew. +func (p *passer) pulse(ctx context.Context, key string, ttl time.Duration) { + kv := p.kvs[delivery.PlacementPulsesBucket] + _ = kv.Delete(ctx, key) + rev, err := kv.Create(ctx, key, []byte(`{}`), jetstream.KeyTTL(ttl)) + if err != nil { + p.logf("the bound %s is not written, so its wait ends at once: %v", key, err) return } - var gone []delivery.PlacementChange - for node, before := range p.seen { - if _, there := current[node]; there { - continue + p.w.putEntry(delivery.PlacementPulsesBucket, key, []byte(`{}`), rev, p.clock()) +} + +// pulsing says whether a bound's key stands. +func (p *passer) pulsing(key string) bool { return p.w.has(delivery.PlacementPulsesBucket, key) } + +// deletePulses deletes every bound of a proposal. +func (p *passer) deletePulses(ctx context.Context, id string) { + for key := range p.w.read(delivery.PlacementPulsesBucket) { + if strings.HasSuffix(key, "."+id) || strings.Contains(key, "."+id+".") { + if err := p.kvs[delivery.PlacementPulsesBucket].Delete(ctx, key); err == nil { + p.w.put(delivery.PlacementPulsesBucket, key, nil) + } } - c := delivery.ChangeOf(before, delivery.Placement{Node: node}) - c.Before = p.seenRev[node] - gone = append(gone, c) - delete(p.seen, node) - delete(p.seenRev, node) } - for n, pl := range current { - p.seen[n], p.seenRev[n] = pl, revs[n] +} + +// ask raises (or raises again) a proposal's authorising ask and its bound, and answers the proposal asked. +func (p *passer) ask(ctx context.Context, prop delivery.PlacementProposal, akey, brief string) (delivery.PlacementProposal, bool) { + a, err := assign.PlacementAsk(akey, prop, brief, assign.PlacementAsking(prop, p.clock()), p.clock()) + if err != nil { + p.logf("the ask about the proposal %s is not written: %v", prop.ID, err) + return prop, false } - if len(gone) == 0 || p.publish == nil { - return - } - delivery.SortChanges(gone) - ev := delivery.PlacementEvent{Proposal: delivery.PlacementProposal{ID: "hand-back", By: "the operator's hand act", - Why: "placement handed back to the controller's compositions"}, Nodes: gone, Ended: "handed back"} - if err := p.publish(delivery.PlacementChangedEvent, ev, fmt.Sprintf("hand-back.%d", p.clock().UnixNano())); err != nil { - p.logf("the hand-back of placement is not said as an event: %v", err) + p.put(ctx, delivery.AssignerAsksBucket, akey, func() ([]byte, error) { return delivery.EncodeAssignerAsk(akey, a) }) + if pk, err := delivery.PulseAskKey(prop.ID); err == nil { + p.pulse(ctx, pk, a.Ask.Expires.Sub(p.clock())) } + prop.State, prop.Ask = delivery.ProposalAsked, a.Ask.ID + return prop, true } // stepProposal takes one proposal one step further. @@ -297,21 +316,38 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos return } akey, _ := delivery.PlacementAskKey(prop.ID) + again := !prop.Again.IsZero() && (prop.Verdict == nil || prop.Again.After(prop.Verdict.At)) if prop.State == delivery.ProposalAsked { + askPulse, _ := delivery.PulseAskKey(prop.ID) switch { case p.answeredWith(akey, assign.OptionYes): - p.warranted(ctx, prop, rev, akey, current, revs) + if prop.HandBack { + p.handBack(ctx, prop, akey, revs) + } else { + p.warranted(ctx, prop, rev, akey, current, revs) + } return case p.answeredWith(akey, assign.OptionNo): p.endProposal(ctx, prop, "declined by the operator") return - } - if a, err := delivery.DecodeAssignerAsk(akey, p.w.value(delivery.AssignerAsksBucket, akey)); err == nil && - !a.Ask.Expires.IsZero() && p.clock().After(a.Ask.Expires) && (prop.Again.IsZero() || prop.Again.Before(prop.Verdict.At)) { + case !p.pulsing(askPulse) && !again: p.endProposal(ctx, prop, "its ask expired unanswered") return } } + if prop.HandBack { + // The operator's alone (§7): no node answers it; its ask is raised at once, and again on again. + if prop.State != delivery.ProposalAsked || again { + brief := fmt.Sprintf("%s (asked by %s): every node's placement key is deleted, %s, and each node returns to the "+ + "controller's compositions; the history is kept", prop.Why, prop.By, strings.Join(prop.Nodes(), ", ")) + now := p.clock().UTC() + prop.Verdict = &delivery.PlacementVerdict{State: delivery.VerdictPass, At: now} + if prop, ok := p.ask(ctx, prop, akey, brief); ok { + p.writeProposal(ctx, prop, rev) + } + } + return + } if moved(prop, revs) { p.layAgain(ctx, prop, rev, current, revs, "a key it was laid over moved since") return @@ -329,11 +365,13 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos if written.After(prop.Written) { prop.Written = written } - again := !prop.Again.IsZero() && (prop.Verdict == nil || prop.Again.After(prop.Verdict.At)) - v := judge.JudgeAnswers(prop, p.intendedOf(prop, current, cands), p.answersTo(prop.ID), candRevs, p.w.online(), p.clock()) - if v.State == delivery.VerdictWaits && p.wakeAt != nil { - p.wakeAt(v.Until) + pulsing := map[string]bool{} + for _, node := range prop.Must { + if k, err := delivery.PulseAnswerKey(prop.ID, node); err == nil { + pulsing[node] = p.pulsing(k) + } } + v := judge.JudgeAnswers(prop, p.intendedOf(prop, current, cands), p.answersTo(prop.ID), candRevs, p.w.online(), pulsing, p.clock()) if prop.Verdict != nil && sameVerdict(*prop.Verdict, v) && !again { return } @@ -345,14 +383,10 @@ func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementPropos prop.State, prop.Ask = delivery.ProposalAnswering, "" case delivery.VerdictPass: if prop.State != delivery.ProposalAsked || again { - asking := assign.PlacementAsking(prop, p.clock()) - a, err := assign.PlacementAsk(akey, prop, judge.Brief(prop, v), asking, p.clock()) - if err != nil { - p.logf("the ask about the proposal %s is not written: %v", prop.ID, err) + var ok bool + if prop, ok = p.ask(ctx, prop, akey, judge.Brief(prop, v)); !ok { return } - p.put(ctx, delivery.AssignerAsksBucket, akey, func() ([]byte, error) { return delivery.EncodeAssignerAsk(akey, a) }) - prop.State, prop.Ask = delivery.ProposalAsked, a.Ask.ID if again { prop.Words = "judged and asked again (again)" } @@ -454,6 +488,11 @@ func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal } p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return delivery.EncodePlacementCandidate(c) }) } + for _, node := range must { + if k, err := delivery.PulseAnswerKey(prop.ID, node); err == nil { + p.pulse(ctx, k, judge.AnswerBound) + } + } prop.Must = must if len(notes) > 0 { prop.Words = strings.TrimPrefix(prop.Words+"; "+strings.Join(notes, "; "), "; ") @@ -473,6 +512,7 @@ func (p *passer) layAgain(ctx context.Context, prop delivery.PlacementProposal, return } p.deleteCandidates(ctx, prop.ID) + p.deletePulses(ctx, prop.ID) for _, node := range again.Nodes() { key, _ := delivery.PlacementCandidateKey(prop.ID, node) c := delivery.PlacementCandidate{Proposal: prop.ID, Placement: laid[node]} @@ -541,8 +581,11 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal, } kv := p.kvs[delivery.PlacementBucket] var changes []delivery.PlacementChange + writtenLaid := map[string]*delivery.Placement{} + after := maps.Clone(current) + afterRevs := maps.Clone(revs) for _, w := range writes { - var after uint64 + var rev2 uint64 var err error switch { case w.Value == nil: @@ -550,47 +593,100 @@ func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal, default: raw, encErr := delivery.EncodePlacement(*w.Value) if encErr != nil { - p.endProposal(ctx, prop, "its placement of "+w.Node+" cannot be written: "+encErr.Error()) - return - } - if w.Base == 0 { - after, err = kv.Create(ctx, w.Key, raw) + err = encErr + } else if w.Base == 0 { + rev2, err = kv.Create(ctx, w.Key, raw) } else { - after, err = kv.Update(ctx, w.Key, raw, w.Base) + rev2, err = kv.Update(ctx, w.Key, raw, w.Base) } } if err != nil { - // Another writer came first: a hand-back. What was written stands, said; the rest is laid again. - p.logf("placement %s is not written at revision %d (%v): the proposal %s is laid again", w.Key, w.Base, err, prop.ID) - p.sayChanged(prop, answer, changes) - p.layAgain(ctx, prop, rev, current, revs, "a key moved while it was written") + // Another writer came first (a hand-back): what is written stands and is followed through; the rest is laid + // again over the keys as this write left them. + p.logf("placement %s is not written at revision %d (%v): the rest of the proposal %s is laid again", w.Key, w.Base, err, prop.ID) + p.followWrite(prop, answer, changes, current, writtenLaid) + if rest, ok := assign.Rest(prop, writtenLaid); ok { + p.layAgain(ctx, rest, rev, after, afterRevs, "a key moved while it was written, after "+strings.Join(sortedNodes(writtenLaid), ", ")+" were") + } else { + p.endProposal(ctx, prop, "a key moved while it was written: "+err.Error()) + } return } before := current[w.Node] - after2 := delivery.Placement{Node: w.Node} + now := delivery.Placement{Node: w.Node} if w.Value != nil { - after2 = *w.Value + now = *w.Value + after[w.Node] = *w.Value + } else { + delete(after, w.Node) } - c := delivery.ChangeOf(before, after2) - c.Before, c.After = w.Base, after + afterRevs[w.Node] = rev2 + c := delivery.ChangeOf(before, now) + c.Before, c.After = w.Base, rev2 changes = append(changes, c) - if p.seen != nil { - if w.Value != nil { - p.seen[w.Node], p.seenRev[w.Node] = *w.Value, after - } else { - delete(p.seen, w.Node) - delete(p.seenRev, w.Node) - } - } + writtenLaid[w.Node] = w.Value } - p.seat.record(answer.Warrant) p.deleteCandidates(ctx, prop.ID) + p.deletePulses(ctx, prop.ID) _ = p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID) p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil) - p.sayChanged(prop, answer, changes) - if p.seat != nil && !assign.Leased { - p.seat.callAll(assign.ControllerCalls(current, laid, prop.ID, answer.Warrant)) + p.followWrite(prop, answer, changes, current, writtenLaid) +} + +// followWrite follows the keys a warrant wrote: the warrant recorded, placement-changed said, and, until the lease +// write, the controller's assignments kept in step for them (§8). The seed calls no controller verb: it is what the +// controller holds already (§9). +func (p *passer) followWrite(prop delivery.PlacementProposal, answer delivery.Answer, changes []delivery.PlacementChange, + before map[string]delivery.Placement, written map[string]*delivery.Placement) { + if len(changes) == 0 { + return } + p.seat.record(answer.Warrant) + p.sayChanged(prop, answer, changes) + if p.seat != nil && !assign.Leased && !prop.Seed { + p.seat.callAll(assign.ControllerCalls(before, written, prop.ID, answer.Warrant)) + } +} + +func sortedNodes(m map[string]*delivery.Placement) []string { + out := make([]string, 0, len(m)) + for n := range m { + out = append(out, n) + } + sort.Strings(out) + return out +} + +// handBack deletes every key on the operator's warrant (§7) at the revision the hand-back read, records it in the +// controller's hand-act log and says placement-changed; every node returns to the controller's compositions. A key moved +// since ends it, and the operator hands back again. +func (p *passer) handBack(ctx context.Context, prop delivery.PlacementProposal, akey string, revs map[string]uint64) { + if err := p.performs(akey, assign.PlacementAct(prop)); err != nil { + p.endProposal(ctx, prop, "its warrant does not bind this hand-back, so nothing is deleted: "+err.Error()) + return + } + answer, _ := p.w.answer(akey) + current, _ := p.seatPlacements() + var changes []delivery.PlacementChange + for _, node := range prop.Nodes() { + key, _ := delivery.PlacementKey(node) + if err := p.kvs[delivery.PlacementBucket].Delete(ctx, key, jetstream.LastRevision(prop.Base[node])); err != nil { + p.sayChanged(prop, answer, changes) + p.endProposal(ctx, prop, fmt.Sprintf("%s moved since the hand-back was asked (%v): hand back again", key, err)) + return + } + c := delivery.ChangeOf(current[node], delivery.Placement{Node: node}) + c.Before = prop.Base[node] + changes = append(changes, c) + } + if p.seat != nil { + p.seat.handAct(prop) + } + p.seat.record(answer.Warrant) + _ = p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID) + p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil) + p.deletePulses(ctx, prop.ID) + p.sayChanged(prop, answer, changes) } // sayChanged says placement-changed for the keys written. @@ -638,6 +734,7 @@ func (p *passer) endProposal(ctx context.Context, prop delivery.PlacementProposa } } p.deleteCandidates(ctx, prop.ID) + p.deletePulses(ctx, prop.ID) if err := p.kvs[delivery.PlacementProposalsBucket].Delete(ctx, prop.ID); err != nil && !errors.Is(err, context.Canceled) { p.logf("the proposal %s is not deleted, and is ended at the next pass: %v", prop.ID, err) return diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go index 5f4d55c..3c7e708 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go @@ -311,8 +311,9 @@ func (s *seatTools) doAdopt(args map[string]any, by string) (any, error) { // handActCall records a hand act in the controller's hand-act log. const handActCall = "seat:" + catalogue.ControllerSeatName + ".hand-act" -// doHandBack is the verb hand-back (§7): every placement key deleted, the history kept, recorded as the operator's hand -// act; every node returns to the controller's compositions. The assigner says placement-changed for each key it sees go. +// doHandBack is the verb hand-back (§7): the operator's alone, so it is asked as a proposal is. It writes a hand-back +// proposal naming every key at its revision; the assigner raises its authorising ask and, on its warrant, deletes the keys, +// records the hand act and says placement-changed; every node returns to the controller's compositions. func (s *seatTools) doHandBack(args map[string]any, by string) (any, error) { if assign.Leased { return nil, errors.New("placement is the mesh's alone since the lease write: there is no controller to hand it back to") @@ -321,23 +322,28 @@ func (s *seatTools) doHandBack(args map[string]any, by string) (any, error) { if why == "" { return nil, errors.New("hand-back says why: it is recorded with the hand act") } - keys, err := s.st.keys(delivery.PlacementState) + if by == "" { + return nil, errors.New("the runtime names no caller: a hand-back is made by someone") + } + _, revs, err := s.placements() if err != nil { - return nil, fmt.Errorf("placement cannot be listed: %w", err) + return nil, err } - sort.Strings(keys) - if _, err := s.ask(handActCall, map[string]any{"what": "mesh-placement hand-back of " + strings.Join(keys, ", "), - "why": why, "cause": "asked by " + by}); err != nil { - return nil, fmt.Errorf("the hand-back is not recorded as a hand act, so nothing is deleted: %w", err) + if len(revs) == 0 { + return nil, errors.New("there is no placement key to hand back") } - var deleted []string - for _, key := range keys { - if err := s.st.del(delivery.PlacementState, key); err != nil { - return map[string]any{"deleted": deleted}, fmt.Errorf("%s is not deleted (%v); the others before it are: hand-back again", key, err) - } - deleted = append(deleted, key) + now := s.now() + p := delivery.PlacementProposal{ID: delivery.NewProposalID(now), By: by, Why: why, At: now.UTC(), Written: now.UTC(), + HandBack: true, Base: revs, State: delivery.ProposalAnswering} + raw, err := delivery.EncodePlacementProposal(p) + if err != nil { + return nil, err } - return map[string]any{"deleted": deleted, "follow": "each node returns to the controller's compositions as it hears its key go"}, nil + if err := s.st.put(delivery.PlacementProposalsState, p.ID, raw); err != nil { + return nil, fmt.Errorf("the hand-back is not written: %w", err) + } + return proposed{Proposal: p, Follow: "the operator is asked to approve it; on the warrant every key is deleted and each " + + "node returns to the controller's compositions"}, nil } // placementAnswer is what the verb placement answers. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index fdb459f..1594f62 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -2,7 +2,6 @@ package main import ( "encoding/json" - "errors" "os" "slices" "strings" @@ -137,14 +136,11 @@ func TestTheSeatsVerbs(t *testing.T) { } } - // hand-back: recorded as the operator's hand act, then every key deleted; refused when the hand act is not recorded. - failing := newSeatTools(st, func(string, any) (json.RawMessage, error) { return nil, errors.New("no controller") }) - if _, err := failing.doHandBack(map[string]any{"why": "back"}, "op"); err == nil || len(st[delivery.PlacementState].values) != 1 { - t.Errorf("a hand-back not recorded deleted keys: %v", err) - } - asked = nil - if _, err := s.doHandBack(map[string]any{"why": "back"}, "op"); err != nil || !slices.Equal(asked, []string{handActCall}) || len(st[delivery.PlacementState].values) != 0 { - t.Errorf("hand-back: %v, asked %v", err, asked) + // hand-back: the operator's alone, written as a hand-back proposal naming every key at its revision, for the assigner + // to ask; nothing is deleted by the verb. + got, err = s.doHandBack(map[string]any{"why": "back"}, "op") + if hb := got.(proposed).Proposal; err != nil || !hb.HandBack || hb.Base["anchor"] != uint64(len(raw)) || len(st[delivery.PlacementState].values) != 1 { + t.Errorf("hand-back = %+v, %v", got, err) } if tools := s.tools(); len(tools) != 7 || tools[2].Name != catalogue.PlacementVerbPropose || tools[2].Run == nil || tools[2].Input == nil { t.Errorf("the seat's verbs as served: %+v", tools) diff --git a/modules/mesh-delivery-assigner/internal/assign/seat.go b/modules/mesh-delivery-assigner/internal/assign/seat.go index 1a7e247..0874e81 100644 --- a/modules/mesh-delivery-assigner/internal/assign/seat.go +++ b/modules/mesh-delivery-assigner/internal/assign/seat.go @@ -240,18 +240,14 @@ func ExportOf(raw []byte) (map[string]delivery.Placement, error) { if a.OK != nil && !*a.OK { return nil, fmt.Errorf("the controller's placement-export refused: %s", strings.TrimSpace(a.Output)) } - data := []byte(a.Answer) - if len(data) == 0 || string(data) == "null" { - out := a.Output - start, end := strings.Index(out, "{"), strings.LastIndex(out, "}") - if start < 0 || end < start { - return nil, fmt.Errorf("the controller's placement-export printed no placements: %q", out) - } - data = []byte(out[start : end+1]) - } var byNode map[string]json.RawMessage - if err := json.Unmarshal(data, &byNode); err != nil { - return nil, fmt.Errorf("the controller's placement-export is no placement by node: %w", err) + if len(a.Answer) > 0 && string(a.Answer) != "null" { + if err := json.Unmarshal(a.Answer, &byNode); err != nil { + return nil, fmt.Errorf("the controller's placement-export is no placement by node: %w", err) + } + } else if err := json.NewDecoder(strings.NewReader(a.Output)).Decode(&byNode); err != nil { + // The command prints its JSON first; what its standard error said after it is left. + return nil, fmt.Errorf("the controller's placement-export printed no placement by node: %w", err) } out := map[string]delivery.Placement{} for node, rawP := range byNode { @@ -279,3 +275,22 @@ func SeedChanges(export map[string]delivery.Placement) []delivery.PlacementPropo } return out } + +// Rest is what a proposal leaves to write once the keys of written are: its changes of the other nodes, to be laid again +// over the keys as the write left them (§5: a key moved while it was written). False when nothing is left. +func Rest(p delivery.PlacementProposal, written map[string]*delivery.Placement) (delivery.PlacementProposal, bool) { + rest := p + rest.Changes, rest.Base = nil, map[string]uint64{} + for _, c := range p.Changes { + if _, done := written[c.Node]; !done { + rest.Changes = append(rest.Changes, c) + } + } + for node, base := range p.Base { + if _, done := written[node]; !done { + rest.Base[node] = base + } + } + rest.Must, rest.Verdict, rest.Ask, rest.State = nil, nil, "", delivery.ProposalAnswering + return rest, len(rest.Changes) > 0 +} diff --git a/modules/mesh-delivery-assigner/internal/assign/seat_test.go b/modules/mesh-delivery-assigner/internal/assign/seat_test.go index accec8f..e478e8c 100644 --- a/modules/mesh-delivery-assigner/internal/assign/seat_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/seat_test.go @@ -39,6 +39,9 @@ func TestThePlacementSeatDecides(t *testing.T) { }{ "a port before the lease": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "web", Ports: map[string]int{"80": 8080}}), "lease write"}, "a holding": {in(delivery.PlacementProposalChange{Op: delivery.OpHold, Node: "anchor", Module: "store", Claim: "mesh-store", Scope: "mesh"}), "lease write"}, + "a binding": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "web", Bindings: map[string]delivery.Provider{"db": {Node: "anchor", Module: "store"}}}), "lease write"}, + "a config place": {in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", Layer: map[string]any{"places": []any{}}}), "lease write"}, + "a config access": {in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", MeshWide: map[string]any{"accesses": []any{}}}), "lease write"}, "a module not held": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "laptop", Module: "ghost"}), "the indexer holds no module ghost"}, "a node with no record": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "stranger", Module: "web"}), "stranger is a node with no record"}, "nothing changes": {in(delivery.PlacementProposalChange{Op: delivery.OpPlace, Node: "anchor", Module: "web"}), "leave every placement as it is"}, @@ -52,6 +55,33 @@ func TestThePlacementSeatDecides(t *testing.T) { } } + // A module's config values may be proposed before the lease write (§13); once the lease is written every change is + // the state's alone, and none waits. + if _, _, f := Propose(in(delivery.PlacementProposalChange{Op: delivery.OpConfig, Node: "anchor", Module: "web", Values: map[string]any{"puid": 1}})); len(f) != 0 { + t.Errorf("config values before the lease write: %v", f) + } + Leased = true + for _, c := range []delivery.PlacementProposalChange{ + {Op: delivery.OpPlace, Node: "laptop", Module: "web", Ports: map[string]int{"80": 8080}, Bindings: map[string]delivery.Provider{"db": {Node: "anchor", Module: "store"}}}, + {Op: delivery.OpConfig, Node: "anchor", Module: "web", Layer: map[string]any{"places": []any{}}}, + {Op: delivery.OpDecide, Node: "laptop", Field: "site", Value: json.RawMessage(`"attic"`)}, + {Op: delivery.OpHold, Node: "anchor", Module: "store", Claim: "node-shell", Scope: "node"}, + } { + if _, _, f := Propose(in(c)); len(f) != 0 { + t.Errorf("after the lease write %s is refused: %v", c.Op, f) + } + } + Leased = false + + // A key moved while the warrant wrote: what is left to write is the other nodes' changes, to be laid and asked again. + if rest, ok := Rest(p, map[string]*delivery.Placement{"anchor": laid["anchor"]}); !ok || len(rest.Changes) != 1 || rest.Changes[0].Node != "laptop" || + len(rest.Base) != 1 || rest.State != delivery.ProposalAnswering || rest.Must != nil { + t.Errorf("the rest = %+v, %v", rest, ok) + } + if _, ok := Rest(p, laid); ok { + t.Error("a proposal written whole leaves something to write") + } + writes, err := Writes(p, laid) if err != nil || len(writes) != 2 || writes[0].Key != "placement.anchor" || writes[0].Base != 4 || writes[1].Base != 9 { t.Fatalf("writes = %+v, %v", writes, err) diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 8c7aa77..1803f4b 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -44,7 +44,7 @@ func TestManifest(t *testing.T) { states[s.Name] = s.History } if want := map[string]int{"asks": 5, "compositions": 5, "differences": 64, "puts": 20, "wants": 20, - "placement": 64, "proposals": 5, "candidates": 5}; !reflect.DeepEqual(states, want) { + "placement": 64, "proposals": 5, "candidates": 5, "pulses": 0}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 78db643..91b6bca 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -59,6 +59,7 @@ { "name": "placement", "history": 64 }, { "name": "proposals", "history": 5 }, { "name": "candidates", "history": 5 }, + { "name": "pulses", "key-ttl": true }, { "name": "asks", "history": 5 }, { "name": "compositions", "history": 5 }, { "name": "differences", "history": 64 },