diff --git a/cmd/mesh-host/seatlist.go b/cmd/mesh-host/seatlist.go index f9b8072..a480650 100644 --- a/cmd/mesh-host/seatlist.go +++ b/cmd/mesh-host/seatlist.go @@ -135,7 +135,8 @@ func (w findingsOfTheNode) Delete(f views.Finding) error { } // startSeatList starts this run's reader: its bootstrapped seats are given to catalogue at once, and every change of what -// it holds again, saying so to the composer and to the waiter that writes the five-minute finding. +// it holds again, saying so to the composer and to the waiter that writes the five-minute finding, and stating again what +// the engine reads (link.Session.RestateReads, ADR 0618 §4): the seats it states are those of the list it holds. func startSeatList(ctx context.Context, session *link.Session, node string, say func(string)) (*seatListState, error) { st := &seatListState{changed: make(chan struct{}, 1)} // Start returns at once and reads the seat on a goroutine, which may announce a change before the waiter exists. @@ -143,6 +144,9 @@ func startSeatList(ctx context.Context, session *link.Session, node string, say reader, err := seatlist.Start(seatlist.Config{Bus: sessionBus{ctx, session}, Bootstrap: seatlist.Bootstrapped(), OnChange: func(seats []catalogue.Seat) { catalogue.UseSeats(seats) + // What the engine reads changed with the list: its record and its event say it (ADR 0618 §4), off the + // reader's goroutine, which returns quickly. + go session.RestateReads(ctx) if f := announce.Load(); f != nil { // the first read may end before the waiter exists (*f)() } @@ -162,6 +166,7 @@ func startSeatList(ctx context.Context, session *link.Session, node string, say } announce.Store(&told) catalogue.UseSeats(reader.Seats()) + session.RestateReads(ctx) // its first statement holds the bootstrapped seats at least (issue 1966) told() // The first read ends whether the seat answered or not, and a read that held what the reader held already says no // change: its end is told to the waiter and the composer, so a status that moved from reading to heard starts or ends @@ -170,6 +175,7 @@ func startSeatList(ctx context.Context, session *link.Session, node string, say select { case <-reader.Ready(): catalogue.UseSeats(reader.Seats()) + session.RestateReads(ctx) told() case <-ctx.Done(): } diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 71c841b..225cd5b 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -38,6 +38,17 @@ type haveLink struct { // stated is what the tool runner's PING says of itself (Stater); statedErr is the runner not answering. stated map[string]string statedErr error + // readsSaid are the node seat's manifest-reads-changed events said on the link (ADR 0618 §4). + readsSaid []nodeengine.ManifestReadsChangedEvent +} + +func (l *haveLink) SayEngine(event string, body any) error { + if e, ok := body.(nodeengine.ManifestReadsChangedEvent); ok && event == nodeengine.EventManifestReadsChanged { + l.hmu.Lock() + l.readsSaid = append(l.readsSaid, e) + l.hmu.Unlock() + } + return nil } func (l *haveLink) Stated(context.Context, string, string) (map[string]string, error) { @@ -330,6 +341,37 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) { if got, _ = nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); got.Record.Contract != NodeKeyContract { t.Errorf("a report without a health statement dropped the contract: %+v", got.Record) } + + // Issue 1966 (ADR 0618 §4): the seats the engine states are those of the seat list it holds now. A change of the list + // writes the record once and says manifest-reads-changed with the new statement; a restatement that changes nothing + // writes and says nothing; the verb answers the statement and when it was made. + all := catalogue.Seats() + defer catalogue.UseSeats(all) + q.RestateReads(context.Background()) // the statement as the list stands, whatever an earlier test left + _, before := q.StatedReads() + at, saidBefore := l.kv["n1.node"].rev, len(l.readsSaid) + catalogue.UseSeats(slices.DeleteFunc(slices.Clone(all), func(s catalogue.Seat) bool { return s.Name == "mesh-vault" })) + q.RestateReads(context.Background()) + got, _ = nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value) + if l.kv["n1.node"].rev != at+1 || slices.Contains(got.Record.ManifestReads, "seat:mesh-vault") || + !slices.Contains(got.Record.ManifestReads, "seat:mesh-catalog") || !slices.Contains(got.Record.ManifestReads, "word:claims") { + t.Errorf("a seat list without mesh-vault: revision %d (was %d), the record states %v", l.kv["n1.node"].rev, at, got.Record.ManifestReads) + } + if said := l.readsSaid[saidBefore:]; len(said) != 1 || said[0].Why != ReadsStarted || !slices.Equal(said[0].ManifestReads, got.Record.ManifestReads) { + t.Errorf("a changed statement said %+v, want one manifest-reads-changed carrying the record's", said) + } + if reads, made := q.StatedReads(); !slices.Equal(reads, got.Record.ManifestReads) || made.Before(before) { + t.Errorf("the verb answers %v made at %v (was %v), want the record's", reads, made, before) + } + q.RestateReads(context.Background()) + if l.kv["n1.node"].rev != at+1 || len(l.readsSaid) != saidBefore+1 { + t.Errorf("an unchanged statement wrote the record or said it again") + } + catalogue.UseSeats(all) + q.RestateReads(context.Background()) + if got, _ = nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); !slices.Equal(got.Record.ManifestReads, catalogue.ManifestReads()) { + t.Errorf("the list back whole: the record states %v, want its contract's", got.Record.ManifestReads) + } } // 1132/AC7: a write the bucket refuses is said in the log, and the next apply writes again. diff --git a/internal/link/record.go b/internal/link/record.go index b228959..6ce8983 100644 --- a/internal/link/record.go +++ b/internal/link/record.go @@ -6,6 +6,7 @@ import ( sdkkeys "git.novox.be/novox/mesh-sdk/go/keys" "slices" "sort" + "strings" "sync" "time" @@ -34,13 +35,68 @@ type RecordKeys struct { Stated []string } -// manifestReads is what this engine reads of a manifest (novox/hq ADR 0614 §4): the reader capabilities of the mesh-sdk it -// is built with, stated in its record so the assigner moves a want only to an engine that reads its manifest. Computed -// once: it is the build's, and never changes while it runs. -var manifestReads = sync.OnceValue(catalogue.ManifestReads) +// buildReads are the words and forms this engine reads of a manifest (novox/hq ADR 0614 §4): the reader capabilities of +// the mesh-sdk it is built with, its seats left out. Computed once: they are the build's, and never change while it runs. +var buildReads = sync.OnceValue(func() []string { + return slices.DeleteFunc(catalogue.ManifestReads(), func(r string) bool { return strings.HasPrefix(r, catalogue.ReadSeat) }) +}) -// readsStatedAt is when this engine computed what it reads: its start (the verb manifest-reads, ADR 0618 §3). -var readsStatedAt = time.Now().UTC() +// readsNow is what this engine reads of a manifest now: its build's words and forms, and the seats of the seat list it +// holds now (catalogue.Seats, which the engine's reader of the list gives catalogue: bootstrapped or read; ADR 0636). +func readsNow() []string { + set := map[string]bool{} + for _, r := range buildReads() { + set[r] = true + } + for _, seat := range catalogue.Seats() { + set[catalogue.ReadSeat+seat.Name] = true + } + out := make([]string, 0, len(set)) + for r := range set { + out = append(out, r) + } + sort.Strings(out) + return out +} + +// readsStatement is what this engine states it reads and since when (the verb manifest-reads, ADR 0618 §3): made at its +// first reading, and made again by Restate whenever the seat list it holds changes what it reads (ADR 0618 §4). The seat +// list is the process's, so the statement is too. +type readsStatement struct { + mu sync.Mutex + reads []string + at time.Time +} + +var stated readsStatement + +// now is the statement, made at the first reading. +func (s *readsStatement) now() ([]string, time.Time) { + s.mu.Lock() + defer s.mu.Unlock() + if s.reads == nil { + s.reads, s.at = readsNow(), time.Now().UTC() + } + return slices.Clone(s.reads), s.at +} + +// restate makes the statement again from what the engine reads now, and says whether it changed. +func (s *readsStatement) restate() ([]string, bool) { + cur := readsNow() + s.mu.Lock() + defer s.mu.Unlock() + if s.reads != nil && slices.Equal(cur, s.reads) { + return slices.Clone(cur), false + } + s.reads, s.at = cur, time.Now().UTC() + return slices.Clone(cur), true +} + +// manifestReads is what this engine states it reads of a manifest now, which its record carries. +func manifestReads() []string { + reads, _ := stated.now() + return reads +} // recordOf is the node's record from one facts report and its keys; nil for a report that carries no profile (one that // says nothing of the machine, such as a rekey), so the record last written stands. diff --git a/internal/link/session.go b/internal/link/session.go index a05b5f3..7068db1 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -93,6 +93,8 @@ type Session struct { sync.Mutex linked bool } + // restating takes RestateReads one at a time. + restating sync.Mutex // haves is what this node-engine writes beside its facts (novox/hq ADR 0429 §4, issue 1132). haves haveState @@ -390,10 +392,42 @@ func (q *Session) RouterRecord(key string) (*stdio.StateEntry, error) { return asking.RouterRecord(key) } -// StatedReads is what this engine states it reads of a manifest and when it computed them, which is its start (the -// node seat's verb manifest-reads, novox/hq ADR 0618 §3): the record is written only when it differs, so its put says -// nothing of when an unchanged statement was made. -func (q *Session) StatedReads() ([]string, time.Time) { return manifestReads(), readsStatedAt } +// StatedReads is what this engine states it reads of a manifest and when it made that statement, at its start or at the +// last change of its seat list that changed it (the node seat's verb manifest-reads, novox/hq ADR 0618 §3): the record is +// written only when it differs, so its put says nothing of when an unchanged statement was made. +func (q *Session) StatedReads() ([]string, time.Time) { return stated.now() } + +// RestateReads states what this engine reads again, after the seat list it holds changed (novox/hq ADR 0618 §4, ADR +// 0636): when the statement changed, its node record carries the new one at once, written where it differs, and the node +// seat's event manifest-reads-changed says it ("started": what it states changed), so the assigner and the builder ask +// the verb again. A statement that did not change writes and says nothing. Calls are taken one at a time, so the last +// write is the newest statement. +func (q *Session) RestateReads(ctx context.Context) { + q.init() + q.restating.Lock() + defer q.restating.Unlock() + reads, changed := stated.restate() + if !changed { + return + } + s := &q.haves + s.mu.Lock() + if s.node != nil && s.node.Record != nil { + node, rec := *s.node, *s.node.Record + rec.ManifestReads = reads + node.Record = &rec + s.node = &node + } + s.mu.Unlock() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + if bus != nil { + q.writeHaves(ctx, bus) + } + q.SayEngine(nodeengine.EventManifestReadsChanged, nodeengine.ManifestReadsChangedEvent{V: sdkkeys.Version, + Node: q.Membership.Node, ManifestReads: reads, At: time.Now().UTC().Format(time.RFC3339Nano), Why: ReadsStarted}) +} // The reasons the engine says its reads, in the event's why (novox/hq ADR 0618 §4): it started, or its node returned // after being away (a link after the first).