Revert "Merge pull request 'Revert L3: its node-engine removed every held module on novox, the bus server among them (issue 774)' (#109) from revert/774-l3 into main"

This reverts commit c1b4533201, reversing changes made to d8274204fb.
2026-10-29 21:05:14 +01:00
parent 883d976b5f
commit b01cf63237
74 changed files with 2550 additions and 1574 deletions
@@ -19,7 +19,11 @@ import (
 func serve(t *testing.T) string {
 	t.Helper()
 	m := view.New()
-	m.Apply(view.Entry{Bucket: view.AssignmentsBucket, Key: "a.x", Value: []byte(`{"node":"a","module":"x","build":"b1"}`)})
+	placed, err := delivery.EncodePlacement(delivery.Placement{Node: "a", Modules: []delivery.PlacedModule{{Module: "x"}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	m.Apply(view.Entry{Bucket: delivery.PlacementBucket, Key: "placement.a", Value: placed})
 	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "c")
 	if err != nil {
 		t.Fatal(err)
@@ -166,6 +166,15 @@ func Tools(r view.Reader, u view.UsageReader) []stdio.Tool {
 				node, module := nodeModuleOf(args)
 				return r.Haves(node, module)
 			}},
+		{Name: view.TrustsReadTool,
+			Description: "What every node trusts a mesh signature from, in one call (novox/hq ADR 0550 §2): each node's record's trusts " +
+				"(from node-engine_have <node>.node), whether it names the vault's signing key as mesh-vault publishes it in its " +
+				"state signers and whether it still names the controller's, the nodes that do not name the vault's yet (not_yet), " +
+				"and every: true once all do, the gate before the controller's lease is handed away. Read-only.",
+			Input: map[string]any{"type": "object", "properties": map[string]any{}},
+			Run: func(map[string]any) (any, error) {
+				return r.Trusts()
+			}},
 	}
 }

@@ -223,8 +232,9 @@ func read(state stdio.KeptState) view.StateRead {
 func main() {
 	r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates")),
 		WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have")),
-		ReachState: read(stdio.State("reach")),
-		HeardState: read(stdio.State("mesh-delivery-indexer.heard"))}
+		ReachState:   read(stdio.State("reach")),
+		HeardState:   read(stdio.State("mesh-delivery-indexer.heard")),
+		SignersState: read(stdio.State("mesh-vault.signers"))}
 	u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")),
 		Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())}
 	for _, lang := range delivery.AskLanguages {
@@ -1,13 +1,17 @@
 package main

 import (
+	"crypto/ed25519"
 	"encoding/json"
 	"fmt"
+	"reflect"
 	"strings"
 	"sync/atomic"
 	"testing"
 	"time"

+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
 	"mesh-delivery-view/internal/view"
 )

@@ -438,6 +442,41 @@ func TestAssignerDifferencesAndNodeHavesToolsAnswerAsStored(t *testing.T) {
 		t.Errorf("haves of ace (the node's own key too) = %v", got)
 	}

+	// L3 s6b (ADR 0550 §2): node_trusts answers for every node in one call whether its record names the vault's key the
+	// vault publishes, and every is true only once all do.
+	vaultPub, _, _ := ed25519.GenerateKey(nil)
+	vaultKey, _ := json.Marshal(delivery.NewSignerKey(vaultPub, time.Now()))
+	vaultFP := delivery.SignerFingerprint(vaultPub)
+	record := func(node string, trusts ...string) string {
+		raw, _ := delivery.EncodeHaveNode(delivery.HaveNode{V: delivery.Version, Record: &delivery.NodeRecord{Node: node, Trusts: trusts}})
+		return string(raw)
+	}
+	for _, row := range []struct {
+		name   string
+		haves  fake
+		every  bool
+		notYet []string
+	}{
+		{"one node trusts the controller's alone", fake{"ace.node": record("ace", "controller key sha256:aa", "vault key "+vaultFP),
+			"g14.node": record("g14", "controller key sha256:aa"), "ace.zsh": `{"fp":"x"}`}, false, []string{"g14"}},
+		{"every node trusts the vault's key", fake{"ace.node": record("ace", "vault key "+vaultFP),
+			"g14.node": record("g14", "controller key sha256:aa", "vault key "+vaultFP)}, true, []string{}},
+	} {
+		tr := view.Reader{HavesState: row.haves.reader().State, SignersState: fake{"key": string(vaultKey)}.reader().State}
+		out, err := run(t, tr, view.TrustsReadTool, map[string]any{})
+		if err != nil {
+			t.Fatalf("%s: %v", row.name, err)
+		}
+		a := out.(view.TrustsAnswer)
+		if a.Every != row.every || !reflect.DeepEqual(a.NotYet, row.notYet) || a.VaultKey != vaultFP || len(a.Nodes) != 2 {
+			t.Errorf("%s: %+v", row.name, a)
+		}
+	}
+	if out, err := run(t, view.Reader{HavesState: fake{"ace.node": record("ace", "vault key "+vaultFP)}.reader().State,
+		SignersState: fake{}.reader().State}, view.TrustsReadTool, map[string]any{}); err != nil || out.(view.TrustsAnswer).Every {
+		t.Errorf("with no key published: %+v, %v", out, err)
+	}
+
 	// More haves than one call reads (4 nodes, 300 each, 1 ms a Get): a node and a module read that one key and list
 	// nothing; the whole answer reads MaxStateReads values in parallel and says it was cut.
 	slow := newSlow(300, time.Millisecond, func(n string, i int) string { return fmt.Sprintf("%s.m%03d", n, i) })
@@ -48,7 +48,7 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8
 	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 	defer cancel()
 	kvs := map[string]jetstream.KeyValue{}
-	for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} {
+	for _, b := range []string{delivery.HaveBucket, delivery.PlacementBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} {
 		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5})
 		if err != nil {
 			t.Fatal(err)
@@ -68,9 +68,20 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8
 		}
 		return b
 	}
