The indexer writes placement.<node> from the catalogue's placement files at each merge, only where it changed, and says placement-changed (issue 1126, ADR 0549 §2)
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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.<node>` 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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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": [
|
||||
|
||||
Reference in New Issue
Block a user