From 1718d4b1d7e7b20d976a660e97ac2577fa35ca3d Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:31:43 +0100 Subject: [PATCH] =?UTF-8?q?Assigner:=20follow=20every=20proposal=20from=20?= =?UTF-8?q?its=20answers=20through=20the=20Approve=20to=20the=20write=20at?= =?UTF-8?q?=20its=20revisions,=20keeping=20the=20controller=20in=20step=20?= =?UTF-8?q?(ADR=200567=20=C2=A73-=C2=A78,=20issue=201126)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cmd/mesh-delivery-assigner/main.go | 10 + .../cmd/mesh-delivery-assigner/pass.go | 16 +- .../cmd/mesh-delivery-assigner/seat.go | 695 ++++++++++++++++++ 3 files changed, 718 insertions(+), 3 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go 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 dbe0465..d2fca5e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -150,10 +150,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := said.Keeps(keptStates...); err != nil { logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err) } + thisNode = account.Node logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs)) return follow(ctx, js, conn, kvs, sayVia(ctx, js), emitting{said}, logger) } +// thisNode is the node this process runs on, as its credential names it: said with each of the seat's events. +var thisNode string + // emitting is the process connection as the puts' caller: its calls, and its events with no options. type emitting struct{ *process.Conn } @@ -318,6 +322,12 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, if bus != nil && kvs[delivery.AssignerPutsBucket] != nil { p.acts = newPutActs(bus, wake, logger.Printf) } + if bus != nil { + p.seat = newSeatActs(bus, wake, logger.Printf) + } + if js != nil { + p.publish = streamPublisher(ctx, js, thisNode) + } wake() for { select { 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 c1f7202..e584912 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -43,9 +43,18 @@ 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 placement candidate's answer bound (ADR 0559 §5), as a context's deadline - // (ADR 0437 §8: no timer). Nil asks none. + // 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. + 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 } func (p *passer) clock() time.Time { @@ -56,6 +65,7 @@ func (p *passer) clock() time.Time { } func (p *passer) pass(ctx context.Context) { + p.placementSeat(ctx) p.compositions(ctx) p.wants(ctx) p.puts(ctx) @@ -120,7 +130,7 @@ func (p *passer) wants(ctx context.Context) { } } wanted := p.w.wantsOf() - frozen := p.sayPlacement(p.w.placementDiffers()) + frozen := p.sayPlacement(p.withCalls(p.w.placementDiffers())) placed := p.w.placedNodes() puts := p.putsByModule() p.staleAsks(ctx, trunk) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go new file mode 100644 index 0000000..a9a9cc1 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat.go @@ -0,0 +1,695 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "reflect" + "sort" + "strings" + "sync" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" + "git.novox.be/novox/mesh-sdk/go/compose/frombus" + "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" + + "mesh-delivery-assigner/internal/assign" +) + +// The seat mesh-placement's holder in the assigner's process (novox/hq ADR 0567 §2-§8): each pass follows every open +// proposal one step, from what the world holds. A proposal the verbs wrote is said (placement-proposed); the nodes it +// must hear from are found by composing every node with and without it, and those its key does not change get a reached +// candidate; their answers are judged (placement-judged) and, on a passing verdict, the operator is asked to approve it; +// on the warrant each changed key is written by compare-and-set at the revision it was laid over (placement-changed), +// and the controller's assignments are kept in step through its verbs until the lease write. A key moved meanwhile +// lays the proposal again; a decline, an expiry or a withdrawal ends it (placement-withdrawn). Every event is said on the +// assigner's own stream, so only this process says them. Nothing polls: the wake at an answer's bound is the one wake. + +// publisher says one of the seat's events on the assigner's stream, under a message id the bus keeps one of. +type publisher func(event string, body any, id string) error + +// streamPublisher publishes on mesh.stream.mesh-delivery-assigner.event., with the headers an emit carries. +func streamPublisher(ctx context.Context, js jetstream.JetStream, node string) publisher { + return func(event string, body any, id string) error { + data, err := json.Marshal(body) + if err != nil { + return err + } + msg := nats.NewMsg(process.ModuleStreamSubject(delivery.AssignerModule, event)) + msg.Header.Set("x-event-id", id) + msg.Header.Set("x-source", delivery.AssignerModule) + msg.Header.Set("x-node", node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + _, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id)) + return err + } +} + +// seatActs keeps what the seat's calls of the controller came to while this process runs: the export read again at a +// seed's warrant, and each call that keeps the controller's assignments in step (§8). +type seatActs struct { + mu sync.Mutex + bus caller + wake func() + logf func(string, ...any) + export map[string]exportCheck + // calls is, by pair (node and module, or node and provision for a pin), what the last call for it came to. + calls map[string]string +} + +// exportCheck is the export read again for one proposal: running, its answer, or why it was not read. +type exportCheck struct { + running bool + export map[string]delivery.Placement + err string +} + +func newSeatActs(bus caller, wake func(), logf func(string, ...any)) *seatActs { + return &seatActs{bus: bus, wake: wake, logf: logf, export: map[string]exportCheck{}, calls: map[string]string{}} +} + +// exportOf is the export read again for a proposal, asked once in the background. +func (a *seatActs) exportOf(id string) exportCheck { + a.mu.Lock() + defer a.mu.Unlock() + if c, ok := a.export[id]; ok { + return c + } + a.export[id] = exportCheck{running: true} + go func() { + defer a.wake() + raw, err := a.bus.Call(exportCall, map[string]any{}) + c := exportCheck{} + if err != nil { + c.err = "the controller's placement-export did not answer: " + err.Error() + } else if c.export, err = assign.ExportOf(raw); err != nil { + c.err = err.Error() + } + a.mu.Lock() + a.export[id] = c + a.mu.Unlock() + }() + return exportCheck{running: true} +} + +// forget drops what is kept of a proposal's export, so a later warrant reads it again. +func (a *seatActs) forget(id string) { + if a == nil { + return + } + a.mu.Lock() + delete(a.export, id) + a.mu.Unlock() +} + +// record records once, in the controller's hand-act log, that placement was written on the operator's warrant (§5): the +// router's record of the teller's ask says who chose and how (ADR 0274). +func (a *seatActs) record(warrant string) { + if a == nil || warrant == "" { + return + } + ask, _, _ := strings.Cut(warrant, "@") + go func() { + if _, err := a.bus.Call("seat:"+catalogue.ControllerSeatName+".warranted", map[string]any{"asker": delivery.TellerModule, "ask": ask}); err != nil { + a.logf("the placement's warrant %s is not recorded in the hand-act log: %v", ask, 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 { + return + } + a.mu.Lock() + for _, c := range calls { + a.calls[c.Node+"."+c.Module] = c.Key() + " is asked" + } + a.mu.Unlock() + go func() { + defer a.wake() + for _, c := range calls { + words := c.Key() + " answered" + if _, err := a.bus.Call("seat:"+catalogue.ControllerSeatName+"."+c.Verb, c.Args); err != nil { + words = c.Key() + " refused: " + err.Error() + } + a.logf("keeping the controller in step with placement: %s", words) + a.mu.Lock() + a.calls[c.Node+"."+c.Module] = words + a.mu.Unlock() + } + }() +} + +// callOf is what the last call for a pair came to, "" when none was made. +func (a *seatActs) callOf(node, module string) string { + if a == nil { + return "" + } + a.mu.Lock() + defer a.mu.Unlock() + return a.calls[node+"."+module] +} + +// withCalls names, on each pair the two sources do not both name, the call the assigner made for it (§8). +func (p *passer) withCalls(pairs []assign.Pair) []assign.Pair { + for i := range pairs { + pairs[i].Call = p.seat.callOf(pairs[i].Node, pairs[i].Module) + } + return pairs +} + +// seatPlacements is every node's placement key and its revision, as the watch holds them. +func (p *passer) seatPlacements() (map[string]delivery.Placement, map[string]uint64) { + out, revs := map[string]delivery.Placement{}, map[string]uint64{} + for key, value := range p.w.read(delivery.PlacementBucket) { + node, err := delivery.ParsePlacementKey(key) + if err != nil { + continue + } + if pl, err := delivery.DecodePlacement(value); err == nil && pl.Node == node { + out[node], revs[node] = pl, p.w.metaOf(delivery.PlacementBucket, key).revision + } + } + return out, revs +} + +// candidatesOf is a proposal's candidates by node, with their keys' revisions and the newest write. +func (p *passer) candidatesOf(id string) (map[string]delivery.PlacementCandidate, map[string]uint64, time.Time) { + out, revs := map[string]delivery.PlacementCandidate{}, map[string]uint64{} + var newest time.Time + for key, value := range p.w.read(delivery.PlacementCandidatesBucket) { + pid, node, err := delivery.ParsePlacementCandidateKey(key) + if err != nil || pid != id { + continue + } + c, err := delivery.DecodePlacementCandidate(value) + if err != nil { + continue + } + meta := p.w.metaOf(delivery.PlacementCandidatesBucket, key) + out[node], revs[node] = c, meta.revision + if meta.at.After(newest) { + newest = meta.at + } + } + return out, revs, newest +} + +// answersTo is, by node, each node's answer to the proposal in its `.node` have. +func (p *passer) answersTo(id string) map[string]*delivery.CandidateAnswer { + out := map[string]*delivery.CandidateAnswer{} + for key, value := range p.w.read(delivery.HaveBucket) { + node, _, isNode, err := delivery.ParseHaveKey(key) + if err != nil || !isNode { + continue + } + h, err := delivery.DecodeHaveNode(value) + if err != nil { + continue + } + if a, ok := h.Candidates[id]; ok { + out[node] = &a + } + } + return out +} + +// placementSeat takes every open proposal one step further, and says the keys a hand-back deleted. +func (p *passer) placementSeat(ctx context.Context) { + if p.kvs[delivery.PlacementProposalsBucket] == nil || p.kvs[delivery.PlacementBucket] == nil || p.kvs[delivery.PlacementCandidatesBucket] == 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 { + ids = append(ids, id) + } + sort.Strings(ids) + for _, id := range ids { + prop, err := delivery.DecodePlacementProposal(raws[id]) + if err != nil || prop.ID != id { + p.logf("the proposal %s cannot be read, so it is left: %v", id, err) + continue + } + p.stepProposal(ctx, prop, p.w.metaOf(delivery.PlacementProposalsBucket, id).revision, current, revs) + } +} + +// 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] + } + return + } + var gone []delivery.PlacementChange + for node, before := range p.seen { + if _, there := current[node]; there { + continue + } + 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] + } + 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) + } +} + +// stepProposal takes one proposal one step further. +func (p *passer) stepProposal(ctx context.Context, prop delivery.PlacementProposal, rev uint64, current map[string]delivery.Placement, revs map[string]uint64) { + if !p.saidProposed[prop.ID] && prop.State == delivery.ProposalAnswering && prop.Verdict == nil { + p.sayEvent(delivery.PlacementProposedEvent, delivery.PlacementEvent{Proposal: prop}, prop.ID) + p.saidProposed[prop.ID] = true + } + if prop.Withdrawn != "" { + p.endProposal(ctx, prop, prop.Withdrawn) + return + } + akey, _ := delivery.PlacementAskKey(prop.ID) + if prop.State == delivery.ProposalAsked { + switch { + case p.answeredWith(akey, assign.OptionYes): + 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)) { + p.endProposal(ctx, prop, "its ask expired unanswered") + return + } + } + if moved(prop, revs) { + p.layAgain(ctx, prop, rev, current, revs, "a key it was laid over moved since") + return + } + cands, candRevs, written := p.candidatesOf(prop.ID) + if len(prop.Must) == 0 { + p.mustAnswer(ctx, prop, rev, current, cands) + return + } + for _, node := range prop.Must { + if _, ok := cands[node]; !ok { + return // a candidate just written is not heard yet: its watch wakes the next pass + } + } + 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) + } + if prop.Verdict != nil && sameVerdict(*prop.Verdict, v) && !again { + return + } + prop.Verdict = &v + switch v.State { + case delivery.VerdictFail: + prop.State, prop.Ask = delivery.ProposalFailed, "" + case delivery.VerdictWaits: + 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) + 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)" + } + } + } + if p.writeProposal(ctx, prop, rev) { + p.sayEvent(delivery.PlacementJudgedEvent, delivery.PlacementEvent{Proposal: prop}, + fmt.Sprintf("%s.judged.%d", prop.ID, v.At.UnixNano())) + } +} + +// sameVerdict says two verdicts say the same, their time aside. +func sameVerdict(a, b delivery.PlacementVerdict) bool { + a.At, b.At = time.Time{}, time.Time{} + return reflect.DeepEqual(a, b) +} + +// moved says whether a key the proposal changes is no longer at the revision it was laid over. +func moved(prop delivery.PlacementProposal, revs map[string]uint64) bool { + for node, base := range prop.Base { + if revs[node] != base { + return true + } + } + return false +} + +// intended is, by proposal and the revisions it was laid over, what Intended computed, kept while this process runs. +func (p *passer) intendedOf(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) map[string]map[string]bool { + key := fmt.Sprint(prop.ID, prop.Base) + if got, ok := p.intended[key]; ok { + return got + } + got, _, _ := p.compute(prop, current, cands) + p.intended[key] = got + return got +} + +// compute composes every node with the current placement and with the proposal laid over it (judge.Intended). +func (p *passer) compute(prop delivery.PlacementProposal, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) (map[string]map[string]bool, []string, []string) { + laid := map[string]*delivery.Placement{} + for _, node := range prop.Nodes() { + if c, ok := cands[node]; ok && !c.Reached { + laid[node] = c.Placement + } + } + get := kvGetter(p.kvs[delivery.ManifestsBucket]) + r := worldReader{w: p.w, manifests: get} + var now, with *compose.Input + if in, err := frombus.Input(context.Background(), r, frombus.Mesh{}, frombus.Scope{Every: true}); err == nil { + now = &in + } + if in, err := frombus.Input(context.Background(), r, frombus.Mesh{}, frombus.Scope{Every: true, Placements: laid}); err == nil { + with = &in + } + nodes := map[string]bool{} + for n := range current { + nodes[n] = true + } + if records, err := r.Records(context.Background()); err == nil { + for n := range records { + nodes[n] = true + } + } + names := make([]string, 0, len(nodes)) + for n := range nodes { + names = append(names, n) + } + fps := map[string]string{} + for name, m := range p.w.trunk() { + if m.FP2 != "" { + fps[name] = m.FP2 + } + } + return judge.Intended(now, with, current, laid, names, fps) +} + +// mustAnswer finds the nodes the proposal must hear from (§3), writes a reached candidate for each whose key it does not +// change, and writes them into the proposal. +func (p *passer) mustAnswer(ctx context.Context, prop delivery.PlacementProposal, rev uint64, current map[string]delivery.Placement, cands map[string]delivery.PlacementCandidate) { + for _, node := range prop.Nodes() { + if _, ok := cands[node]; !ok { + return // the verbs write the candidates before the proposal: wait to hear them + } + } + intended, must, notes := p.compute(prop, current, cands) + p.intended[fmt.Sprint(prop.ID, prop.Base)] = intended + for _, node := range must { + if _, ok := cands[node]; ok { + continue + } + key, err := delivery.PlacementCandidateKey(prop.ID, node) + if err != nil { + continue + } + c := delivery.PlacementCandidate{Proposal: prop.ID, Reached: true} + if pl, ok := current[node]; ok { + c.Placement = &pl + } + p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return delivery.EncodePlacementCandidate(c) }) + } + prop.Must = must + if len(notes) > 0 { + prop.Words = strings.TrimPrefix(prop.Words+"; "+strings.Join(notes, "; "), "; ") + } + p.writeProposal(ctx, prop, rev) +} + +// layAgain lays a proposal over the keys as they stand now (§5: a key moved): its candidates are written again, the +// nodes that must answer found again, and it is judged and asked again; a proposal that no longer lays ends. +func (p *passer) layAgain(ctx context.Context, prop delivery.PlacementProposal, rev uint64, current map[string]delivery.Placement, revs map[string]uint64, why string) { + held, recorded := p.lays(ctx) + again, laid, findings := assign.Propose(assign.ProposeIn{ID: prop.ID, By: prop.By, Why: prop.Why, At: prop.At, Changes: prop.Changes, + Current: current, Revisions: revs, Held: held, Recorded: recorded, + Seed: prop.Seed, ExportDigest: prop.ExportDigest}) + if len(findings) > 0 { + p.endProposal(ctx, prop, why+", and laid over the keys now it is refused: "+strings.Join(findings, "; ")) + return + } + p.deleteCandidates(ctx, prop.ID) + for _, node := range again.Nodes() { + key, _ := delivery.PlacementCandidateKey(prop.ID, node) + c := delivery.PlacementCandidate{Proposal: prop.ID, Placement: laid[node]} + p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return delivery.EncodePlacementCandidate(c) }) + } + again.Written, again.Words = p.clock().UTC(), why+": laid again over the keys as they stand, to be judged and asked again" + p.seat.forget(prop.ID) + p.writeProposal(ctx, again, rev) +} + +// lays is what a proposal is checked against, as the verbs check it: the modules the indexer holds, by key, and the +// nodes with a record. +func (p *passer) lays(ctx context.Context) (func(string) bool, func(string) bool) { + held := map[string]bool{} + for key := range p.w.read(delivery.ModulesBucket) { + if name, err := delivery.ParseModulesKey(key); err == nil { + held[name] = true + } + } + records, _ := worldReader{w: p.w}.Records(ctx) + return func(m string) bool { return held[m] }, func(n string) bool { _, ok := records[n]; return ok } +} + +// warranted writes an approved proposal (§5): for the seed, after the export is read again and found unchanged (§7); +// each changed key by compare-and-set at the revision it was laid over; then the warrant recorded, the candidates and the +// proposal deleted, placement-changed said, and the controller kept in step (§8). +func (p *passer) warranted(ctx context.Context, prop delivery.PlacementProposal, rev uint64, akey string, current map[string]delivery.Placement, revs map[string]uint64) { + if err := p.performs(akey, assign.PlacementAct(prop)); err != nil { + p.endProposal(ctx, prop, "its warrant does not bind these changes, so nothing is written: "+err.Error()) + return + } + answer, _ := p.w.answer(akey) + if prop.Seed { + if p.seat == nil { + return + } + c := p.seat.exportOf(prop.ID) + switch { + case c.running: + return // its answer wakes the next pass + case c.err != "": + if words := "the export cannot be read again at the warrant: " + c.err; prop.Words != words { + prop.Words = words + p.writeProposal(ctx, prop, rev) + } + return + case delivery.ExportDigest(c.export) != prop.ExportDigest: + p.endProposal(ctx, prop, "the controller's export changed since the seed was proposed: the new export is proposed as the seed") + p.proposeSeed(ctx, prop, c.export) + return + } + } + if moved(prop, revs) { + p.layAgain(ctx, prop, rev, current, revs, "a key moved since it was approved, so its warrant is spent") + return + } + cands, _, _ := p.candidatesOf(prop.ID) + laid := map[string]*delivery.Placement{} + for _, node := range prop.Nodes() { + laid[node] = cands[node].Placement + } + writes, err := assign.Writes(prop, laid) + if err != nil { + p.logf("the proposal %s cannot be written: %v", prop.ID, err) + return + } + kv := p.kvs[delivery.PlacementBucket] + var changes []delivery.PlacementChange + for _, w := range writes { + var after uint64 + var err error + switch { + case w.Value == nil: + err = kv.Delete(ctx, w.Key, jetstream.LastRevision(w.Base)) + 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) + } else { + after, 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") + return + } + before := current[w.Node] + after2 := delivery.Placement{Node: w.Node} + if w.Value != nil { + after2 = *w.Value + } + c := delivery.ChangeOf(before, after2) + c.Before, c.After = w.Base, after + 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) + } + } + } + p.seat.record(answer.Warrant) + p.deleteCandidates(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)) + } +} + +// sayChanged says placement-changed for the keys written. +func (p *passer) sayChanged(prop delivery.PlacementProposal, answer delivery.Answer, changes []delivery.PlacementChange) { + if len(changes) == 0 { + return + } + at := p.clock().UTC() + if _, when, ok := strings.Cut(answer.Warrant, "@"); ok { + if t, err := time.Parse(time.RFC3339Nano, when); err == nil { + at = t + } + } + delivery.SortChanges(changes) + p.sayEvent(delivery.PlacementChangedEvent, delivery.PlacementEvent{Proposal: prop, Nodes: changes, + Warrant: &delivery.PlacementWarrant{Warrant: answer.Warrant, Words: answer.Words, At: at}}, prop.ID+".changed") +} + +// proposeSeed proposes an export as the seed again, by the same proposer and why. +func (p *passer) proposeSeed(ctx context.Context, was delivery.PlacementProposal, export map[string]delivery.Placement) { + held, recorded := p.lays(ctx) + now := p.clock() + seed, laid, findings := assign.Propose(assign.ProposeIn{ID: delivery.NewProposalID(now), By: was.By, Why: was.Why, At: now, + Changes: assign.SeedChanges(export), Current: map[string]delivery.Placement{}, Revisions: map[string]uint64{}, + Held: held, Recorded: recorded, Seed: true, ExportDigest: delivery.ExportDigest(export)}) + if len(findings) > 0 { + p.logf("the new export is not proposed as the seed: %s", strings.Join(findings, "; ")) + return + } + for _, node := range seed.Nodes() { + key, _ := delivery.PlacementCandidateKey(seed.ID, node) + c := delivery.PlacementCandidate{Proposal: seed.ID, Placement: laid[node]} + p.put(ctx, delivery.PlacementCandidatesBucket, key, func() ([]byte, error) { return delivery.EncodePlacementCandidate(c) }) + } + seed.Written = now.UTC() + p.writeProposal(ctx, seed, 0) +} + +// endProposal ends a proposal: its ask, its candidates and its key deleted, and placement-withdrawn said carrying it +// whole with why it ended. +func (p *passer) endProposal(ctx context.Context, prop delivery.PlacementProposal, why string) { + if akey, err := delivery.PlacementAskKey(prop.ID); err == nil && p.w.has(delivery.AssignerAsksBucket, akey) { + if err := p.kvs[delivery.AssignerAsksBucket].Delete(ctx, akey); err == nil { + p.w.put(delivery.AssignerAsksBucket, akey, nil) + } + } + p.deleteCandidates(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 + } + p.w.put(delivery.PlacementProposalsBucket, prop.ID, nil) + p.seat.forget(prop.ID) + p.logf("the proposal %s ends: %s", prop.ID, why) + p.sayEvent(delivery.PlacementWithdrawnEvent, delivery.PlacementEvent{Proposal: prop, Ended: why}, prop.ID+".withdrawn") +} + +// deleteCandidates deletes every candidate of a proposal. +func (p *passer) deleteCandidates(ctx context.Context, id string) { + for key := range p.w.read(delivery.PlacementCandidatesBucket) { + if pid, _, err := delivery.ParsePlacementCandidateKey(key); err == nil && pid == id { + if err := p.kvs[delivery.PlacementCandidatesBucket].Delete(ctx, key); err == nil { + p.w.put(delivery.PlacementCandidatesBucket, key, nil) + } + } + } +} + +// writeProposal writes a proposal over the revision the pass read, by compare-and-set: a verb that marked it meanwhile +// (withdraw, again) wins, and the next pass reads that. It says whether it wrote. +func (p *passer) writeProposal(ctx context.Context, prop delivery.PlacementProposal, rev uint64) bool { + raw, err := delivery.EncodePlacementProposal(prop) + if err != nil { + p.logf("the proposal %s is not written: %v", prop.ID, err) + return false + } + kv := p.kvs[delivery.PlacementProposalsBucket] + var at uint64 + if rev == 0 { + at, err = kv.Create(ctx, prop.ID, raw) + } else { + at, err = kv.Update(ctx, prop.ID, raw, rev) + } + if err != nil { + if !errors.Is(err, context.Canceled) { + p.logf("the proposal %s is not written, and is taken further at the next pass: %v", prop.ID, err) + } + return false + } + p.w.putEntry(delivery.PlacementProposalsBucket, prop.ID, raw, at, p.clock()) + return true +} + +// say says one of the seat's events, said in the journal when the bus refuses it. +func (p *passer) sayEvent(event string, body delivery.PlacementEvent, id string) { + if p.publish == nil { + return + } + if err := p.publish(event, body, id); err != nil { + p.logf("%s of %s is not said: %v", event, body.Proposal.ID, err) + } +}