diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 0b4258d..6aa4776 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -253,6 +253,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + placement, err := process.EssentialBucket(ctx, js, "indexer", delivery.PlacementBucket, "write a node's placement (ADR 0549 §2)") + if err != nil { + return err + } // The controller's build records only name the directory of a module whose key names none (fillPaths): without them // such a module waits for its path, every other is indexed (issue 1945). builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf) @@ -287,7 +291,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer calls.Close() ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Builds: known, Logf: logger.Printf, - Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call)} + Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), + Placement: kvStore{placement}, Say: func(event string, body any) error { return calls.Emit(event, body) }} // The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key // main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does // not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate. diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 5a8d354..83d542f 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026171025-5db53e738f3a github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index e793069..783167c 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026171025-5db53e738f3a h1:JjQUrojy2rfxVlkTDC8tkP8yPh+rZ57mpuiPWqPKy3o= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026171025-5db53e738f3a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 592b309..cd48a70 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -60,6 +60,10 @@ type Indexer struct { // Contains says whether a commit is the second or an ancestor of it (gitea.gitea_contains). Contains func(ctx context.Context, repository, commit, in string) (bool, error) Logf func(format string, args ...any) + // Placement is the placement bucket (delivery.PlacementBucket, ADR 0549 §2); nil writes no placement. + Placement Store + // Say says one of the indexer's events (placement-changed); nil says none. + Say func(event string, body any) error mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach source sourceNow @@ -96,6 +100,9 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { } ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945) } + if err := ix.applyPlacement(ctx, f); err != nil { + return err + } var digests map[string]string var held map[string]delivery.Module if f.Kind == delivery.ForgePull { diff --git a/modules/mesh-delivery-indexer/internal/index/placement.go b/modules/mesh-delivery-indexer/internal/index/placement.go new file mode 100644 index 0000000..8fbc357 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/placement.go @@ -0,0 +1,115 @@ +package index + +import ( + "context" + "fmt" + "maps" + "slices" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Placement of modules on nodes (novox/hq ADR 0549 §2, issue 1126): at each merge of the catalogue the forge bridge +// carries every placement file of the merge commit, and the indexer writes each node's key `placement.` in its +// state `placement` only when it differs (the revision alone is no change), deletes the key of a node whose file is +// gone, and says `placement-changed` for each node it wrote or deleted, naming the modules added, removed or changed. +// A merge that changes no placement writes nothing and says nothing. +// +// A file that does not read as a placement, or names another node than its file, is said in the log and its key is left +// as it was: the catalogue's own check refuses such a file before it merges (ADR 0549 §1), so it is a fault of that +// check, and a node keeps composing from the placement it had rather than from none. + +// applyPlacement takes the placement files of a merged message in; a message that read none (another repository, an +// older bridge) changes nothing. +func (ix *Indexer) applyPlacement(ctx context.Context, f delivery.Forge) error { + if ix.Placement == nil || f.Kind != delivery.ForgeMerged || !f.PlacementsRead { + return nil + } + keys, err := ix.Placement.Keys(ctx) + if err != nil { + return fmt.Errorf("reading the placement keys: %w", err) + } + gone := map[string]bool{} + for _, key := range keys { + if node, err := delivery.ParsePlacementKey(key); err == nil { + gone[node] = true + } + } + var refused []string + for _, node := range slices.Sorted(maps.Keys(f.Placements)) { + delete(gone, node) + next, err := delivery.DecodePlacement([]byte(f.Placements[node])) + if err == nil && next.Node != node { + err = fmt.Errorf("the file of %s names the node %q", node, next.Node) + } + if err != nil { + refused = append(refused, node) + ix.logf("%s at %s: the placement file of %s is refused and placement.%s is left as it was (the catalogue's check should have refused it): %v", + f.Repository, f.Merge, node, node, err) + continue + } + next.Revision = f.Merge + if err := ix.writePlacement(ctx, node, &next); err != nil { + return err + } + } + for _, node := range slices.Sorted(maps.Keys(gone)) { + if err := ix.writePlacement(ctx, node, nil); err != nil { + return err + } + } + if len(refused) > 0 { + // Said, and the merge's candidates are still taken out: a refused file holds nobody's delivery. + ix.logf("%s at %s: the placement files of %v do not read; their keys are left as they were", f.Repository, f.Merge, refused) + } + return nil +} + +// writePlacement writes one node's key, or deletes it for next nil, only when it differs, and says what changed. +func (ix *Indexer) writePlacement(ctx context.Context, node string, next *delivery.Placement) error { + key, err := delivery.PlacementKey(node) + if err != nil { + return err + } + var change *delivery.PlacementChange + err = write(ctx, ix.Placement, key, func(value []byte, found bool) ([]byte, bool, error) { + change = nil + var before delivery.Placement + if found { + cur, err := delivery.DecodePlacement(value) + if err != nil { + ix.logf("placement.%s holds what is not a placement, written again: %v", node, err) + } else { + before = cur + } + } + if next == nil { + if !found { + return nil, false, nil + } + c := delivery.ChangeOf(before, delivery.Placement{Node: node}) + change = &c + return nil, true, nil + } + if found && delivery.SamePlacement(before, *next) { + return nil, false, nil + } + data, err := delivery.EncodePlacement(*next) + if err != nil { + return nil, false, err + } + c := delivery.ChangeOf(before, *next) + change = &c + return data, false, nil + }) + if err != nil || change == nil { + return err + } + if ix.Say != nil { + if err := ix.Say(delivery.PlacementChangedEvent, *change); err != nil { + // The key is written: a reader that watches it hears the change; the event is said loudly as not said. + ix.logf("placement.%s is written, and %s could not be said: %v", node, delivery.PlacementChangedEvent, err) + } + } + return nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/placement_test.go b/modules/mesh-delivery-indexer/internal/index/placement_test.go new file mode 100644 index 0000000..1d74fbd --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/placement_test.go @@ -0,0 +1,97 @@ +package index + +import ( + "context" + "reflect" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ADR 0549 §2 (issue 1126): a merge changing one node's file writes that key alone and says one event; a merge with no +// placement change writes nothing and says nothing; a node whose file is gone loses its key and says its modules +// removed; a file that does not read leaves its key; a message that read no placement changes none. +func TestPlacementIsWrittenAtAMergeOnlyWhereItChanged(t *testing.T) { + ctx := context.Background() + ix, _, logs := newIndexer(t) + place := newMem() + ix.Placement = place + var said []delivery.PlacementChange + ix.Say = func(event string, body any) error { + if event != delivery.PlacementChangedEvent { + t.Errorf("said %s", event) + } + said = append(said, body.(delivery.PlacementChange)) + return nil + } + merged := func(merge string, files map[string]string) delivery.Forge { + return delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: delivery.CatalogueRepository, Branch: "main", + Number: 7, Head: "h", Merge: merge, Root: "r", Trees: map[string]string{"": "t"}, Placements: files, PlacementsRead: true} + } + ace := `{"v":1,"node":"ace","modules":[{"module":"plex"}]}` + g14 := `{"v":1,"node":"g14","modules":[{"module":"zsh"}]}` + written := func(node string) (delivery.Placement, uint64) { + t.Helper() + v, ok := place.vals["placement."+node] + if !ok { + return delivery.Placement{}, 0 + } + p, err := delivery.DecodePlacement(v) + if err != nil { + t.Fatal(err) + } + return p, place.revs["placement."+node] + } + + if err := ix.Apply(ctx, merged("m1", map[string]string{"ace": ace, "g14": g14})); err != nil { + t.Fatal(err) + } + if p, _ := written("ace"); p.Revision != "m1" || !reflect.DeepEqual(p.Assigned(), []string{"plex"}) || len(said) != 2 { + t.Fatalf("first merge: %+v, said %v", p, said) + } + _, aceRev := written("ace") + + said = nil + if err := ix.Apply(ctx, merged("m2", map[string]string{"ace": ace, "g14": g14})); err != nil { + t.Fatal(err) + } + if p, rev := written("ace"); rev != aceRev || p.Revision != "m1" || len(said) != 0 { + t.Errorf("a merge with no placement change wrote %+v (revision %d, was %d) and said %v", p, rev, aceRev, said) + } + + said = nil + if err := ix.Apply(ctx, merged("m3", map[string]string{"ace": ace, "g14": `{"v":1,"node":"g14","modules":[{"module":"zsh"},{"module":"xterm"}]}`})); err != nil { + t.Fatal(err) + } + if _, rev := written("ace"); rev != aceRev || len(said) != 1 || said[0].Node != "g14" || said[0].Revision != "m3" || + !reflect.DeepEqual(said[0].Added, []string{"xterm"}) { + t.Errorf("one file changed: ace's revision %d (was %d), said %+v", rev, aceRev, said) + } + + said = nil + if err := ix.Apply(ctx, merged("m4", map[string]string{"ace": ace, "g14": `{"v":1,"node":"ace","modules":[]}`})); err != nil { + t.Fatal(err) + } + if p, _ := written("g14"); len(p.Modules) != 2 || len(said) != 0 || !strings.Contains(logs.String(), "the placement file of %s is refused") { + t.Errorf("a file naming another node replaced the key: %+v, said %v, logs %s", p, said, logs) + } + + said = nil + if err := ix.Apply(ctx, merged("m5", map[string]string{"ace": ace})); err != nil { + t.Fatal(err) + } + if _, rev := written("g14"); rev != 0 || len(said) != 1 || !reflect.DeepEqual(said[0].Removed, []string{"xterm", "zsh"}) { + t.Errorf("a file removed: g14's key at revision %d, said %+v", rev, said) + } + + said = nil + unread := merged("m6", nil) + unread.PlacementsRead = false + if err := ix.Apply(ctx, unread); err != nil { + t.Fatal(err) + } + if p, _ := written("ace"); p.Revision != "m1" || len(said) != 0 { + t.Errorf("a message that read no placement changed it: %+v, %v", p, said) + } +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 46b3589..8becfb2 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -38,6 +38,7 @@ type manifest struct { Version string `json:"version"` Slug string `json:"slug"` Tools []string `json:"tools"` + Emits []string `json:"emits"` Consumes []string `json:"consumes"` State []state `json:"state"` Reads []string `json:"reads"` @@ -80,7 +81,11 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if m.Module != "mesh-delivery-indexer" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { + // 1126 (ADR 0549 §2): the indexer's placement state, and the event it says when a node's placement changed. + if want := []string{"placement-changed"}; !reflect.DeepEqual(m.Emits, want) { + t.Errorf("emits = %v, want %v", m.Emits, want) + } + if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { @@ -107,8 +112,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []string{"indexer_take_in"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } - // No emits, no streams, no bus, no seat (ADR 0437 §4). - for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { + // No streams, no bus, no seat (ADR 0437 §4); its one event is placement-changed (ADR 0549 §2), checked above. + for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the indexer has none", k) } diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index a30c24e..ce8432b 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -5,6 +5,9 @@ "tools": [ "indexer_take_in" ], + "emits": [ + "placement-changed" + ], "consumes": [ "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", @@ -15,6 +18,7 @@ { "name": "candidates", "history": 10 }, { "name": "manifests", "history": 1 }, { "name": "modules", "history": 20 }, + { "name": "placement", "history": 10 }, { "name": "takeins", "history": 1 } ], "reads": [