diff --git a/go/buscompose/buscompose_test.go b/go/buscompose/buscompose_test.go index fbb033f..7825671 100644 --- a/go/buscompose/buscompose_test.go +++ b/go/buscompose/buscompose_test.go @@ -3,11 +3,6 @@ package buscompose import ( "encoding/json" "errors" - "time" - - "git.novox.be/novox/mesh-sdk/go/announce" - "git.novox.be/novox/mesh-sdk/go/delivery" - "github.com/nats-io/nats.go/jetstream" "os" "path/filepath" "reflect" @@ -15,8 +10,11 @@ import ( "slices" "strings" "testing" + "time" + "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/delivery" + "github.com/nats-io/nats.go/jetstream" ) // Defends issue 1377 and novox/hq ADR 0547 §5: this is the controller's bus composer copied, and it composes byte for @@ -439,7 +437,7 @@ func TestTheCopiedComposerComposesAsTheControllers(t *testing.T) { } } announced := map[string]announce.States{ControllerSeat: ControllerAnnouncement("novox")} - for reader, want := range map[string]bool{VaultSeat: true, GrantsModule: true, AssignerModule: true, "mesh-delivery-view": false} { + for reader, want := range map[string]bool{VaultSeat: true, GrantsModule: true, "mesh-delivery-assigner": true, "mesh-delivery-view": false} { granted, _ := ReadsAnnouncedBy(reader, []string{ControllerLease}, announced) if (len(granted) == 1) != want { t.Errorf("%s reads the lease: %v, want %v", reader, granted, want) diff --git a/go/buscompose/userlist_switch.go b/go/buscompose/userlist_switch.go index 5829ff2..6e2cb07 100644 --- a/go/buscompose/userlist_switch.go +++ b/go/buscompose/userlist_switch.go @@ -79,13 +79,10 @@ const VaultBusAccounts = "mesh-vault.bus-accounts" // NamedReaders are the states only the modules named may read, by `.`: the grants part's wanted accounts // are read by the vault alone (novox/hq ADR 0561 §2). -// The controller's lease is read by the two parts that turn on at its write (ADR 0561 §3, §6) and by the assigner, which -// reads the grants from the grants part from then on (ADR 0547 §4, §5; delivery.GrantSource). +// The controller's lease is read by the parts that turn on at its write (ADR 0561 §3, §6), and by the assigner, which +// reads the catalogue's placement alone from then (ADR 0549 §5, issue 774). var NamedReaders = map[string][]string{VaultBusAccounts: {GrantsModule}, GrantsAccounts: {VaultSeat}, - ControllerLease: {VaultSeat, GrantsModule, AssignerModule}} - -// AssignerModule is the assigner, which reads the lease to choose the source of the grants. -const AssignerModule = "mesh-delivery-assigner" + ControllerLease: {VaultSeat, GrantsModule, delivery.AssignerModule}} // ReaderMay says whether module may read the state read (`.`) by the named readers: always, for a state // that names none. diff --git a/go/delivery/buildrun/standing.go b/go/delivery/buildrun/standing.go new file mode 100644 index 0000000..f230d0f --- /dev/null +++ b/go/delivery/buildrun/standing.go @@ -0,0 +1,225 @@ +package buildrun + +import ( + "context" + "errors" + "fmt" + "maps" + "slices" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// A build module writes . of its toolchains state on every node it runs on, and nothing deletes the key +// when the module leaves a node: no part writes another module's state, and the module is gone. A report of a node the +// build module no longer runs on is therefore no builder's (issue 1940: mesh-build-node taken off ace on 2026-10-23 left +// ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree"). +// +// ToolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a +// ToolchainSet): an entry is passed on only while the catalogue's placement (the indexer's state `placement`, read by +// delivery.Placed; ADR 0549 §3) places its build module on its node, and only once it was written since that placement; +// it is passed on as deleted otherwise, said once, and passed on when its node writes it again. A report written before +// the placement is one an earlier run of the module left (issue 1940, 2026-10-25 20:56: mesh-build-node put on ace again +// brought ace.typescript v26.12.0 of 2026-10-23 back into the agreement before the module ran there). Until the +// placement is settled, and while no node's placement is held (before the first placement file reaches the indexer, or +// where the state cannot be read), every entry is passed on. Builders that run and report different versions still +// disagree: that refusal is the guard of novox/hq ADR 0465, and it stays. +// +// This replaces the copies of the gate the indexer, the builder and the checker each kept over the controller's +// assignments (issue 1940's subtask, issue 774: those go with the controller's lease, ADR 0549 §6). +type ToolchainGate struct { + mu sync.Mutex + modules map[string]bool // the build modules whose toolchains pass the gate + forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds + logf func(format string, a ...any) + placed *delivery.Placed + known bool // the placement is settled + raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it + left map[string]string // / -> why it was said left out, so each reason is said once +} + +// NewToolchainGate is a gate over the toolchains of modules, passing what stands on to forward and saying with logf what +// it leaves out. +func NewToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *ToolchainGate { + g := &ToolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, placed: delivery.NewPlaced(), + raw: map[string]map[string][]byte{}, left: map[string]string{}} + for _, m := range modules { + g.modules[m] = true + } + return g +} + +// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed. +func (g *ToolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool { + g.mu.Lock() + defer g.mu.Unlock() + if g.raw[module] == nil { + g.raw[module] = map[string][]byte{} + } + if deleted { + delete(g.raw[module], key) + } else { + g.raw[module][key] = value + } + return g.pass(module, key) +} + +// Placement takes one revision of the indexer's placement in (at is when the bus wrote it) and answers whether what it +// passed on changed. A value that does not read is said, and places nothing on its node. +func (g *ToolchainGate) Placement(key string, value []byte, deleted bool, at time.Time) bool { + g.mu.Lock() + defer g.mu.Unlock() + was := g.placed.Seeded() + node, changed, err := g.placed.Put(key, value, deleted, at) + if err != nil { + g.logf("%s: %v (issue 1940)", delivery.PlacementBucket, err) + } + if !g.known { + return false + } + if was != g.placed.Seeded() { + // The first placement came, or the last went: every report's standing changed. + return g.all() + } + return g.again(node, changed) +} + +// Settled says the placement's history is read: from now on only the reports of placed nodes pass, once a node's +// placement is held. It answers whether what it passed on changed. +func (g *ToolchainGate) Settled() bool { + g.mu.Lock() + defer g.mu.Unlock() + g.placed.Settle() + g.known = true + return g.all() +} + +// all passes on every report again. The caller holds mu. +func (g *ToolchainGate) all() bool { + changed := false + for _, module := range slices.Sorted(maps.Keys(g.raw)) { + for _, k := range slices.Sorted(maps.Keys(g.raw[module])) { + changed = g.pass(module, k) || changed + } + } + return changed +} + +// again passes on every report of node by a build module among modules. The caller holds mu. +func (g *ToolchainGate) again(node string, modules []string) bool { + changed := false + for _, module := range modules { + if !g.modules[module] { + continue + } + for _, k := range slices.Sorted(maps.Keys(g.raw[module])) { + if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node { + changed = g.pass(module, k) || changed + } + } + } + return changed +} + +// pass hands one key on as it stands: its value when its node is placed and wrote it since (or the placement is not +// known), deleted otherwise. The caller holds mu. +func (g *ToolchainGate) pass(module, key string) bool { + value, held := g.raw[module][key] + if !held { + return g.forward(module, key, nil, true) + } + node, _, err := delivery.ParseToolchainKey(key) + why := "" + if err == nil && g.known && g.placed.Seeded() { + why = g.leftOut(module, node, value) + } + if why == "" { + delete(g.left, module+"/"+key) + return g.forward(module, key, value, false) + } + if g.left[module+"/"+key] != why { + g.left[module+"/"+key] = why + g.logf("%s.%s is left out of the agreed toolchain: %s (issue 1940)", delivery.ToolchainsBucket(module), key, why) + } + return g.forward(module, key, nil, true) +} + +// leftOut is why a node's report is no builder's, or empty when it counts: the catalogue does not place the module on its +// node, or the report was written before it did. A value that is no toolchain is passed on for its reader to refuse. +// The caller holds mu. +func (g *ToolchainGate) leftOut(module, node string, value []byte) string { + since, placed := g.placed.Since(node, module) + if !placed { + return fmt.Sprintf("the catalogue does not place %s on %s, so it is no builder", module, node) + } + tc, err := delivery.DecodeToolchain(value) + if err == nil && tc.At.Before(since) { + return fmt.Sprintf("it was written at %s, before %s was placed on %s at %s: it counts once %s writes it again", + tc.At.UTC().Format(time.RFC3339), module, node, since.UTC().Format(time.RFC3339), node) + } + return "" +} + +// FollowPlacement reads the indexer's placement with its history into gate, says it settled, and follows it; changed +// is told whenever what the gate passes on changed. Where the state cannot be read or watched (not there, or this part +// may not read it) it says so and the gate passes every report on: not fatal, since an indexer, builder or checker that +// stops for it stops every delivery (issue 1945). An error of the watch after that ends on the channel. +func FollowPlacement(ctx context.Context, js process.BucketOpener, gate *ToolchainGate, changed func(context.Context), logf func(string, ...any)) (<-chan error, error) { + errs := make(chan error, 1) + kv, err := js.KeyValue(ctx, delivery.PlacementBucket) + if err != nil { + logf("the state %s cannot be read, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", delivery.PlacementBucket, err) + return errs, nil + } + w, err := kv.WatchAll(ctx, jetstream.IncludeHistory()) + if err != nil { + logf("the state %s cannot be watched, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", delivery.PlacementBucket, err) + return errs, nil + } + put := func(e jetstream.KeyValueEntry) bool { + return gate.Placement(e.Key(), e.Value(), e.Operation() != jetstream.KeyValuePut, e.Created()) + } + for settled := false; !settled; { + select { + case <-ctx.Done(): + _ = w.Stop() + return errs, ctx.Err() + case e, ok := <-w.Updates(): + if !ok { + return errs, errors.New("the watch of " + delivery.PlacementBucket + " ended before it settled") + } + if e == nil { + settled = true + continue + } + put(e) + } + } + if gate.Settled() { + changed(ctx) + } + go func() { + defer func() { _ = w.Stop() }() + for { + select { + case <-ctx.Done(): + errs <- ctx.Err() + return + case e, ok := <-w.Updates(): + if !ok { + errs <- errors.New("the watch of " + delivery.PlacementBucket + " ended") + return + } + if e != nil && put(e) { + changed(ctx) + } + } + } + }() + return errs, nil +} diff --git a/go/delivery/buildrun/standing_test.go b/go/delivery/buildrun/standing_test.go new file mode 100644 index 0000000..6377d36 --- /dev/null +++ b/go/delivery/buildrun/standing_test.go @@ -0,0 +1,235 @@ +package buildrun + +import ( + "bytes" + "context" + "errors" + "fmt" + "log" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func gateReport(t *testing.T, node, version string, at time.Time) (string, []byte) { + t.Helper() + key, err := delivery.ToolchainKey(node, "typescript") + if err != nil { + t.Fatal(err) + } + value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: at}) + if err != nil { + t.Fatal(err) + } + return key, value +} + +// placing is a value of the state `placement` placing modules on node. +func placing(t *testing.T, node string, modules ...string) (string, []byte) { + t.Helper() + p := delivery.Placement{Node: node} + for _, m := range modules { + p.Modules = append(p.Modules, delivery.PlacedModule{Module: m}) + } + key, err := delivery.PlacementKey(node) + if err != nil { + t.Fatal(err) + } + value, err := delivery.EncodePlacement(p) + if err != nil { + t.Fatal(err) + } + return key, value +} + +func agreedOf(set *ToolchainSet) (string, string) { + tc, none := delivery.AgreedToolchain(set.Entries("typescript")) + return tc.Version, none +} + +// Issue 1940, the gate over the catalogue's placement (issue 774, ADR 0549 §3): a build node's report counts only while +// the placement places its build module on its node, and only once written since; before the placement is settled, and +// while no node's placement is held, every report counts. The live cases: mesh-build-node taken off ace left +// ace.typescript at v26.12.0 against v26.12.1 on the three others; put on ace again, its report of 2026-10-23 06:29 came +// back before the module ran there. +func TestToolchainGate(t *testing.T) { + old := time.Date(2026, 10, 23, 6, 29, 18, 0, time.UTC) // ace's report from its earlier run + placed := time.Date(2026, 10, 25, 19, 56, 45, 0, time.UTC) // mesh-build-node placed on ace again + three := []string{"g14", "novox", "shanks"} + start := func(t *testing.T, seeded bool) (*ToolchainGate, *ToolchainSet, *[]string) { + set := &ToolchainSet{} + said := &[]string{} + gate := NewToolchainGate([]string{GoBuildModule, NodeBuildModule}, set.Put, func(f string, a ...any) { *said = append(*said, fmt.Sprintf(f, a...)) }) + for _, node := range three { + if seeded { + key, value := placing(t, node, NodeBuildModule) + gate.Placement(key, value, false, old.Add(-time.Hour)) + } + key, value := gateReport(t, node, "v26.12.1", placed.Add(time.Minute)) + gate.Toolchain(NodeBuildModule, key, value, false) + } + return gate, set, said + } + + t.Run("not placed on ace: its report is no builder's", func(t *testing.T) { + gate, set, said := start(t, true) + key, value := gateReport(t, "ace", "v26.12.0", old) + gate.Toolchain(NodeBuildModule, key, value, false) + if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") { + t.Fatalf("before the placement is settled every report counts; got %q", none) + } + k, v := placing(t, "ace", GoBuildModule) // another build module on ace: no matter + gate.Placement(k, v, false, old) + if !gate.Settled() { + t.Fatal("settling left ace's report out, so what was passed on changed") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("the placed nodes agree on v26.12.1; got %q, %q", v, none) + } + if len(*said) != 1 || !strings.Contains((*said)[0], "does not place mesh-build-node on ace") { + t.Fatalf("ace's report is said left out once; said %q", *said) + } + if gate.Toolchain(NodeBuildModule, key, value, false) { + t.Fatal("a report of a node not placed changed what was passed on") + } + // Placed on ace with a report written since: it counts, and the builders really disagree. + k, v = placing(t, "ace", GoBuildModule, NodeBuildModule) + gate.Placement(k, v, false, old.Add(-time.Minute)) + if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") { + t.Fatalf("builders that run different versions still disagree; got %q", none) + } + // Taken off ace again (its key deleted): its report goes again. + if !gate.Placement(k, nil, true, placed) { + t.Fatal("taking ace's placement away passed nothing on") + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("got %q, %q", v, none) + } + // A report deleted from the bucket goes whether its node is placed or not. + key, _ = gateReport(t, "g14", "v26.12.1", placed) + if !gate.Toolchain(NodeBuildModule, key, nil, true) || len(set.Entries("typescript")) != 2 { + t.Fatalf("deleting g14's report: %d entries left, want novox and shanks", len(set.Entries("typescript"))) + } + }) + + for _, c := range []struct { + name string + at time.Time // when ace wrote its report + counts bool + }{ + {"a report left before the placement", old, false}, + {"a report written since the placement", placed.Add(30 * time.Second), true}, + } { + t.Run(c.name, func(t *testing.T) { + gate, set, said := start(t, true) + gate.Settled() + k, v := placing(t, "ace", NodeBuildModule) + if gate.Placement(k, v, false, placed) { + t.Fatal("a placement with no report passed something on") + } + key, value := gateReport(t, "ace", "v26.12.0", c.at) + if got := gate.Toolchain(NodeBuildModule, key, value, false); got != c.counts { + t.Fatalf("ace's report passed on: %v, want %v", got, c.counts) + } + // ace's placement written again with another module beside: its standing does not move, nothing is said again. + k, v = placing(t, "ace", GoBuildModule, NodeBuildModule) + if gate.Placement(k, v, false, placed.Add(time.Minute)) { + t.Fatal("the same placement of mesh-build-node changed what was passed on") + } + if c.counts { + return + } + if v, none := agreedOf(set); v != "v26.12.1" || none != "" { + t.Fatalf("ace's left report is no builder's; got %q, %q", v, none) + } + if len(*said) != 1 || !strings.Contains((*said)[0], "before mesh-build-node was placed on ace at 2026-10-25T19:56:45Z") { + t.Fatalf("said %q, want once why ace's report is left out", *said) + } + key, value = gateReport(t, "ace", "v26.12.1", placed.Add(time.Minute)) + if !gate.Toolchain(NodeBuildModule, key, value, false) || len(set.Entries("typescript")) != 4 { + t.Fatal("ace's new report was not passed on") + } + }) + } + + t.Run("no placement held: every report counts", func(t *testing.T) { + gate, set, _ := start(t, false) + key, value := gateReport(t, "ace", "v26.12.0", old) + gate.Toolchain(NodeBuildModule, key, value, false) + gate.Settled() + if n := len(set.Entries("typescript")); n != 4 { + t.Fatalf("before the first placement file every report counts; got %d", n) + } + k, v := placing(t, "g14", NodeBuildModule) + if !gate.Placement(k, v, false, old.Add(-time.Hour)) || len(set.Entries("typescript")) != 1 { + t.Fatalf("the first placement leaves every node it does not place out; got %d", len(set.Entries("typescript"))) + } + if !gate.Placement(k, nil, true, placed) || len(set.Entries("typescript")) != 4 { + t.Fatal("the last placement gone: every report counts again") + } + }) + + t.Run("the history: a full history's oldest revision stands since before it", func(t *testing.T) { + p := delivery.NewPlaced() + k, v := placing(t, "ace", NodeBuildModule) + for i := range delivery.PlacementHistory { + p.Put(k, v, false, old.Add(time.Duration(i)*time.Minute)) + } + k2, v2 := placing(t, "g14", NodeBuildModule) + p.Put(k2, v2, false, old) + if _, _, err := p.Put("placement.novox", []byte("not json"), false, old); err == nil { + t.Error("a value that does not read was not said") + } + p.Settle() + if s, ok := p.Since("ace", NodeBuildModule); !ok || !s.IsZero() { + t.Errorf("ace's module stands since %v, %v; want the zero time", s, ok) + } + if s, ok := p.Since("g14", NodeBuildModule); !ok || !s.Equal(old) { + t.Errorf("g14's module stands since %v, %v; want its first revision", s, ok) + } + if got := p.PlacedNodes(NodeBuildModule); strings.Join(got, ",") != "ace,g14" || len(p.ByNode()) != 2 || !p.Seeded() { + t.Errorf("PlacedNodes = %v, ByNode = %v", got, p.ByNode()) + } + }) + + t.Run("a placement the bus refuses is said, and the part goes on", func(t *testing.T) { + var logs bytes.Buffer + logger := log.New(&logs, "", 0) + refused := errors.New("nats: permissions violation for publish to $JS.API.CONSUMER.CREATE.KV_mesh-delivery-indexer_placement") + gate, _, _ := start(t, false) + for _, bus := range []refusingBus{{err: refused}, {refuseOpen: true, err: refused}} { + if _, err := FollowPlacement(context.Background(), bus, gate, func(context.Context) {}, logger.Printf); err != nil { + t.Errorf("a refused placement stopped the start: %v", err) + } + } + if !strings.Contains(logs.String(), "cannot be watched") || !strings.Contains(logs.String(), "cannot be read") { + t.Errorf("the refusals were not said: %q", logs.String()) + } + }) +} + +// refusingBus is a bus that refuses: every open (refuseOpen), or every watch of a bucket it opens, with err. +type refusingBus struct { + refuseOpen bool + err error +} + +func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { + if r.refuseOpen { + return nil, r.err + } + return refusedWatch{err: r.err}, nil +} + +type refusedWatch struct { + jetstream.KeyValue + err error +} + +func (r refusedWatch) WatchAll(context.Context, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { + return nil, r.err +} diff --git a/go/delivery/buildrun/toolchains_test.go b/go/delivery/buildrun/toolchains_test.go index e434b8a..4df0d38 100644 --- a/go/delivery/buildrun/toolchains_test.go +++ b/go/delivery/buildrun/toolchains_test.go @@ -35,10 +35,8 @@ func TestToolchainSetKeepsEachNodesEntryByLanguageAndSaysWhenOneChanged(t *testi if !s.Put(GoBuildModule, "ace.go", nil, true) || len(s.Entries("go")) != 1 { t.Error("a delete did not remove the node's entry") } -} - -func TestToolchainSetLeavesOutWhatIsNotAToolchainOfTheModulesLanguage(t *testing.T) { // 1607/AC6 - var s ToolchainSet + // What is not a toolchain of the module's language is left out. + s = ToolchainSet{} if s.Put(GoBuildModule, "novox.typescript", tcValue(t, "typescript", "x"), false) { t.Error("a typescript toolchain in the go build module's bucket counted") } diff --git a/go/delivery/placed.go b/go/delivery/placed.go new file mode 100644 index 0000000..e73937a --- /dev/null +++ b/go/delivery/placed.go @@ -0,0 +1,150 @@ +package delivery + +import ( + "fmt" + "maps" + "slices" + "sync" + "time" +) + +// Placed is where the indexer's placement (the state `placement`, ADR 0549 §2, §3) places each module, and since when: +// the one reader of that state for every part that asks "does module stand on node" (issue 774, design 47's third +// milestone), in place of each part reading the controller's assignments (`mesh-controller_assignments`, which goes +// with the controller's lease, ADR 0549 §6). +// +// Its input is the state's revisions in the order the bus wrote them, as a watch with the history (jetstream's +// IncludeHistory) hands them: Put takes each, and Settle ends the history. A module stands since the revision of its +// node's key that first named it in an unbroken run; a module in the oldest revision of a key whose history is full +// (PlacementHistory) stands since before what the bus still keeps (the zero time), since an earlier revision may have +// named it. A key's value that does not read places nothing on its node, and Put says why. +// +// Safe for use from a watch and from a reader at once. +type Placed struct { + mu sync.Mutex + since map[string]map[string]time.Time // node -> module -> since when it stands + seen map[string]int // node -> revisions read before Settle + first map[string]time.Time // node -> when its first revision read was written + settled bool +} + +// NewPlaced is a Placed that holds no placement yet. +func NewPlaced() *Placed { + return &Placed{since: map[string]map[string]time.Time{}, seen: map[string]int{}, first: map[string]time.Time{}} +} + +// Put takes one revision of a key of the state `placement` (its value, whether it is a delete, and when the bus wrote +// it). It answers the node the key is of and the modules whose standing on it changed (placed or removed), in name +// order, and an error for a key or a value that does not read (the node then places nothing). +func (p *Placed) Put(key string, value []byte, deleted bool, at time.Time) (node string, changed []string, err error) { + node, err = ParsePlacementKey(key) + if err != nil { + return "", nil, err + } + now := map[string]bool{} + if !deleted { + pl, derr := DecodePlacement(value) + switch { + case derr != nil: + err = fmt.Errorf("the placement of %s does not read, so it places nothing there: %w", node, derr) + case pl.Node != node: + err = fmt.Errorf("the placement under %s names the node %q, so it places nothing there", key, pl.Node) + default: + for _, m := range pl.Assigned() { + now[m] = true + } + } + } + p.mu.Lock() + defer p.mu.Unlock() + if !p.settled { + if p.seen[node] == 0 { + p.first[node] = at + } + p.seen[node]++ + } + was := p.since[node] + next := map[string]time.Time{} + for m := range now { + if s, ok := was[m]; ok { + next[m] = s + } else { + next[m] = at + changed = append(changed, m) + } + } + for m := range was { + if !now[m] { + changed = append(changed, m) + } + } + if len(next) == 0 { + delete(p.since, node) + } else { + p.since[node] = next + } + slices.Sort(changed) + return node, changed, err +} + +// Settle ends the history: from now on every Put is a new revision. A module in the oldest kept revision of a key whose +// history is full stands since the zero time. +func (p *Placed) Settle() { + p.mu.Lock() + defer p.mu.Unlock() + if p.settled { + return + } + p.settled = true + for node, n := range p.seen { + if n < PlacementHistory { + continue + } + for m, s := range p.since[node] { + if s.Equal(p.first[node]) { + p.since[node][m] = time.Time{} + } + } + } +} + +// Seeded says some node's placement is held: before the first placement file reaches the indexer nothing is placed, +// and a reader keeps what it did before (ADR 0549 §5). +func (p *Placed) Seeded() bool { + p.mu.Lock() + defer p.mu.Unlock() + return len(p.since) > 0 +} + +// Since says whether module stands on node, and since when (the zero time: since before the revisions kept). +func (p *Placed) Since(node, module string) (time.Time, bool) { + p.mu.Lock() + defer p.mu.Unlock() + s, ok := p.since[node][module] + return s, ok +} + +// PlacedNodes is every node module is placed on, in name order. +func (p *Placed) PlacedNodes(module string) []string { + p.mu.Lock() + defer p.mu.Unlock() + var out []string + for node, mods := range p.since { + if _, ok := mods[module]; ok { + out = append(out, node) + } + } + slices.Sort(out) + return out +} + +// ByNode is, by node, the modules placed on it in name order. +func (p *Placed) ByNode() map[string][]string { + p.mu.Lock() + defer p.mu.Unlock() + out := make(map[string][]string, len(p.since)) + for node, mods := range p.since { + out[node] = slices.Sorted(maps.Keys(mods)) + } + return out +}