diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 7c232b3..737cee7 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -88,7 +88,7 @@ const goBuildModule = "mesh-build-go" // (ADR 0567 §1), named after the assigner as every module state is. var ownBuckets = []string{delivery.AssignerAsksBucket, delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerPutsBucket, delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket, - delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket, assign.PoliciesBucket} + delivery.PlacementCandidatesBucket, delivery.PlacementPulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, delivery.SeatHoldersBucket} // seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants // them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind). @@ -112,6 +112,7 @@ var keptStates = []announce.State{ {Name: "wants", Readable: true}, {Name: health.FindingsState, Readable: true}, {Name: delivery.AssignerPoliciesState, Readable: true}, + {Name: delivery.SeatHoldersState, Readable: true}, } // buckets is every bucket the assigner opened, by name. @@ -160,6 +161,12 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err) continue } + if name == delivery.SeatHoldersBucket && err != nil { + // The seat's state, new with ADR 0571 §4: bound once the bus says it was made; until then nobody is told who + // holds which seat here, and the controller's seats still answer. + logf("the bucket %s cannot be opened (%v): it is bound once the bus says it was made", name, err) + continue + } if name == keptBucket && err != nil { // mesh-config's held values (ADR 0565 §5): without them a composition lays none, said in the journal. logf("the bucket %s cannot be opened (%v): no held value is laid on a composition until it is bound", name, err) @@ -329,6 +336,7 @@ var watched = []string{ assign.PoliciesBucket, provisions.ProvisionsBucket, keptBucket, + delivery.SeatHoldersBucket, } // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after @@ -425,7 +433,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } started := 0 for _, name := range watched { - if kvs[name] == nil && (slices.Contains(seatBuckets, name) || sourceOfGrants(name) || name == provider.AssignmentsBucket || name == assign.PoliciesBucket || name == provisions.ProvisionsBucket || name == keptBucket) { + if kvs[name] == nil && (slices.Contains(seatBuckets, name) || sourceOfGrants(name) || name == provider.AssignmentsBucket || name == assign.PoliciesBucket || name == provisions.ProvisionsBucket || name == keptBucket || name == delivery.SeatHoldersBucket) { continue // not on the bus at this start (open said so) } started++ @@ -455,7 +463,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, } } // And the two that choose where grants are read from, with the same delivery as the grants part that writes one. - for _, name := range []string{delivery.PartGrantsBucket, buscompose.LeaseBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, keptBucket} { + for _, name := range []string{delivery.PartGrantsBucket, buscompose.LeaseBucket, assign.PoliciesBucket, provisions.ProvisionsBucket, keptBucket, delivery.SeatHoldersBucket} { if kvs[name] == nil && js != nil { name := name process.AwaitBucket(ctx, js, nc, name, "until then the bucket is not read", diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 3ba3270..93a28ff 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "errors" "maps" "slices" @@ -145,6 +146,7 @@ func (p *passer) wants(ctx context.Context) { missing := map[string]bool{} defer func() { p.provisionsMissing(ctx, missing) }() defer p.unplacedFindings() + defer p.seatHolders(ctx) defer p.policySplits(manifests, chosen) frozen := p.sayPlacement(p.withCalls(p.w.placementDiffers())) placed := p.w.placedNodes() @@ -1186,3 +1188,46 @@ func (p *passer) unplacedFindings() { } } } + +// seatHolders writes the seat mesh-placement's state seat-holders from every node's placement (ADR 0571 §4): one key per +// seat held once, ``, and one per node of a node seat, `.`, written only where it differs, and a key no +// placement holds any more deleted. A pass follows each placement change and each manifest the indexer takes in, so no +// timer writes it. +func (p *passer) seatHolders(ctx context.Context) { + kv := p.kvs[delivery.SeatHoldersBucket] + if kv == nil { + return + } + placements := map[string]delivery.Placement{} + for key, value := range p.w.read(delivery.PlacementBucket) { + node, err := delivery.ParsePlacementKey(key) + if err != nil { + continue + } + if pl, err := delivery.DecodePlacement(value); err == nil && pl.Node == node { + placements[node] = pl + } + } + if len(placements) == 0 { + return // no placement read yet: nothing is known of who holds what, and nothing is deleted + } + want := delivery.SeatHoldersOf(placements) + have := p.w.read(delivery.SeatHoldersBucket) + for key, sh := range want { + value, err := json.Marshal(sh) + if err != nil || string(have[key]) == string(value) { + continue + } + p.put(ctx, delivery.SeatHoldersBucket, key, func() ([]byte, error) { return value, nil }) + } + for key := range have { + if _, ok := want[key]; ok { + continue + } + if err := kv.Delete(ctx, key); err != nil { + p.logf("the seat holders %s are not deleted: %v", key, err) + continue + } + p.w.put(delivery.SeatHoldersBucket, key, nil) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go index 3c683f7..46f1530 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools.go @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sort" "strings" "time" @@ -375,6 +376,9 @@ type historyEntry struct { // doPlacement is the verb placement (§10, §6): one node's key, one module's nodes, or every key; with history, the // changes in order from the revisions the state keeps. func (s *seatTools) doPlacement(args map[string]any) (any, error) { + if arg(args, "seats") == "true" { + return s.seats(arg(args, "node")) + } all, revs, err := s.placements() if err != nil { return nil, err @@ -538,3 +542,32 @@ func (s *seatTools) tools() []stdio.Tool { } return out } + +// seats is the verb placement with seats (ADR 0571 §4): who holds each seat and on which node, as the state seat-holders +// holds it; with node, the keys of seats held on that node. +func (s *seatTools) seats(node string) (any, error) { + keys, err := s.st.keys(delivery.SeatHoldersState) + if err != nil { + return nil, fmt.Errorf("seat-holders cannot be listed: %w", err) + } + sort.Strings(keys) + out := []delivery.SeatHolders{} + for _, key := range keys { + raw, _, found, err := s.st.get(delivery.SeatHoldersState, key) + if err != nil { + return nil, fmt.Errorf("seat-holders %s cannot be read: %w", key, err) + } + if !found { + continue + } + var sh delivery.SeatHolders + if json.Unmarshal(raw, &sh) != nil { + continue + } + if node != "" && sh.Node != node && !slices.ContainsFunc(sh.Holders, func(h delivery.SeatHolder) bool { return h.Node == node }) { + continue + } + out = append(out, sh) + } + return map[string]any{"seats": out, "holds": len(keys)}, nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index ea1b3ef..87c0017 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -201,6 +201,40 @@ func TestTheSeatsVerbs(t *testing.T) { t.Errorf("placed again: findings %v", findings) } }) + t.Run("1955/L7 (ADR 0571 §4): seat-holders is written from placement, only the keys a change moves, and answered by placement seats", func(t *testing.T) { + writes := 0 + kv := &countingKV{memKV: memKV{values: map[string][]byte{}}, writes: &writes} + p := &passer{w: newWorld(), logf: t.Logf, kvs: buckets{delivery.SeatHoldersBucket: kv}} + place := func(node string, holds ...delivery.Holding) { + pl, err := delivery.EncodePlacement(delivery.Placement{V: delivery.Version, Node: node, Modules: []delivery.PlacedModule{{Module: "web"}, {Module: "nats"}}, Holds: holds}) + if err != nil { + t.Fatal(err) + } + k, _ := delivery.PlacementKey(node) + p.w.putEntry(delivery.PlacementBucket, k, pl, 1, time.Now()) + } + place("ace", delivery.Holding{Claim: "mesh-broker", Scope: "mesh", Module: "nats"}, delivery.Holding{Claim: "node-shell", Scope: "node", Module: "web"}) + place("g14", delivery.Holding{Claim: "node-shell", Scope: "node", Module: "web"}) + ctx := context.Background() + p.seatHolders(ctx) + if writes != 3 || kv.values["mesh-broker"] == nil || kv.values["node-shell.ace"] == nil || kv.values["node-shell.g14"] == nil { + t.Fatalf("%d writes: %v", writes, kv.values) + } + p.seatHolders(ctx) + if writes != 3 { + t.Errorf("an unchanged placement wrote %d keys", writes-3) + } + place("g14") // g14 holds nothing now + p.seatHolders(ctx) + if writes != 3 || kv.values["node-shell.g14"] != nil || kv.values["node-shell.ace"] == nil { + t.Errorf("after g14's change: %d writes, %v", writes, kv.values) + } + st := memStore{delivery.SeatHoldersState: &memKV{values: kv.values}} + out, err := newSeatTools(st, nil).seats("ace") + if err != nil || len(out.(map[string]any)["seats"].([]delivery.SeatHolders)) != 2 { + t.Errorf("seats on ace = %+v, %v", out, err) + } + }) t.Run("1954/D1: a failed provision is the assigner's finding until it is provided", func(t *testing.T) { findings := map[string][]byte{} p := &passer{w: newWorld(), logf: func(string, ...any) {}, kvs: buckets{provisions.ProvisionsBucket: nil}, findings: assign.Findings{ @@ -367,3 +401,14 @@ func readManifest() (catalogue.Manifest, error) { } return catalogue.ParseManifest(raw) } + +// countingKV is a memKV that counts its puts. +type countingKV struct { + memKV + writes *int +} + +func (c *countingKV) Put(ctx context.Context, key string, value []byte) (uint64, error) { + *c.writes++ + return c.memKV.Put(ctx, key, value) +} diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index a9b4fbe..0744ac9 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-assigner go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213929-66ff33dbd35d + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 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-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum index 326dffd..ecec070 100644 --- a/modules/mesh-delivery-assigner/go.sum +++ b/modules/mesh-delivery-assigner/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213929-66ff33dbd35d h1:J1QAvzlaIvCCyZGBQX0XGJjYWdNPjLXBF4zaAeIkUOg= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029213929-66ff33dbd35d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 h1:OsA1v6n736a7RqEV8Nkip8pf5ILMbzj0PWuKqcBOhK0= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121/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-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 84bee99..b939b72 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -44,7 +44,7 @@ func TestManifest(t *testing.T) { states[s.Name] = s.History } if want := map[string]int{"asks": 5, "compositions": 5, "differences": 64, "puts": 20, "wants": 20, - "placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20}; !reflect.DeepEqual(states, want) { + "placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index ea1b4b7..9dae8b6 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -82,7 +82,8 @@ { "name": "puts", "history": 20 }, { "name": "wants", "history": 20 }, { "name": "findings", "history": 5 }, - { "name": "policies", "history": 20 } + { "name": "policies", "history": 20 }, + { "name": "seat-holders", "history": 5 } ], "streams": [ {