From 06df6a6167f9c51c91ef47170086d3220ba28e64 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 28 Oct 2026 18:31:44 +0100 Subject: [PATCH] =?UTF-8?q?Assigner:=20prove=20a=20proposal=20answered,=20?= =?UTF-8?q?approved=20and=20written=20whole=20on=20the=20bus,=20and=20a=20?= =?UTF-8?q?moved=20key=20laid=20again=20(ADR=200567=20=C2=A75,=20issue=201?= =?UTF-8?q?126)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../integration_test.go | 180 ++++++++++++++++++ 1 file changed, 180 insertions(+) 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 3575068..1eb6ed4 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 @@ -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 `.`. +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") + } +}