Assigner: write who holds which seat from placement into seat-holders, only the keys a change moves, and answer it from placement's seats, so readers no longer ask the controller's seats (issue 1955, ADR 0571 §4)

Pins mesh-sdk f9a496d.
This commit is contained in:
2026-10-29 23:14:28 +01:00
parent b44f01f89c
commit c045057ca0
8 changed files with 140 additions and 8 deletions
@@ -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",
@@ -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, `<seat>`, and one per node of a node seat, `<seat>.<node>`, 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)
}
}
@@ -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
}
@@ -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)
}
+1 -1
View File
@@ -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
)
+2 -2
View File
@@ -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=
@@ -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 {
+2 -1
View File
@@ -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": [
{