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 4c2a403..40a914d 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -317,6 +317,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Its placement state (ADR 0549 §2) is bound when it appears, as every state a delivery adds (issue 1790): until then a // catalogue merge's placement is said and not written, and the rest of the indexer runs. placement := process.AwaitBucket(ctx, js, conn, delivery.PlacementBucket, "no catalogue merge's placement is written", logger.Printf) + // Its placement candidates (issue 1126, ADR 0549 §4) are bound the same way: until the state appears a catalogue pull + // request's placement is said and not written, so no node answers it and its check names every placed node as silent. + placementCandidates := process.AwaitBucket(ctx, js, conn, delivery.PlacementCandidatesBucket, "no catalogue pull request's placement candidate is written", logger.Printf) // 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) @@ -352,7 +355,7 @@ 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}, Taken: awaitedStore{taken}, Builds: known, Logf: logger.Printf, Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), - Placement: awaitedStore{placement}, Say: func(event string, body any) error { return calls.Emit(event, body) }} + Placement: awaitedStore{placement}, PlacementCandidates: awaitedStore{placementCandidates}, 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 285279f..f7e08a9 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.20261026211913-68d65e6bbf61 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa 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 b3dab4f..2742997 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,9 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f h1:aWqn4HHTPUp/+28h7b8uFqQptAYvQFs05CALMHN7SkI= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027010747-7e99c354ac7f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/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 80d9abf..656184d 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -67,6 +67,9 @@ type Indexer struct { Logf func(format string, args ...any) // Placement is the placement bucket (delivery.PlacementBucket, ADR 0549 §2); nil writes no placement. Placement Store + // PlacementCandidates is the placement candidates bucket (delivery.PlacementCandidatesBucket, issue 1126); nil writes + // no candidate (placement_candidates.go). + PlacementCandidates Store // Say says one of the indexer's events (placement-changed); nil says none. Say func(event string, body any) error // Drained says whether the indexer has applied every message of the forge stream (the consumer has none pending and none @@ -117,6 +120,9 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { if err := ix.applyPlacement(ctx, f); err != nil { return err } + if err := ix.applyPlacementCandidates(ctx, f); err != nil { + return err + } var digests map[string]string var held map[string]delivery.Module if f.Kind == delivery.ForgePull { @@ -171,6 +177,14 @@ func (ix *Indexer) applyOpen(ctx context.Context, f delivery.Forge) error { for _, p := range f.Open { named[p] = true } + if ix.PlacementCandidates != nil { + // A pull request the forge no longer has open keeps no placement candidate (issue 1126). + if err := ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool { + return !named[delivery.OpenPull{Repository: c.Pull.Repository, Number: c.Pull.Number}] + }, "its pull request is not open"); err != nil { + return err + } + } keys, err := ix.Store.Keys(ctx) if err != nil { return fmt.Errorf("listing the candidates: %w", err) diff --git a/modules/mesh-delivery-indexer/internal/index/placement_candidates.go b/modules/mesh-delivery-indexer/internal/index/placement_candidates.go new file mode 100644 index 0000000..3df2c88 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/placement_candidates.go @@ -0,0 +1,180 @@ +package index + +import ( + "context" + "errors" + "fmt" + "maps" + "slices" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// Placement candidates (issue 1126; novox/hq ADR 0549 §4: a placement change passes the composition check before any +// node sees it). An updated pull request of the catalogue that touches placement/ carries every placement file at its +// head (the forge bridge, PlacementsRead). For each node whose file at the head differs from the trunk's key +// `placement.` (SamePlacement, the revision left out), or whose file the head removes, the indexer writes the key +// `.` of its state `placement-candidates`; a node whose file is the trunk's has none, and a key the head no +// longer needs is deleted. The key is written only when its value changes, so its write is the node's event to compose +// in shadow and answer. When the pull request closes or merges, or an open message no longer names it, its keys are +// deleted. A pull message that read no placement (a forge resend, a pull request touching none) leaves them as they are. +// +// A file at the head that does not read as a placement, or names another node, is said in the log and that node gets no +// key (one it had is deleted): the catalogue's own check refuses the pull request (ADR 0549 §1), and a node answers no +// candidate that is not one. + +// applyPlacementCandidates takes one forge message into the placement candidates. +func (ix *Indexer) applyPlacementCandidates(ctx context.Context, f delivery.Forge) error { + if ix.PlacementCandidates == nil || !delivery.SameRepository(f.Repository, delivery.CatalogueRepository) || trunkLike[f.Branch] { + return nil + } + if f.Kind == delivery.ForgeMerged || f.Action == delivery.PullClosed { + return ix.dropPlacementCandidates(ctx, func(_ string, c delivery.PlacementCandidate) bool { + return c.Group == f.Branch && delivery.SameRepository(c.Pull.Repository, f.Repository) && c.Pull.Number == f.Number + }, fmt.Sprintf("%s#%d %s", f.Repository, f.Number, map[bool]string{true: "merged", false: "closed"}[f.Kind == delivery.ForgeMerged])) + } + if f.Kind != delivery.ForgePull || !f.PlacementsRead { + return nil + } + trunk, err := ix.trunkPlacements(ctx) + if err != nil { + ix.logf("%s#%d at %s: its placement candidates are not written: %v", f.Repository, f.Number, f.Head, err) + return nil + } + nodes := map[string]bool{} + for node := range f.Placements { + nodes[node] = true + } + for node := range trunk { + nodes[node] = true + } + pull := delivery.Head{Repository: f.Repository, Number: f.Number, Head: f.Head} + for _, node := range slices.Sorted(maps.Keys(nodes)) { + key, err := delivery.PlacementCandidateKey(f.Branch, node) + if err != nil { + return fmt.Errorf("%w: %s#%d: %v", ErrUnreadable, f.Repository, f.Number, err) + } + var next *delivery.PlacementCandidate // nil: no key for this node + raw, atHead := f.Placements[node] + before, inTrunk := trunk[node] + switch { + case atHead: + p, err := delivery.DecodePlacement([]byte(raw)) + if err == nil && p.Node != node { + err = fmt.Errorf("the file of %s names the node %q", node, p.Node) + } + if err != nil { + ix.logf("%s#%d at %s: the placement file of %s is refused, so %s has no placement candidate (the catalogue's check refuses the pull request): %v", + f.Repository, f.Number, f.Head, node, node, err) + break + } + if inTrunk && delivery.SamePlacement(before, p) { + break + } + next = &delivery.PlacementCandidate{Group: f.Branch, Pull: pull, Placement: &p} + case inTrunk: + next = &delivery.PlacementCandidate{Group: f.Branch, Pull: pull} // the head removes the node's file + } + if err := ix.writePlacementCandidate(ctx, key, next); err != nil { + return err + } + } + // A node the head and the trunk both no longer name keeps no key of this group. + return ix.dropPlacementCandidates(ctx, func(node string, c delivery.PlacementCandidate) bool { + return c.Group == f.Branch && delivery.SameRepository(c.Pull.Repository, f.Repository) && c.Pull.Number == f.Number && !nodes[node] + }, fmt.Sprintf("%s#%d at %s", f.Repository, f.Number, f.Head)) +} + +// trunkPlacements is every node's placement in the trunk's state placement, by node. +func (ix *Indexer) trunkPlacements(ctx context.Context) (map[string]delivery.Placement, error) { + out := map[string]delivery.Placement{} + if ix.Placement == nil { + return nil, errors.New("the indexer reads no placement state") + } + keys, err := ix.Placement.Keys(ctx) + if err != nil { + return nil, fmt.Errorf("reading the placement keys: %w", err) + } + for _, key := range keys { + node, err := delivery.ParsePlacementKey(key) + if err != nil { + continue + } + value, _, found, err := ix.Placement.Get(ctx, key) + if err != nil { + return nil, fmt.Errorf("reading %s: %w", key, err) + } + if !found { + continue + } + p, err := delivery.DecodePlacement(value) + if err != nil { + return nil, fmt.Errorf("%s holds what is not a placement: %w", key, err) + } + out[node] = p + } + return out, nil +} + +// writePlacementCandidate writes one candidate key, or deletes it for next nil, only when it changes. +func (ix *Indexer) writePlacementCandidate(ctx context.Context, key string, next *delivery.PlacementCandidate) error { + var data []byte + if next != nil { + var err error + if data, err = delivery.EncodePlacementCandidate(*next); err != nil { + return fmt.Errorf("%w: %s.%s: %v", ErrUnreadable, delivery.PlacementCandidatesState, key, err) + } + } + err := write(ctx, ix.PlacementCandidates, key, func(value []byte, found bool) ([]byte, bool, error) { + switch { + case next == nil: + return nil, found, nil + case found && string(value) == string(data): + return nil, false, nil + } + return data, false, nil + }) + if errors.Is(err, process.ErrNotYet) { + ix.logf("%s.%s is not written: %v", delivery.PlacementCandidatesState, key, err) + return nil + } + return err +} + +// dropPlacementCandidates deletes every candidate key whose node and value gone says, and says each with why. +func (ix *Indexer) dropPlacementCandidates(ctx context.Context, gone func(node string, c delivery.PlacementCandidate) bool, why string) error { + keys, err := ix.PlacementCandidates.Keys(ctx) + if errors.Is(err, process.ErrNotYet) { + return nil + } + if err != nil { + return fmt.Errorf("reading the placement candidate keys: %w", err) + } + for _, key := range keys { + _, node, err := delivery.ParsePlacementCandidateKey(key) + if err != nil { + continue + } + dropped := false + err = write(ctx, ix.PlacementCandidates, key, func(value []byte, found bool) ([]byte, bool, error) { + if !found { + return nil, false, nil + } + c, err := delivery.DecodePlacementCandidate(value) + if err != nil { + ix.logf("%s.%s holds what is not a placement candidate, deleted: %v", delivery.PlacementCandidatesState, key, err) + return nil, true, nil + } + dropped = gone(node, c) + return nil, dropped, nil + }) + if err != nil { + return err + } + if dropped { + ix.logf("%s.%s is deleted: %s", delivery.PlacementCandidatesState, key, why) + } + } + return nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/placement_test.go b/modules/mesh-delivery-indexer/internal/index/placement_test.go index adb48f1..9e24c79 100644 --- a/modules/mesh-delivery-indexer/internal/index/placement_test.go +++ b/modules/mesh-delivery-indexer/internal/index/placement_test.go @@ -3,6 +3,7 @@ package index import ( "context" "reflect" + "slices" "strings" "testing" @@ -94,4 +95,77 @@ func placementIsWrittenAtAMergeOnlyWhereItChanged(t *testing.T) { if p, _ := written("ace"); p.Revision != "m1" || len(said) != 0 { t.Errorf("a message that read no placement changed it: %+v, %v", p, said) } + + // Issue 1126 (ADR 0549 §4), placement candidates: the trunk now places ace alone. A catalogue pull request writes + // `.` for each node whose file at its head differs from the trunk's, or that the head removes, and only + // when the value changes; a pull message that read no placement, or a pull request changing none, writes nothing; a + // file that does not read gets no key; a merge, a close or an open message not naming it deletes its keys. + cands := newMem() + ix.PlacementCandidates = cands + said = nil + pullOf := func(branch string, number int, head string, files map[string]string) delivery.Forge { + return delivery.Forge{V: delivery.Version, Kind: delivery.ForgePull, Repository: delivery.CatalogueRepository, Branch: branch, + Number: number, Head: head, Action: delivery.PullUpdated, Placements: files, PlacementsRead: files != nil} + } + candidate := func(key string) (delivery.PlacementCandidate, uint64, bool) { + t.Helper() + v, ok := cands.vals[key] + if !ok { + return delivery.PlacementCandidate{}, 0, false + } + c, err := delivery.DecodePlacementCandidate(v) + if err != nil { + t.Fatal(err) + } + return c, cands.revs[key], true + } + keys := func() string { k, _ := cands.Keys(ctx); slices.Sort(k); return strings.Join(k, ",") } + g14New := `{"v":1,"node":"g14","modules":[{"module":"zsh"}]}` + for _, row := range []struct { + name string + f delivery.Forge + keys string + }{ + {"a pull request adding g14", pullOf("feat/p", 9, h1, map[string]string{"ace": ace, "g14": g14New}), "feat/p.g14"}, + {"the same head again", pullOf("feat/p", 9, h1, map[string]string{"ace": ace, "g14": g14New}), "feat/p.g14"}, + {"a pull message that read no placement", pullOf("feat/p", 9, h2, nil), "feat/p.g14"}, + {"a pull request changing no placement", pullOf("feat/q", 10, h1, map[string]string{"ace": ace}), "feat/p.g14"}, + {"a new head removing ace", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"}, + {"a head whose g14 file does not read", pullOf("feat/p", 9, h3, map[string]string{"g14": `{"v":1,"node":"ace","modules":[]}`}), "feat/p.ace"}, + {"the head adding g14 again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"}, + {"an open message not naming it", delivery.Forge{V: delivery.Version, Kind: delivery.ForgeOpen, Open: []delivery.OpenPull{{Repository: delivery.CatalogueRepository, Number: 10}}}, ""}, + {"the pull request heard again", pullOf("feat/p", 9, h2, map[string]string{"g14": g14New}), "feat/p.ace,feat/p.g14"}, + {"its merge", func() delivery.Forge { + m := merged("m7", map[string]string{"g14": g14New}) + m.Branch, m.Number = "feat/p", 9 + return m + }(), ""}, + } { + _, revBefore, _ := candidate("feat/p.g14") + if err := ix.Apply(ctx, row.f); err != nil { + t.Fatalf("%s: %v", row.name, err) + } + if got := keys(); got != row.keys { + t.Errorf("%s: keys %q, want %q", row.name, got, row.keys) + } + c, rev, ok := candidate("feat/p.g14") + switch row.name { + case "a pull request adding g14": + if !ok || c.Group != "feat/p" || c.Pull != (delivery.Head{Repository: delivery.CatalogueRepository, Number: 9, Head: h1}) || c.Placement == nil || + !reflect.DeepEqual(c.Placement.Assigned(), []string{"zsh"}) { + t.Errorf("%s: %+v", row.name, c) + } + case "the same head again", "a pull message that read no placement", "a pull request changing no placement": + if rev != revBefore { + t.Errorf("%s: feat/p.g14 written again (revision %d, was %d)", row.name, rev, revBefore) + } + case "a new head removing ace": + if a, _, _ := candidate("feat/p.ace"); a.Placement != nil || a.Pull.Head != h2 || c.Pull.Head != h2 { + t.Errorf("%s: ace %+v, g14 at %s", row.name, a, c.Pull.Head) + } + } + } + if p, _ := written("g14"); len(p.Modules) != 1 || p.Revision != "m7" { + t.Errorf("the merge's trunk placement of g14: %+v", p) + } } diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 0122ca7..e184318 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -85,7 +85,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build 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: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { + if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "placement-candidates", History: 5}, {Name: "taken", History: 1}, {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) { diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 04bca8c..4882fb3 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -19,6 +19,7 @@ { "name": "manifests", "history": 1 }, { "name": "modules", "history": 20 }, { "name": "placement", "history": 10 }, + { "name": "placement-candidates", "history": 5 }, { "name": "taken", "history": 1 }, { "name": "takeins", "history": 1 } ],