The assigner reads the seat list, so a proposal no longer refuses a replicated seat (issue 1966, recovery) #183
@@ -29,6 +29,7 @@ import (
|
||||
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
@@ -415,6 +416,12 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
// The assigner's one pull of its durable on the shared events stream: the indexer's views and the engines' reads changed
|
||||
// are handlers of it, started once both are registered (process.EventHub).
|
||||
hub := process.NewEventHub(logger.Printf)
|
||||
// The seat list (seatlist.go), read over the connection and heard on the hub, given to catalogue at every change, and a
|
||||
// pass made then: registered before the hub starts.
|
||||
gate := &seatGate{}
|
||||
if bus != nil {
|
||||
gate.follow(seatlist.ProcessBus(bus.Call, hub), wake, logger.Printf)
|
||||
}
|
||||
// The indexer's manifests, modules and module graph, through its first views and views family, never its buckets
|
||||
// (DLV-F2r, ADR 0631): the first pass waits for the three answers, as it waits for every watch to hand what stands.
|
||||
if iv != nil {
|
||||
@@ -465,12 +472,12 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
|
||||
}
|
||||
}
|
||||
p := &passer{kvs: kvs, w: w, logf: logger.Printf, say: say, findings: findings, said: map[string]bool{}, answeredAt: map[string]uint64{},
|
||||
reads: reads}
|
||||
reads: reads, seats: gate}
|
||||
// The node-engine seat's manifest-reads-changed, for one node (ADR 0618 §4): that node is asked again, with no timer.
|
||||
// Heard on the assigner's durable consumer on EVENTS, which the grants part makes from the manifest's consumes; where it
|
||||
// is not there yet, a node is asked again only when it comes back online.
|
||||
if js != nil {
|
||||
hearReadsChanged(ctx, js, hub, reads, findings, logger.Printf)
|
||||
hearReadsChanged(ctx, js, hub, reads, findings, gate.eventsLive, logger.Printf)
|
||||
}
|
||||
if bus != nil && kvs[buckets.Puts] != nil {
|
||||
p.acts = newPutActs(bus, wake, logger.Printf)
|
||||
@@ -540,7 +547,7 @@ func sayVia(ctx context.Context, js jetstream.JetStream) func(oc.Condition) erro
|
||||
// hearReadsChanged hears the node-engine seat's event manifest-reads-changed on the assigner's one hub of its durable
|
||||
// consumer on EVENTS, and asks the node it names again; the hub also hands the indexer's views to the followers of
|
||||
// indexerViews, and starts once both are registered.
|
||||
func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, logf func(string, ...any)) {
|
||||
func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.EventHub, reads *enginereads.Keeper, findings assign.Findings, live func(), logf func(string, ...any)) {
|
||||
account := process.Account{Node: thisNode, Module: assignment.Module}
|
||||
w := notHeardWriter(findings, logf)
|
||||
handle := func(m jetstream.Msg) {
|
||||
@@ -566,6 +573,9 @@ func hearReadsChanged(ctx context.Context, js jetstream.JetStream, hub *process.
|
||||
return
|
||||
}
|
||||
w.Heard(process.EventsStream)
|
||||
if live != nil {
|
||||
live() // the seat list, unread while the seat was silent, is read now (seatGate.eventsLive)
|
||||
}
|
||||
go func() {
|
||||
<-ctx.Done()
|
||||
hub.Stop()
|
||||
|
||||
@@ -71,7 +71,10 @@ type passer struct {
|
||||
callsCleared bool
|
||||
// The seat mesh-module-assigner's holder (seat.go, ADR 0567): its acts (nil: none is made), how it says
|
||||
// the seat's events (nil: none is said), the proposals it said, and what Intended computed by proposal and base.
|
||||
seat *seatActs
|
||||
seat *seatActs
|
||||
// seats is the seat list the process reads (seatlist.go): until it is read, seat-holders is not written and a proposal
|
||||
// laid again is not ended on a seat it does not name (ADR 0636 §1). Nil reads nothing.
|
||||
seats *seatGate
|
||||
publish publisher
|
||||
saidProposed map[string]bool
|
||||
// copied says the retired state was copied into `assignments` by this process, or found empty (copyRetired).
|
||||
@@ -1119,6 +1122,14 @@ func (p *passer) seatHolders(ctx context.Context) {
|
||||
if len(assignments) == 0 {
|
||||
return // no assignments read yet: nothing is known of who holds what, and nothing is deleted
|
||||
}
|
||||
if !p.seats.read() {
|
||||
// ADR 0636 §1: which claim names which seat is the list's, so nothing is written from the bootstrap alone; the
|
||||
// reader's change makes a pass.
|
||||
if p.seats != nil {
|
||||
p.seats.waitSaid.Do(func() { p.logf("seat-holders is not written until the seat list is read: %s", p.seats.health()) })
|
||||
}
|
||||
return
|
||||
}
|
||||
want := plc.SeatHoldersOf(assignments)
|
||||
have := p.w.read(plc.SeatHoldersBucket)
|
||||
for key, sh := range want {
|
||||
|
||||
@@ -440,7 +440,12 @@ func (p *passer) mustAnswer(ctx context.Context, prop mma.Proposal, rev uint64,
|
||||
func (p *passer) layAgain(ctx context.Context, prop mma.Proposal, rev uint64, current map[string]mma.Assignments, revs map[string]uint64, why string) {
|
||||
held, recorded := p.lays(ctx)
|
||||
again, laid, findings := assign.Propose(assign.ProposeIn{ID: prop.ID, By: prop.By, Why: prop.Why, At: prop.At, Changes: prop.Changes,
|
||||
Current: current, Revisions: revs, Held: held, Recorded: recorded})
|
||||
Current: current, Revisions: revs, Held: held, Recorded: recorded, SeatListRead: p.seats.read()})
|
||||
if slices.ContainsFunc(findings, plc.WaitsForSeatList) {
|
||||
// ADR 0636 §1: nothing is ended on a list not read; the reader's change makes a pass, which lays it again.
|
||||
p.logf("the proposal %s is not laid again yet: %s", prop.ID, p.seats.health())
|
||||
return
|
||||
}
|
||||
if len(findings) > 0 {
|
||||
p.endProposal(ctx, prop, why+", and laid over the keys now it is refused: "+strings.Join(findings, "; "))
|
||||
return
|
||||
|
||||
@@ -84,6 +84,9 @@ type seatTools struct {
|
||||
st seatStore
|
||||
ask func(key string, body any) (json.RawMessage, error)
|
||||
now func() time.Time
|
||||
// seats is the seat list the bundle reads (seatlist.go): until it is read, propose judges no mesh seat the bootstrapped
|
||||
// definitions do not name as held once (ADR 0636 §1). Nil reads nothing.
|
||||
seatList *seatGate
|
||||
}
|
||||
|
||||
func newSeatTools(st seatStore, ask func(string, any) (json.RawMessage, error)) *seatTools {
|
||||
@@ -262,7 +265,7 @@ func (s *seatTools) doPropose(args map[string]any, by string) (any, error) {
|
||||
enrollers := plc.EnrollersOf(current)
|
||||
now := s.now()
|
||||
p, laid, findings := assign.Propose(assign.ProposeIn{ID: plc.NewProposalID(now), By: by, Why: arg(args, "why"), At: now,
|
||||
Changes: changes, Current: current, Revisions: revs, Held: held, Recorded: recorded, Warrant: warrant})
|
||||
Changes: changes, Current: current, Revisions: revs, Held: held, Recorded: recorded, Warrant: warrant, SeatListRead: s.seatList.read()})
|
||||
if refusal := plc.AddNodeRefusal(p, moduleOfCaller(by, enrollers), enrollers); refusal != "" {
|
||||
findings = append(findings, refusal)
|
||||
}
|
||||
|
||||
@@ -225,6 +225,33 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
})
|
||||
}
|
||||
indexer.refused = ""
|
||||
// Issue 1966: the replicated mesh-dns-resolver held on two nodes is proposed through the verb as the bundle serves it,
|
||||
// over the seat list it reads (ADR 0636 §1): unread, the check says it waits for the list and writes nothing; read, the
|
||||
// proposal is laid.
|
||||
for _, c := range []struct {
|
||||
name string
|
||||
silent bool
|
||||
want string
|
||||
}{
|
||||
{"a replicated seat on two nodes waits for an unread seat list", true, "the seat list is not read yet, so whether the seat mesh-dns-resolver may be held on anchor and laptop waits for it"},
|
||||
{"a replicated seat on two nodes, the seat list read", false, ""},
|
||||
} {
|
||||
t.Run("1966: "+c.name, func(t *testing.T) {
|
||||
st[mma.StateProposals].values, st[mma.StateCandidates].values = map[string][]byte{}, map[string][]byte{}
|
||||
s.seatList = gateOver(t, c.silent)
|
||||
_, err := s.doPropose(map[string]any{"why": "a second resolver", "changes": []any{
|
||||
map[string]any{"op": "hold", "node": "anchor", "module": "web", "claim": "mesh-dns-resolver", "scope": "mesh"},
|
||||
map[string]any{"op": "place", "node": "laptop", "module": "web"},
|
||||
map[string]any{"op": "hold", "node": "laptop", "module": "web", "claim": "mesh-dns-resolver", "scope": "mesh"}}}, "op")
|
||||
switch {
|
||||
case c.want == "" && err != nil:
|
||||
t.Errorf("propose: %v", err)
|
||||
case c.want != "" && (err == nil || !strings.Contains(err.Error(), c.want) || strings.Contains(err.Error(), "held once") || len(st[mma.StateProposals].values) != 0):
|
||||
t.Errorf("propose: %v, want a refusal saying %q, never held once, and nothing written", err, c.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
s.seatList = nil
|
||||
t.Run("1953/D4-D6: the seat mesh-wants' upgrade and bus", wantsVerbs)
|
||||
t.Run("1955/L7 (ADR 0571 §6): rule 7 reads the node's own composition; an unplaced node is said", func(t *testing.T) {
|
||||
w := newWorld()
|
||||
@@ -258,14 +285,24 @@ func TestTheSeatsVerbs(t *testing.T) {
|
||||
t.Run("1955/L7 (ADR 0571 §4): seat-holders is written from assignments, only the keys a change moves, and answered by assignments 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: kvSet{plc.SeatHoldersBucket: kv}}
|
||||
// Issue 1966: nothing is written while the seat list is unread (ADR 0636 §1), and every seat the list names is no
|
||||
// seat outside it once read.
|
||||
p := &passer{w: newWorld(), logf: t.Logf, kvs: kvSet{plc.SeatHoldersBucket: kv}, seats: gateOver(t, true)}
|
||||
place := func(node string, holds ...mma.Holding) {
|
||||
putAssignments(p.w, mma.Assignments{V: keys.Version, Node: node, Modules: []mma.PlacedModule{{Module: "web"}, {Module: "nats"}}, Holds: holds})
|
||||
}
|
||||
place("ace", mma.Holding{Claim: "mesh-broker", Scope: "mesh", Module: "nats"}, mma.Holding{Claim: "node-shell", Scope: "node", Module: "web"})
|
||||
place("g14", mma.Holding{Claim: "node-shell", Scope: "node", Module: "web"})
|
||||
ctx := context.Background()
|
||||
if p.seatHolders(ctx); writes != 0 {
|
||||
t.Fatalf("seat-holders was written from an unread seat list: %v", kv.values)
|
||||
}
|
||||
p.seats = gateOver(t, false)
|
||||
p.seatHolders(ctx)
|
||||
var broker mma.SeatHolders
|
||||
if err := json.Unmarshal(kv.values["mesh-broker"], &broker); err != nil || broker.Outside {
|
||||
t.Errorf("mesh-broker's holders %+v, %v: a seat the list names is no seat outside it", broker, err)
|
||||
}
|
||||
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)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,101 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
)
|
||||
|
||||
// The assigner reads the seat list as every part does (novox/hq ADR 0635 §1, ADR 0636 §1, §7; issue 1966): through the seat
|
||||
// mesh-catalog's verb seat-definitions and its two events, from the bootstrapped definitions until the seat has answered. What
|
||||
// the reader holds is given to catalogue at every change, which the check of a proposal and the seat-holders state read: which
|
||||
// mesh seat is replicated, and which name is a seat's. Both of its runtimes hold one: the process (the proposals it lays again,
|
||||
// the state seat-holders) over its connection and the event hub, and the tools' bundle (the seat's verb propose) over the
|
||||
// runtime. Until the list is read, a mesh seat the bootstrapped definitions do not name is never judged held once
|
||||
// (plc.CheckAssignments), and seat-holders is not written.
|
||||
|
||||
// Seat list tools (ADR 0636 §7): what the reader holds, and reading it again now, which every operator's session and the
|
||||
// healers may ask.
|
||||
const (
|
||||
seatListTool = "assigner_seat_list"
|
||||
seatListAgainTool = "assigner_seat_list_again"
|
||||
)
|
||||
|
||||
var seatListAgainCallers = []string{stdio.AnyPerson, "mesh-healers"}
|
||||
|
||||
// seatGate is the reader a runtime of the assigner judges by, set once it is started; before, the list counts as unread.
|
||||
type seatGate struct {
|
||||
reader atomic.Pointer[seatlist.Reader]
|
||||
// waitSaid says the journal was told that seat-holders waits for the list, so it is told once per wait.
|
||||
waitSaid sync.Once
|
||||
}
|
||||
|
||||
// read says the seat mesh-catalog has answered the reader: the list is read (seatlist.Status's Heard). A nil gate, or one
|
||||
// whose reader has not started, has read nothing.
|
||||
func (g *seatGate) read() bool {
|
||||
if g == nil {
|
||||
return false
|
||||
}
|
||||
r := g.reader.Load()
|
||||
return r != nil && r.Status().Heard
|
||||
}
|
||||
|
||||
// health is why the list is not read, or empty.
|
||||
func (g *seatGate) health() string {
|
||||
if g == nil {
|
||||
return "the assigner follows no seat list"
|
||||
}
|
||||
r := g.reader.Load()
|
||||
if r == nil {
|
||||
return "the seat list reader has not started yet"
|
||||
}
|
||||
return r.Status().Health()
|
||||
}
|
||||
|
||||
// follow starts the reader over bus: what it holds is given to catalogue at once and at every change, and changed is called
|
||||
// then (the process makes a pass, which lays again what waited).
|
||||
func (g *seatGate) follow(bus seatlist.Bus, changed func(), logf func(string, ...any)) *seatlist.Reader {
|
||||
reader, err := seatlist.Start(seatlist.Config{Bus: bus, Bootstrap: seatlist.Bootstrapped(),
|
||||
OnChange: func(seats []catalogue.Seat) {
|
||||
catalogue.UseSeats(seats)
|
||||
if changed != nil {
|
||||
changed()
|
||||
}
|
||||
}})
|
||||
if err != nil {
|
||||
logf("the assigner cannot follow its seat list, so a mesh seat the bootstrapped definitions do not name waits for it: %v", err)
|
||||
}
|
||||
if reader != nil {
|
||||
catalogue.UseSeats(reader.Seats())
|
||||
g.reader.Store(reader)
|
||||
}
|
||||
return reader
|
||||
}
|
||||
|
||||
// eventsLive is called once the process's consumer on EVENTS is heard: the seat may have said it speaks before that consumer
|
||||
// existed, whose events start at the stream's end, so while the list is unread it is read now (ADR 0636 §1).
|
||||
func (g *seatGate) eventsLive() {
|
||||
if r := g.reader.Load(); r != nil {
|
||||
r.ReloadIfUnread()
|
||||
}
|
||||
}
|
||||
|
||||
// tools are the reader's two tools over whatever reader the gate holds at the call.
|
||||
func (g *seatGate) tools() []stdio.Tool {
|
||||
shape := seatlist.Tools(&seatlist.Reader{}, seatListTool, seatListAgainTool, seatListAgainCallers)
|
||||
for i := range shape {
|
||||
run := i
|
||||
shape[i].Run = func(args map[string]any) (any, error) {
|
||||
r := g.reader.Load()
|
||||
if r == nil {
|
||||
return nil, errors.New("the seat list reader has not started yet: the runtime has not served this module")
|
||||
}
|
||||
return seatlist.Tools(r, seatListTool, seatListAgainTool, seatListAgainCallers)[run].Run(args)
|
||||
}
|
||||
}
|
||||
return shape
|
||||
}
|
||||
@@ -0,0 +1,63 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist/seatlisttest"
|
||||
)
|
||||
|
||||
// testDefinitions are seatlisttest's, parsed once.
|
||||
var testDefinitions = sync.OnceValue(func() []seatdef.Definition { return seatlisttest.Definitions() })
|
||||
|
||||
// standInCatalog is the seat mesh-catalog as a test's reader asks it (ADR 0636 §4): every definition of seatlisttest, or,
|
||||
// silent, no answer at all.
|
||||
type standInCatalog struct{ silent bool }
|
||||
|
||||
func (c standInCatalog) Ask(key string, body any) (json.RawMessage, error) {
|
||||
if c.silent || key != seatlist.Key() {
|
||||
return nil, errors.New("no responders")
|
||||
}
|
||||
defs := testDefinitions()
|
||||
seat, _ := body.(map[string]any)["seat"].(string)
|
||||
if seat == "" {
|
||||
l := seatlist.Listing{Revision: 1}
|
||||
for _, d := range defs {
|
||||
l.Seats = append(l.Seats, seatlist.Entry{Seat: d.Seat, Source: seatlist.SourceTrunk, Digest: "d-" + d.Seat})
|
||||
}
|
||||
return json.Marshal(l)
|
||||
}
|
||||
for _, d := range defs {
|
||||
if d.Seat == seat {
|
||||
raw, err := json.Marshal(d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.Marshal(seatlist.Answer{Seat: seat, Found: true, Revision: 1,
|
||||
Entry: &seatlist.Entry{Seat: seat, Source: seatlist.SourceTrunk, Digest: "d-" + seat, Definition: raw}})
|
||||
}
|
||||
}
|
||||
return json.Marshal(seatlist.Answer{Seat: seat, Revision: 1})
|
||||
}
|
||||
|
||||
func (standInCatalog) Subscribe(string, func(string, json.RawMessage) error) error { return nil }
|
||||
|
||||
// gateOver is a seat gate following the stand-in, once its first read has ended. A read one gives catalogue the test list
|
||||
// (catalogue.UseSeats), as a host's does, for the tests after it too: the list is the same everywhere.
|
||||
func gateOver(t *testing.T, silent bool) *seatGate {
|
||||
t.Helper()
|
||||
g := &seatGate{}
|
||||
r := g.follow(standInCatalog{silent: silent}, nil, t.Logf)
|
||||
if r == nil {
|
||||
t.Fatal("the reader did not start")
|
||||
}
|
||||
<-r.Ready()
|
||||
if g.read() == silent {
|
||||
t.Fatalf("the gate read %v over a stand-in silent %v: %s", g.read(), silent, g.health())
|
||||
}
|
||||
return g
|
||||
}
|
||||
@@ -17,6 +17,7 @@ import (
|
||||
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
|
||||
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads"
|
||||
)
|
||||
@@ -363,7 +364,16 @@ func serveTools() error {
|
||||
time.Sleep(wait)
|
||||
}
|
||||
}()
|
||||
tools := append(s.tools(), newSeatTools(runtimeStore{}, stdio.Ask).tools()...)
|
||||
// The seat list (seatlist.go), read through the runtime once Serve has its channel: the seat's verbs judge by it.
|
||||
gate := &seatGate{}
|
||||
go func() {
|
||||
<-stdio.Served()
|
||||
gate.follow(seatlist.StdioBus(), nil, func(f string, a ...any) { fmt.Fprintf(os.Stderr, "mesh-delivery-assigner: "+f+"\n", a...) })
|
||||
}()
|
||||
seatVerbs := newSeatTools(runtimeStore{}, stdio.Ask)
|
||||
seatVerbs.seatList = gate
|
||||
tools := append(s.tools(), seatVerbs.tools()...)
|
||||
tools = append(tools, gate.tools()...)
|
||||
wants := newWantsTools(runtimeStore{}, stdio.Ask, func(e, id string, b any) error {
|
||||
// The seat's event, said on the seat: a claimed seat's event is never the module's own.
|
||||
return stdio.SeatPublish(wantsEventSubject(e), b, id)
|
||||
|
||||
@@ -36,6 +36,9 @@ type ProposeIn struct {
|
||||
Recorded func(node string) bool
|
||||
// Warrant is the enrolment's warrant an add-node carries (ADR 0587 §7); only that change has one.
|
||||
Warrant *mma.Warrant
|
||||
// SeatListRead says the host has read the seat list from the seat mesh-catalog (ADR 0636 §1); until then a mesh seat
|
||||
// the bootstrapped definitions do not name is not judged held once (plc.CheckAssignments).
|
||||
SeatListRead bool
|
||||
}
|
||||
|
||||
// Propose lays a proposal over the current keys (§2): the proposal (state answering, each changed node's base revision),
|
||||
@@ -110,7 +113,7 @@ func Propose(in ProposeIn) (mma.Proposal, map[string]*mma.Assignments, []string)
|
||||
was := recorded
|
||||
recorded = func(node string) bool { return joining[node] || was(node) }
|
||||
}
|
||||
for _, r := range plc.CheckAssignments(all, held, recorded) {
|
||||
for _, r := range plc.CheckAssignments(all, held, recorded, in.SeatListRead) {
|
||||
findings = append(findings, r)
|
||||
}
|
||||
return p, out, findings
|
||||
|
||||
@@ -67,17 +67,31 @@ func TestAssignments(t *testing.T) {
|
||||
t.Run("the check of every assignments", func(t *testing.T) {
|
||||
known := func(m string) bool { return m != "ghost" }
|
||||
enrolled := func(n string) bool { return n != "stranger" }
|
||||
if r := CheckAssignments(current, known, enrolled); len(r) != 0 {
|
||||
if r := CheckAssignments(current, known, enrolled, true); len(r) != 0 {
|
||||
t.Fatalf("a sound assignments is refused: %v", r)
|
||||
}
|
||||
// A replicated mesh seat (ADR 0223) is held on two nodes; a seat that is not replicated is refused there.
|
||||
for name, refused := range map[string]bool{"mesh-dns-resolver": false, "mesh-broker": true} {
|
||||
a := seat.Assignments{Node: "ace", Modules: []seat.PlacedModule{{Module: "dns"}}, Holds: []seat.Holding{{Claim: name, Scope: "mesh", Module: "dns"}}}
|
||||
// A replicated mesh seat (ADR 0223) is held on two nodes; a seat that is not replicated is refused there. Issue 1966:
|
||||
// a seat the list does not name, while the list is unread, waits for it and is never said held once (ADR 0636 §1);
|
||||
// once read, it is judged held once.
|
||||
for _, c := range []struct {
|
||||
seat string
|
||||
listRead bool
|
||||
want string
|
||||
}{
|
||||
{"mesh-dns-resolver", true, ""},
|
||||
{"mesh-dns-resolver", false, ""},
|
||||
{"mesh-broker", true, "held once for the mesh, is held on ace and novox"},
|
||||
{"mesh-broker", false, "held once for the mesh, is held on ace and novox"},
|
||||
{"a-seat-not-listed", false, "the seat list is not read yet, so whether the seat a-seat-not-listed may be held on ace and novox waits for it"},
|
||||
{"a-seat-not-listed", true, "held once for the mesh, is held on ace and novox"},
|
||||
} {
|
||||
a := seat.Assignments{Node: "ace", Modules: []seat.PlacedModule{{Module: "dns"}}, Holds: []seat.Holding{{Claim: c.seat, Scope: "mesh", Module: "dns"}}}
|
||||
b := a
|
||||
b.Node = "novox"
|
||||
r := CheckAssignments(map[string]seat.Assignments{"ace": a, "novox": b}, known, enrolled)
|
||||
if (len(r) != 0) != refused {
|
||||
t.Errorf("%s on two nodes: refusals %v, want refused %v", name, r, refused)
|
||||
r := CheckAssignments(map[string]seat.Assignments{"ace": a, "novox": b}, known, enrolled, c.listRead)
|
||||
got := strings.Join(r, "; ")
|
||||
if (c.want == "") != (len(r) == 0) || !strings.Contains(got, c.want) || (!c.listRead && c.seat == "a-seat-not-listed" && strings.Contains(got, "held once")) {
|
||||
t.Errorf("%s on two nodes, the list read %v: refusals %v, want %q", c.seat, c.listRead, r, c.want)
|
||||
}
|
||||
}
|
||||
clash := g14
|
||||
@@ -89,7 +103,7 @@ func TestAssignments(t *testing.T) {
|
||||
twice := novox
|
||||
twice.Node = "stranger"
|
||||
twice.Decided.Address = "10.77.0.2"
|
||||
r := CheckAssignments(map[string]seat.Assignments{"g14": clash, "novox": novox, "stranger": twice}, known, enrolled)
|
||||
r := CheckAssignments(map[string]seat.Assignments{"g14": clash, "novox": novox, "stranger": twice}, known, enrolled, true)
|
||||
for _, want := range []string{"holds no module ghost", "ghost and app are both published on host port 18080",
|
||||
"bound to ace/postgres", "pin of shell names g14/bash", "stranger is a node with no record",
|
||||
"mesh-store, held once for the mesh, is held on novox and stranger",
|
||||
|
||||
@@ -33,7 +33,13 @@ var KeptApart = []struct{ A, B, Record string }{
|
||||
// (catalogue.Seat's Replicated, ADR 0223), which is held once per node; two nodes with one address on the private
|
||||
// network; more than one hub; two modules KeptApart on one node. A host port a manifest declares without a number and a
|
||||
// requirement left unbound are the composition's to find (its ForeseenPorts and Unbound).
|
||||
func CheckAssignments(assignments map[string]seat.Assignments, known func(module string) bool, enrolled func(node string) bool) []string {
|
||||
//
|
||||
// **A seat the list does not name is not judged held once while the list is unread** (novox/hq ADR 0636 §1, issue 1966):
|
||||
// listRead says the host has read the seat list from the seat mesh-catalog; before, catalogue holds the bootstrapped
|
||||
// definitions alone, and a mesh seat it does not name, held on two nodes, may be a replicated one. Its refusal then says the
|
||||
// judgement waits for the list (SeatListWaits), never that it is held twice; once the list is read, a seat it does not name
|
||||
// is judged as a seat held once, as before.
|
||||
func CheckAssignments(assignments map[string]seat.Assignments, known func(module string) bool, enrolled func(node string) bool, listRead bool) []string {
|
||||
var refusals []string
|
||||
nodes := sortedKeys(assignments)
|
||||
placed := func(p seat.Provider) bool {
|
||||
@@ -105,13 +111,18 @@ func CheckAssignments(assignments map[string]seat.Assignments, known func(module
|
||||
continue
|
||||
}
|
||||
name := h.Claim + "/" + h.Kind
|
||||
if s, known := catalogue.SeatNamed(h.Claim); known && s.Replicated {
|
||||
s, defined := catalogue.SeatNamed(h.Claim)
|
||||
if defined && s.Replicated {
|
||||
// A replicated mesh seat (ADR 0223) is held on several nodes at once, each answering the same thing;
|
||||
// it is still held once on each node.
|
||||
name += "/" + node
|
||||
}
|
||||
if other, twice := meshSeats[name]; twice {
|
||||
refusals = append(refusals, fmt.Sprintf("the seat %s, held once for the mesh, is held on %s and %s", h.Claim, other, node))
|
||||
if !defined && !listRead {
|
||||
refusals = append(refusals, fmt.Sprintf(SeatListWaits, h.Claim, other, node))
|
||||
} else {
|
||||
refusals = append(refusals, fmt.Sprintf("the seat %s, held once for the mesh, is held on %s and %s", h.Claim, other, node))
|
||||
}
|
||||
}
|
||||
meshSeats[name] = node
|
||||
}
|
||||
@@ -122,6 +133,16 @@ func CheckAssignments(assignments map[string]seat.Assignments, known func(module
|
||||
return refusals
|
||||
}
|
||||
|
||||
// SeatListWaits is the refusal of a mesh seat held on two nodes that the unread seat list does not name (ADR 0636 §1): the
|
||||
// seat, and the two nodes.
|
||||
const SeatListWaits = "the seat list is not read yet, so whether the seat %s may be held on %s and %s waits for it: " +
|
||||
"the assigner judges it once it has read the list from the seat mesh-catalog (novox/hq ADR 0636 §1)"
|
||||
|
||||
// WaitsForSeatList says a refusal is SeatListWaits': it ends nothing, and is judged again once the list is read.
|
||||
func WaitsForSeatList(refusal string) bool {
|
||||
return strings.HasPrefix(refusal, SeatListWaits[:strings.Index(SeatListWaits, "%")])
|
||||
}
|
||||
|
||||
// Change is what `assignments-changed` says of one node (ADR 0567 §6): the node, the revision of its key before (0: it had
|
||||
// none) and after (0: the key was deleted), and its modules added, removed or changed against the assignments before.
|
||||
type Change struct {
|
||||
|
||||
@@ -95,8 +95,8 @@ func TestManifest(t *testing.T) {
|
||||
// which lists and asks again what each node's engine reads.
|
||||
if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assignments", "seat-holders", "wants",
|
||||
"assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies",
|
||||
"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads"}) {
|
||||
t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools and assigner_engine_reads", m.Tools)
|
||||
"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads", "assigner_seat_list", "assigner_seat_list_again"}) {
|
||||
t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools, assigner_engine_reads and the seat list's two (ADR 0636 §7)", m.Tools)
|
||||
}
|
||||
if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs {
|
||||
t.Errorf("runs-as = %q: the bundle runs on the assigner's own account (ADR 0621 §1)", m.RunsAs)
|
||||
@@ -120,11 +120,12 @@ func TestManifest(t *testing.T) {
|
||||
t.Errorf("emits lists %s, an event the seat says", e)
|
||||
}
|
||||
}
|
||||
// ADR 0618 §4: and it hears each node's engine say its reads changed.
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", "node-engine.manifest-reads-changed"}) {
|
||||
// ADR 0618 §4: and it hears each node's engine say its reads changed; ADR 0636 §4 (issue 1966): and the seat list's two events.
|
||||
if !slices.Equal(m.Consumes, []string{"mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-delivery-indexer.views.graph", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules", "node-engine.manifest-reads-changed"}) {
|
||||
t.Errorf("consumes = %v, want the indexer's views of graph, manifests and modules and the engines' reads changed", m.Consumes)
|
||||
}
|
||||
if want := []string{"mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
|
||||
"seat:mesh-catalog.seat-definitions", // ADR 0636 §4: the seat list
|
||||
"seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone
|
||||
"seat:node-backup.backed-up", "seat:node-backup.now",
|
||||
"seat:node-engine.manifest-reads"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0618 §3
|
||||
|
||||
@@ -39,6 +39,8 @@
|
||||
"operator-channel"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-catalog.seat-definition-changed",
|
||||
"mesh-catalog.seat-definition-removed",
|
||||
"mesh-delivery-indexer.views.graph",
|
||||
"mesh-delivery-indexer.views.manifests",
|
||||
"mesh-delivery-indexer.views.modules",
|
||||
@@ -48,6 +50,7 @@
|
||||
"mesh-delivery-indexer.graph",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"seat:mesh-catalog.seat-definitions",
|
||||
"seat:mesh-hand-acts.hand-act",
|
||||
"seat:mesh-hand-acts.hand-acts",
|
||||
"seat:mesh-hand-acts.warranted",
|
||||
@@ -82,7 +85,9 @@
|
||||
"assigner_read_copies",
|
||||
"assigner_first_views",
|
||||
"assigner_first_view_again",
|
||||
"assigner_engine_reads"
|
||||
"assigner_engine_reads",
|
||||
"assigner_seat_list",
|
||||
"assigner_seat_list_again"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "assignments", "history": 64 },
|
||||
|
||||
Reference in New Issue
Block a user