Write a catalogue pull request's changed placement as placement-candidates <group>.<node> and delete them when it merges or closes, so each placed node composes it in shadow and answers (issue 1126)

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.

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
 )

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=

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)

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.<node>` (SamePlacement, the revision left out), or whose file the head removes, the indexer writes the key
+// `<group>.<node>` 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
+}

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
+	// `<group>.<node>` 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)
+	}
 }

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) {

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 }
   ],