Assigner: prove a proposal answered, approved and written whole on the bus, and a moved key laid again (ADR 0567 §5, issue 1126)

2026-10-28 18:31:44 +01:00
parent 1718d4b1d7
commit 06df6a6167
1 changed files with 180 additions and 0 deletions
@@ -9,6 +9,7 @@ import (
 	"io"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -17,6 +18,7 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"

+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/announce"
 	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -1000,3 +1002,181 @@ func TestAHoldWrittenAgainAfterALiftAndARestartIsAskedAgain(t *testing.T) {
 		}
 	}
 }
+
+// bedStore is the runtime's states over the bed's buckets, as the seat's verbs read and write them: the assigner's own by
+// their local names, another module's `<module>.<state>`.
+type bedStore struct{ b *bed }
+
+func (s bedStore) kv(state string) jetstream.KeyValue {
+	if strings.Contains(state, ".") {
+		return s.b.buckets[strings.Replace(state, ".", "_", 1)]
+	}
+	return s.b.buckets[delivery.AssignerModule+"_"+state]
+}
+
+func (s bedStore) get(state, key string) ([]byte, uint64, bool, error) {
+	e, err := s.kv(state).Get(s.b.ctx, key)
+	if err != nil {
+		return nil, 0, false, nil
+	}
+	return e.Value(), e.Revision(), true, nil
+}
+
+func (s bedStore) put(state, key string, value []byte) error {
+	_, err := s.kv(state).Put(s.b.ctx, key, value)
+	return err
+}
+
+func (s bedStore) keys(state string) ([]string, error) {
+	keys, err := s.kv(state).Keys(s.b.ctx)
+	if err != nil && strings.Contains(err.Error(), "no keys") {
+		return nil, nil
+	}
+	return keys, err
+}
+
+func (s bedStore) del(state, key string) error { return s.kv(state).Delete(s.b.ctx, key) }
+
+func (s bedStore) history(string, string) ([]stdio.StateRevision, error) { return nil, nil }
+
+// 1126 (ADR 0567 §2-§6), the seat mesh-placement whole on the bus: a proposal moving web from the anchor to the laptop is
+// proposed through the verb, the assigner finds both nodes must answer, they answer the same in their haves, the
+// operator is asked and approves, and each key is written at a new revision while placement-changed is heard on the
+// assigner's stream naming both nodes. A second proposal laid over the old revision of a key moved since is laid again,
+// and declined, writes nothing and is withdrawn.
+func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
+	b := startBed(t, "")
+	if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "MOD_" + delivery.AssignerModule + "_placement",
+		Subjects: []string{process.ModuleStreamSubject(delivery.AssignerModule, ">")}}); err != nil {
+		t.Fatal(err)
+	}
+	heard := make(chan delivery.PlacementEvent, 16)
+	nc, err := nats.Connect(b.srv.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	if _, err := nc.Subscribe(process.ModuleStreamSubject(delivery.AssignerModule, delivery.PlacementChangedEvent), func(m *nats.Msg) {
+		var e delivery.PlacementEvent
+		if json.Unmarshal(m.Data, &e) == nil {
+			heard <- e
+		}
+	}); err != nil {
+		t.Fatal(err)
+	}
+	for _, node := range []string{"anchor", "laptop"} {
+		rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: node}})
+		b.putJSON(t, delivery.HaveBucket, node+".node", rec)
+	}
+	b.putJSON(t, delivery.ModulesBucket, "web", []byte(`{"v":1}`))
+	b.putJSON(t, delivery.ModulesBucket, "shell", []byte(`{"v":1}`))
+	place := func(node string, modules ...string) uint64 {
+		p := delivery.Placement{Node: node, Modules: []delivery.PlacedModule{}}
+		for _, m := range modules {
+			p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
+		}
+		raw, _ := delivery.EncodePlacement(p)
+		key, _ := delivery.PlacementKey(node)
+		rev, err := b.buckets[delivery.PlacementBucket].Put(b.ctx, key, raw)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return rev
+	}
+	anchorRev, laptopRev := place("anchor", "web"), place("laptop", "shell")
+	b.start(t)
+	verbs := newSeatTools(bedStore{b}, func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"ok":true}`), nil })
+	propose := func() delivery.PlacementProposal {
+		t.Helper()
+		got, err := verbs.doPropose(map[string]any{"why": "move web", "changes": `[{"op":"take-off","node":"anchor","module":"web"},` +
+			`{"op":"place","node":"laptop","module":"web"}]`}, "operator@laptop")
+		if err != nil {
+			t.Fatal(err)
+		}
+		return got.(proposed).Proposal
+	}
+	proposalNow := func(id, what string, ok func(delivery.PlacementProposal) bool) delivery.PlacementProposal {
+		t.Helper()
+		var last delivery.PlacementProposal
+		for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+			if e, err := b.buckets[delivery.PlacementProposalsBucket].Get(b.ctx, id); err == nil {
+				if p, err := delivery.DecodePlacementProposal(e.Value()); err == nil {
+					if last = p; ok(p) {
+						return p
+					}
+				}
+			}
+		}
+		t.Fatalf("the proposal %s never was %s: %+v", id, what, last)
+		return last
+	}
+	answerAll := func(p delivery.PlacementProposal) {
+		t.Helper()
+		for _, node := range p.Must {
+			key, _ := delivery.PlacementCandidateKey(p.ID, node)
+			e, err := b.buckets[delivery.PlacementCandidatesBucket].Get(b.ctx, key)
+			if err != nil {
+				t.Fatalf("no candidate %s: %v", key, err)
+			}
+			have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: node}, Candidates: map[string]delivery.CandidateAnswer{
+				p.ID: {Proposal: p.ID, Revision: e.Revision(), Verdict: delivery.CandidateSame, Same: []string{"node"}}}})
+			b.putJSON(t, delivery.HaveBucket, node+".node", have)
+		}
+	}
+
+	p := propose()
+	p = proposalNow(p.ID, "told who must answer", func(p delivery.PlacementProposal) bool { return len(p.Must) > 0 })
+	if !slices.Equal(p.Must, []string{"anchor", "laptop"}) {
+		t.Fatalf("must answer = %v, want both nodes", p.Must)
+	}
+	answerAll(p)
+	p = proposalNow(p.ID, "asked", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalAsked })
+	akey, _ := delivery.PlacementAskKey(p.ID)
+	b.answer(t, akey, assign.OptionYes)
+	select {
+	case e := <-heard:
+		if e.Proposal.ID != p.ID || len(e.Nodes) != 2 || e.Nodes[0].Before != anchorRev || e.Nodes[1].Before != laptopRev ||
+			e.Nodes[0].After <= anchorRev || e.Warrant == nil || !slices.Equal(e.Nodes[0].Removed, []string{"web"}) {
+			t.Errorf("placement-changed = %+v", e)
+		}
+	case <-time.After(15 * time.Second):
+		t.Fatal("no placement-changed heard")
+	}
+	for node, want := range map[string][]string{"anchor": {}, "laptop": {"shell", "web"}} {
+		key, _ := delivery.PlacementKey(node)
+		e, err := b.buckets[delivery.PlacementBucket].Get(b.ctx, key)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if got, err := delivery.DecodePlacement(e.Value()); err != nil || !slices.Equal(got.Assigned(), want) {
+			t.Errorf("%s = %v, %v, want %v", node, got.Assigned(), err, want)
+		}
+	}
+	if b.hasKey(delivery.PlacementProposalsBucket, p.ID) {
+		t.Error("an applied proposal stays in the state")
+	}
+
+	// A proposal whose key moves before its warrant is laid again; then declined, it writes nothing and ends.
+	q, err := verbs.doPropose(map[string]any{"why": "back", "changes": `[{"op":"take-off","node":"laptop","module":"web"}]`}, "op")
+	if err != nil {
+		t.Fatal(err)
+	}
+	moved := q.(proposed).Proposal
+	place("laptop", "shell", "web") // another writer moved the key
+	relaid := proposalNow(moved.ID, "laid again", func(p delivery.PlacementProposal) bool {
+		return strings.Contains(p.Words, "laid again") && len(p.Must) > 0
+	})
+	answerAll(relaid)
+	proposalNow(moved.ID, "asked", func(p delivery.PlacementProposal) bool { return p.State == delivery.ProposalAsked })
+	akey, _ = delivery.PlacementAskKey(moved.ID)
+	b.answer(t, akey, assign.OptionNo)
+	for deadline := time.Now().Add(15 * time.Second); b.hasKey(delivery.PlacementProposalsBucket, moved.ID); time.Sleep(50 * time.Millisecond) {
+		if time.Now().After(deadline) {
+			t.Fatal("a declined proposal was not ended")
+		}
+	}
+	key, _ := delivery.PlacementKey("laptop")
+	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")
+	}
+}