Assigner: follow every proposal from its answers through the Approve to the write at its revisions, keeping the controller in step (ADR 0567 §3-§8, issue 1126)

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 {

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)

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.<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>.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)
+	}
+}