-	put(view.AssignmentsBucket, "a.x", []byte(`{"node":"a","module":"x","build":"b1"}`))
-	put(view.AssignmentsBucket, "a.y", []byte(`{"node":"a","module":"y","build":"b2"}`))
-	put(view.AssignmentsBucket, "b.x", []byte(`{"node":"b","module":"x","build":"b1","ended":"2026-10-18T11:30:00Z"}`))
+	// The seat's placement (issue 774, ADR 0549 §3): x and y on a, nothing on b.
+	placeValue := func(node string, modules ...string) []byte {
+		p := delivery.Placement{Node: node}
+		for _, m := range modules {
+			p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
+		}
+		v, err := delivery.EncodePlacement(p)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return v
+	}
+	put(delivery.PlacementBucket, "placement.a", placeValue("a", "x", "y"))
+	put(delivery.PlacementBucket, "placement.b", placeValue("b"))
 	put(view.BuildsBucket, "x.build.b1", []byte(`{"id":"b1","module":"x"}`))
 	put(view.BuildsBucket, "x.newest", []byte("b1"))
 	put(delivery.HaveBucket, "a.x", haveValue("b1", delivery.PhaseChecking, delivery.HealthWaits))
@@ -150,7 +161,7 @@ func TestProcessWholeShowsWhatTheWatchmanSays(t *testing.T) { // 1168/AC5
 	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 	defer cancel()
 	kvs := map[string]jetstream.KeyValue{}
-	for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} {
+	for _, b := range []string{delivery.HaveBucket, delivery.PlacementBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} {
 		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5})
 		if err != nil {
 			t.Fatal(err)
@@ -236,7 +247,7 @@ func TestProcessWholeShowsTheIndexersBucketsOnceTheyExist(t *testing.T) { // 124
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
 	defer cancel()
-	for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} {
+	for _, b := range []string{delivery.HaveBucket, delivery.PlacementBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket} {
 		if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5}); err != nil {
 			t.Fatal(err)
 		}
@@ -1,4 +1,4 @@
-// Command mesh-delivery-view is the operator's view of delivery (novox/hq ADR 0429 §3): it watches the have, assignments
+// Command mesh-delivery-view is the operator's view of delivery (novox/hq ADR 0429 §3): it watches the have, the seat's placement
 // and builds buckets on its own bus credential, derives one row per module and node in memory, and serves them on a
 // local socket (for mesh-delivery-cli) and on a local web page. It starts, stops or retries nothing; the one thing it writes is the operator's answer to an ask that needs no proof of who
 // answered, under `<ask key>.view` in its bucket answers (ADR 0528 clause 6).
@@ -106,7 +106,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	})
 	model.SetBacklog(view.BacklogOver(js, 5*time.Second))
-	for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket, delivery.TakenBucket} {
+	for _, bucket := range []string{delivery.HaveBucket, delivery.PlacementBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket, delivery.TakenBucket} {
 		go follow(ctx, js, bucket, false, model, logger)
 	}
 	// The modules' and the source's history give each fingerprint's first commit and each head's source revision (ADR 0556).