diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go index 3b9d5ba1..ef2e4617 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/askended.go @@ -18,7 +18,6 @@ import ( oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "mesh-delivery-assigner/internal/assign" - plc "mesh-delivery-assigner/internal/assignment" "mesh-delivery-assigner/internal/buckets" "mesh-delivery-assigner/internal/puts" ) @@ -119,11 +118,7 @@ func (cw callWait) keep(ctx context.Context, id, callee string, call func() (jso if said || clock().Sub(since) < mcc.CallUnansweredAfter || cw.heard == nil || !cw.heard() || cw.findings.Put == nil { continue } - o := mcc.CallUnanswered(id, callee, "", err.Error(), "assigner."+mcc.KindCallUnanswered, since) - value, eerr := mcc.EncodeFinding(o.Finding(plc.Module, "", clock().UTC())) - if eerr == nil { - eerr = cw.findings.Put(key, value) - } + eerr := cw.findings.Put(assign.CallUnanswered(id, callee, "", err.Error(), "assigner."+mcc.KindCallUnanswered, since).Finding(clock())) if eerr != nil { logf("the finding %s is not written: %v", key, eerr) continue diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go index 07829862..b57d1575 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/halves.go @@ -9,99 +9,64 @@ import ( "sync" "time" - "github.com/nats-io/nats.go/jetstream" - mcc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/views" ) // halves follows the catalog half of each watched branch through the seat mesh-catalog (design 47 G §14, §184, §195): one -// meshcatalogclient.Catalog per branch, a first view of the verb `revision` and then each `catalog-revision` of that branch, -// each entry taken into the world as the value and the graph node the rules read. Of the view the assigner reads only the catalog -// half, never its own output back (§195). +// meshcatalogclient.Catalog per branch, adopted through the module's views reader (A-26), which asks the first view of the verb +// `revision`, hears each `catalog-revision` of that branch and raises and clears the follower's findings; each entry is taken +// into the world as the value and the graph node the rules read. Of the view the assigner reads only the catalog half, never its +// own output back (§195). type halves struct { - w *world - call mcc.Call - wake func() - logf func(string, ...any) - ctx context.Context + w *world + call mcc.Call + reader *views.ProcessReader + wake func() + logf func(string, ...any) + ctx context.Context mu sync.Mutex cats map[string]*mcc.Catalog - // started is called with each catalog made, for the part that reads their status. - started func(*mcc.Catalog) // settled is told once when a branch's first view is answered, for the first pass to wait on (the trunk's alone). settled chan<- struct{} } -// follow makes the catalog of branch and asks its first view, once; true when it was made now. +// follow adopts the catalog of branch through the reader, once; true when it was made now. func (h *halves) follow(branch string) bool { h.mu.Lock() if h.cats[branch] != nil { h.mu.Unlock() return false } - c, err := mcc.NewCatalog(h.call, branch) + c, _, err := mcc.WireCatalogProcess(h.reader, h.call, branch, func(module string, entry json.RawMessage, deleted bool, _ time.Time) { + h.changed(branch, module, entry, deleted) + }) if err != nil { h.mu.Unlock() h.logf("the catalog half of branch %q is not followed: %v", branch, err) return false } - c.Follower.Logf = h.logf - c.Follower.Changed = func(module string, entry json.RawMessage, deleted bool, _ time.Time) { - h.changed(branch, module, entry, deleted) - } if h.cats == nil { h.cats = map[string]*mcc.Catalog{} } h.cats[branch] = c h.mu.Unlock() - if h.started != nil { - h.started(c) - } - go func() { - if err := c.Start(h.ctx); err != nil { - if h.ctx.Err() == nil { - h.logf("the first view of the catalog half of branch %q is not answered: %v", branch, err) + if branch == trunkBranch && h.settled != nil { + go func() { + select { + case <-c.Ready(): + h.settled <- struct{}{} + case <-h.ctx.Done(): } - return - } - if branch == trunkBranch && h.settled != nil { - h.settled <- struct{}{} - } - }() + }() + } return true } -// hear gives a catalog-revision to every catalog: each takes its own branch's and passes over another's. -func (h *halves) hear(m jetstream.Msg) { - h.mu.Lock() - cats := make([]*mcc.Catalog, 0, len(h.cats)) - for _, c := range h.cats { - cats = append(cats, c) - } - h.mu.Unlock() - for _, c := range cats { - if err := c.Hear(h.ctx, m.Data()); err != nil { - h.logf("%s on branch %q: %v", m.Subject(), c.Branch, err) - } - } - _ = m.Ack() -} - -// followers are the first views the catalogs hold. -func (h *halves) followers() []*mcc.Catalog { - h.mu.Lock() - defer h.mu.Unlock() - out := make([]*mcc.Catalog, 0, len(h.cats)) - for _, c := range h.cats { - out = append(out, c) - } - return out -} - // changed takes one entry of branch's half into the world, or a module taken out of it. func (h *halves) changed(branch, module string, raw json.RawMessage, deleted bool) { defer h.wake() diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index e10e9312..618cfb82 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "fmt" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/keys" "io" "maps" @@ -52,7 +53,6 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-assigner/internal/assign" @@ -179,6 +179,10 @@ func startBed(t *testing.T, leaveOut string) *bed { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { t.Fatal(err) } + // The assigner's own stream of its findings' events (design 47 G §383), which its finding writer says into. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNER_FINDINGS", Subjects: []string{"mesh.stream.mesh-delivery-assigner.event.finding-written.*", "mesh.stream.mesh-delivery-assigner.event.finding-cleared.*", "mesh.stream.mesh-delivery-assigner.event.findings-stated.*"}}); err != nil { + t.Fatal(err) + } // The assigner's durable consumer on EVENTS, as the grants part makes it from the manifest's consumes: a node-engine seat's // event names the engine's machine last, so the filter is the event with every machine (mesh-grants buscompose // machineEvent; the grants part's own test holds the derivation). @@ -508,7 +512,8 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { "seat-holders": {"n1.node-tools"}, "wants": {"n1.mesh-grants"}, }) - graphKey := meshconditionsclient.Key(meshconditionsclient.ScopeModule, assignment.Module+"."+meshcatalog.Seat+"."+views.BranchState(meshcatalog.StateCatalogRevisions, trunkBranch), views.UnansweredKind) + // The reader raises the finding of the first view through the module's finding writer (A-26), its id escaped as a key's name. + graphKey := views.FindingKey(views.FindingScope, keys.EscapeName(assignment.Module+"."+meshcatalog.Seat+"."+views.BranchState(meshcatalog.StateCatalogRevisions, trunkBranch)), views.UnansweredKind) graphRaised := func() bool { e, err := b.buckets[assign.FindingsBucket].Get(b.ctx, graphKey) return err == nil && strings.Contains(string(e.Value()), "mesh-catalog.revision") @@ -855,12 +860,12 @@ func (b *bed) conditionsSaid(t *testing.T) map[string]oc.Condition { c := out[e.Key()] c.Key, c.State = e.Key(), occ.StateCleared if e.Operation() == jetstream.KeyValuePut { - f, err := meshconditionsclient.DecodeFinding(e.Value()) - if err != nil { + var f mcseat.Finding + if err := json.Unmarshal(e.Value(), &f); err != nil { t.Fatal(err) } c = oc.Condition{Key: e.Key(), State: occ.StateRaised, Severity: string(f.Severity), Explanation: f.Explanation, - Resolved: f.Resolved, Subject: oc.ConditionSubject{Scope: string(f.Subject.Scope), ID: f.Subject.ID, Machine: f.Subject.Machine}} + Subject: oc.ConditionSubject{Scope: string(f.Subject.Scope), ID: f.Subject.ID, Machine: f.Subject.Machine}} } out[e.Key()] = c } 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 e99492ea..ffd7ab90 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -8,7 +8,6 @@ package main import ( "context" - "errors" "fmt" "io" "log" @@ -18,6 +17,7 @@ import ( "os" "os/signal" "slices" + "strings" "syscall" "time" @@ -25,7 +25,7 @@ import ( "github.com/nats-io/nats.go/jetstream" mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" - mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/announce" @@ -110,7 +110,7 @@ var keptStates = []announce.State{ {Name: "differences", Readable: true}, {Name: "puts", Readable: true}, {Name: "wants", Readable: true}, - {Name: mcc.FindingsState, Readable: true}, + {Name: assign.FindingsState, Readable: true}, {Name: buckets.PoliciesState, Readable: true}, {Name: mma.StateSeatHolders, Readable: true}, {Name: views.MarksState, Readable: true}, // what its views family said each state up to (views.go) @@ -235,11 +235,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } thisNode = account.Node logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs)) - findings := findingsOf(ctx, kvs, logger.Printf) + findings, writer := findingsOf(ctx, account, conn, js, said, kvs, logger.Printf) + if writer != nil { + defer writer.Close() + } // Each change of the states its readers read is said on its views family (ADR 0615 §5, ADR 0619 §2); a watch of it // that ends stops the program, as a watch of what a pass reads does. viewErrs := make(chan error, len(viewedStates)) - startViews(ctx, js, conn, account, kvs, findings, viewErrs, logger.Printf) + startViews(ctx, js, conn, account, kvs, wireOf(writer), viewErrs, logger.Printf) go func() { select { case <-ctx.Done(): @@ -248,27 +251,52 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro conn.Close() } }() - iv := &catalogViews{js: js, account: account, call: said.Call, findings: findings, logf: logger.Printf, + iv := &catalogViews{js: js, account: account, call: said.Call, writer: wireOf(writer), logf: logger.Printf, status: process.AwaitBucket(ctx, js, conn, firstViewsBucket, "assigner_first_views reads nothing and assigner_first_view_again is not heard", logger.Printf)} return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, iv, logger) } // findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, which the // condition keeper carries (novox/hq ADR 0590 §3). Without it the conditions are said straight to the router. -func findingsOf(ctx context.Context, kvs kvSet, logf func(string, ...any)) assign.Findings { +func findingsOf(ctx context.Context, account process.Account, conn *nats.Conn, js jetstream.JetStream, said *process.Conn, kvs kvSet, logf func(string, ...any)) (assign.Findings, *heldWriter) { var f assign.Findings - if kv := kvs[assign.FindingsBucket]; kv != nil { - f.Put = func(key string, value []byte) error { _, err := kv.Put(ctx, key, value); return err } - f.Delete = func(key string) error { return kv.Delete(ctx, key) } - f.Keys = func() ([]string, error) { - keys, err := kv.Keys(ctx) - if errors.Is(err, jetstream.ErrNoKeysFound) { - return nil, nil - } - return keys, err - } + if kvs[assign.FindingsBucket] == nil { + return f, nil } - return f + state := said.State(assign.FindingsState) + w, err := views.NewHeldWriter(views.WriterConfig[mcseat.Finding, mcseat.FindingCleared, mcseat.FindingsStated, mcseat.FindingsAskedEvent]{ + Module: account.Module, State: state, Node: account.Node, Asked: mcseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf}) + if err != nil { + logf("the assigner's finding writer cannot be made, so its findings are not written: %v", err) + return f, nil + } + // A writer that cannot start (the stream of its asks is not there yet) is said; its findings are then kept in the + // state alone and said by the next start (as the node-engine's is). + if err := w.Start(ctx); err != nil { + logf("the finding writer cannot start, its findings are kept in the state alone: %v", err) + } + f.Put = func(x mcseat.Finding) error { return w.Write(ctx, x) } + f.Delete = func(key string) error { + scope, rest, _ := strings.Cut(key, ".") + i := strings.LastIndex(rest, ".") + if i < 0 { + return fmt.Errorf("%q is no finding's key", key) + } + return w.Clear(ctx, scope, rest[:i], rest[i+1:], "the assigner no longer observes it") + } + f.Keys = state.Keys + return f, w +} + +// heldWriter is the assigner's finding writer: its findings are raised as they happen, so the writer holds them (design 47 G A-29). +type heldWriter = views.HeldWriter[mcseat.Finding, mcseat.FindingCleared, mcseat.FindingsStated, mcseat.FindingsAskedEvent] + +// wireOf is w as the views parts' finding writer; nil when the assigner has none, so no part writes through a nil writer. +func wireOf(w *heldWriter) views.WireWriter { + if w == nil { + return nil + } + return w } // thisNode is the node this process runs on, as its credential names it: said with each of the seat's events. diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index f1a66b44..3155a771 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -1,6 +1,9 @@ package main import ( + "encoding/json" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-assigner/internal/assign" plc "mesh-delivery-assigner/internal/assignment" "mesh-delivery-assigner/internal/buckets" @@ -31,8 +34,7 @@ import ( // (`.` read as the bucket `_`), so it is granted what it reads and reads nothing else. func TestOpensWhatItsManifestNames(t *testing.T) { // Review round 2: the assigner's views emitter writes its finding view-unsaid (Emitter.Findings set) when it keeps findings. - findings := assign.Findings{Put: func(string, []byte) error { return nil }, Delete: func(string) error { return nil }} - if viewsEmitter(nil, process.Account{}, nil, findings, nil, t.Logf).Findings == nil { + if viewsEmitter(nil, process.Account{}, nil, &heldWriter{}, t.Logf).Findings == nil { t.Error("the assigner's views emitter writes no view-unsaid finding") } // 1297/B2a: without a credential the program does not start, and says why. @@ -197,3 +199,12 @@ func (e kvEntry) Created() time.Time { return time.Time{} } func (w *watchKV) Keys(context.Context, ...jetstream.WatchOpt) ([]string, error) { return nil, jetstream.ErrNoKeysFound } + +// putInto is an assign.Findings Put that keeps each finding in into, as the finding writer would, by its key and its words. +func putInto(into map[string][]byte) func(mcseat.Finding) error { + return func(f mcseat.Finding) error { + raw, err := json.Marshal(f) + into[views.FindingKey(f.Scope, f.ID, f.Kind)] = raw + return err + } +} 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 33382b2d..1999a44b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -149,13 +149,9 @@ func (p *passer) unansweredReads(known []string, online map[string]bool) { if p.calls == nil || p.calls.Keeper != p.reads { p.calls = &enginereads.ReadsCalls{Keeper: p.reads, Bound: mcc.CallUnansweredAfter, Now: p.clock, Logf: p.logf, Raise: func(_ context.Context, u enginereads.UnansweredCall) error { - o := mcc.CallUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "assigner."+mcc.KindCallUnanswered, u.Since) + o := assign.CallUnanswered(enginereads.CallID(u.Node), enginereads.Invoke, u.Node, u.Words, "assigner."+mcc.KindCallUnanswered, u.Since) o.Needs = "assigner_engine_reads with ask_again asks the node again now and shows its statement." - value, err := mcc.EncodeFinding(o.Finding(plc.Module, "", u.At)) - if err == nil { - err = p.findings.Put(o.Key(), value) - } - return err + return p.findings.Put(o.Finding(u.At)) }, Clear: func(_ context.Context, node string) error { return p.findings.Delete(mcc.Key(mcc.ScopeCall, enginereads.CallID(node), mcc.KindCallUnanswered)) @@ -1126,9 +1122,9 @@ func (p *passer) provisionFindings(failed map[string]meshprovisionerclient.Provi if p.failedProvisions[key] { continue } - value, err := assign.ProvisionFailedFinding(key, pr, now()) + f, err := assign.ProvisionFailedFinding(key, pr, now()) if err == nil { - err = p.findings.Put(assign.ProvisionFailedKey(key), value) + err = p.findings.Put(f) } if err != nil { p.logf("the failed provision %s is not said: %v", key, err) @@ -1178,9 +1174,9 @@ func (p *passer) provisionsMissing(ctx context.Context, missing map[string]bool) if p.missingSaid[key] || p.findings.Put == nil { continue } - value, err := assign.ProvisionMissingFinding(key, MissingBound, p.clock()) + f, err := assign.ProvisionMissingFinding(key, MissingBound, p.clock()) if err == nil { - err = p.findings.Put(assign.ProvisionMissingKey(key), value) + err = p.findings.Put(f) } if err != nil { p.logf("the missing provision %s is not said: %v", key, err) @@ -1226,9 +1222,9 @@ func (p *passer) unplacedFindings() { if p.unplaced[node] { continue } - value, err := assign.UnplacedFinding(node, p.clock()) + f, err := assign.UnplacedFinding(node, p.clock()) if err == nil { - err = p.findings.Put(assign.UnplacedKey(node), value) + err = p.findings.Put(f) } if err != nil { p.logf("the unplaced node %s is not said: %v", node, err) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 93654d0d..f884f342 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -209,7 +209,7 @@ func TestAHoldAskedAgainAfterALiftAndARestartHasANewID(t *testing.T) { for _, heard := range []bool{false, true} { clock, findings := now, map[string][]byte{} cw := callWait{heard: func() bool { return heard }, now: func() time.Time { return clock }, - findings: assign.Findings{Put: func(k string, v []byte) error { findings[k] = v; return nil }, + findings: assign.Findings{Put: putInto(findings), Delete: func(k string) error { delete(findings, k); return nil }}} tries, during := 0, 0 answer, err := cw.keep(context.Background(), asksCallID, channelAsksVerb, func() (json.RawMessage, error) { @@ -332,7 +332,7 @@ func TestAnAnswerCountsOnlyForTheAskingItAnswered(t *testing.T) { return nil, context.DeadlineExceeded } q := &passer{w: w, logf: t.Logf, now: func() time.Time { return clock }, findings: assign.Findings{ - Put: func(k string, v []byte) error { held[k] = v; return nil }, + Put: putInto(held), Delete: func(k string) error { delete(held, k); return nil }, Keys: func() ([]string, error) { return slices.Collect(maps.Keys(held)), nil }}, reads: &enginereads.Keeper{Call: silent, Sync: true, Now: func() time.Time { return clock }}} 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 0cd6b438..792f9f18 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 @@ -24,7 +24,6 @@ import ( mmac "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" - mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -147,6 +146,9 @@ func TestTheSeatsVerbs(t *testing.T) { } } for _, stream := range raw0.Streams { + if strings.HasSuffix(stream.Name, "-findings") { // the part's own finding events, not a state of the seat + continue + } if !reads["stream "+stream.Name] { t.Errorf("no verb reads the seat's stream %s", stream.Name) } @@ -283,7 +285,7 @@ func TestTheSeatsVerbs(t *testing.T) { } findings := map[string][]byte{} p := &passer{w: w, logf: func(string, ...any) {}, findings: assign.Findings{ - Put: func(k string, v []byte) error { findings[k] = v; return nil }, + Put: putInto(findings), Delete: func(k string) error { delete(findings, k); return nil }}} p.unplacedFindings() if _, ok := findings[assign.UnplacedKey("g14")]; !ok || len(findings) != 1 { @@ -471,7 +473,7 @@ func TestTheSeatsVerbs(t *testing.T) { 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: kvSet{buckets.Provisions: nil}, findings: assign.Findings{ - Put: func(k string, v []byte) error { findings[k] = v; return nil }, + Put: putInto(findings), Delete: func(k string) error { delete(findings, k); return nil }}} failed, _ := meshprovisionerclient.Encode(meshprovisionerclient.Provision{Provision: "db", Consumer: meshprovisionerclient.End{Module: "photos", Node: "ace"}, Provider: meshprovisionerclient.End{Module: "postgres", Node: "novox"}, State: meshprovisionerclient.Failed, Why: "the role exists already"}) @@ -612,7 +614,7 @@ func wantsVerbs(t *testing.T) { state, watches[state], watches[state+" by key"], indexer.asks[state]) } } - for _, state := range []string{haveState, mcc.FindingsState} { + for _, state := range []string{haveState, assign.FindingsState} { if watches[state] != 1 { t.Errorf("%s was watched %d times, want once", state, watches[state]) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 4186a832..f52e96e7 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -6,23 +6,19 @@ import ( "mesh-delivery-assigner/internal/assignment" "mesh-delivery-assigner/internal/buckets" "slices" - "sync" "time" "github.com/nats-io/nats.go/jetstream" mma "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" - mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" - sdkfindings "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/findings" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient/heard" - mcat "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" - "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "mesh-delivery-assigner/internal/assign" plc "mesh-delivery-assigner/internal/assignment" @@ -47,7 +43,7 @@ var readTools = []viewedTool{ {"assigner_read_assignments", mma.StateAssignments, "where each module is placed, by module."}, {"assigner_read_seat_holders", mma.StateSeatHolders, "who holds which seat on which node."}, {"assigner_read_wants", "wants", "what each node should run: want..."}, - {"assigner_read_findings", mcc.FindingsState, "the assigner's findings, which the condition keeper raises and clears."}, + {"assigner_read_findings", assign.FindingsState, "the assigner's findings, which the condition keeper raises and clears."}, {"assigner_read_asks", "asks", "the asks the assigner put to a person, and their answers."}, {"assigner_read_differences", "differences", "every difference between a want and what a node runs."}, {"assigner_read_copies", copiesState, "the copies the assigner made of a retired state: when the copy into assignments ended and how many keys it copied."}, @@ -58,7 +54,7 @@ var viewedStates = map[string]string{ mma.StateAssignments: plc.AssignmentsBucket, mma.StateSeatHolders: plc.SeatHoldersBucket, "wants": buckets.Wants, - mcc.FindingsState: assign.FindingsBucket, + assign.FindingsState: assign.FindingsBucket, "asks": buckets.Asks, "differences": buckets.Differences, copiesState: copiesBucket, @@ -83,18 +79,10 @@ func keptSource(name string) views.Source { return views.KeptSource{State: stdio // startViews says each change of the assigner's viewed states on its views family, once its marks state is bound, // each state as soon as its bucket is (kvs, or bound once it appears). A watch that ends is sent on errs. -func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, account process.Account, kvs kvSet, findings assign.Findings, errs chan<- error, logf func(string, ...any)) { - tooLarge := func(state, key string, size int) { - if findings.Put == nil { - return - } - if err := viewFindings(findings).Put(views.TooLargeFinding(assignment.Module, state, key, size, time.Now())); err != nil { - logf("views: the finding of %s.%s too large is not written: %v", state, key, err) - } - } +func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, account process.Account, kvs kvSet, writer views.WireWriter, errs chan<- error, logf func(string, ...any)) { process.AwaitBucket(ctx, js, nc, marksBucket, "no change is said on the views family, so every reader waits on its first view", logf).Ready(func(mk jetstream.KeyValue) { - em := viewsEmitter(js, account, mk, findings, tooLarge, logf) - for _, state := range []string{"asks", "differences", mcc.FindingsState, mma.StateAssignments, mma.StateSeatHolders, "wants", copiesState} { + em := viewsEmitter(js, account, mk, writer, logf) + for _, state := range []string{"asks", "differences", assign.FindingsState, mma.StateAssignments, mma.StateSeatHolders, "wants", copiesState} { state := state start := func(kv jetstream.KeyValue) { if err := em.Start(ctx, state, kv, errs); err != nil { @@ -111,35 +99,27 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib } // viewsEmitter is the assigner's emitter of its views family: a change not said past views.Bound is its finding -// view-unsaid, in its findings (review round 2); with no findings state it is said in the log only. -func viewsEmitter(js jetstream.JetStream, account process.Account, mk jetstream.KeyValue, findings assign.Findings, tooLarge func(string, string, int), logf func(string, ...any)) views.Emitter { - em := views.Emitter{Owner: assignment.Module, Subject: account.EventSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf} - if findings.Put != nil && findings.Delete != nil { - em.Findings = viewFindings(findings) - } - return em -} - -// viewFindings is the assigner's findings as the views' FindingWriter, in the seat mesh-conditions' layout. -func viewFindings(f assign.Findings) views.FindingWriter { - return sdkfindings.WriterOver(f.Put, f.Delete) +// view-unsaid, in its findings, written through its finding writer (review round 2); with none it is said in the log only. +func viewsEmitter(js jetstream.JetStream, account process.Account, mk jetstream.KeyValue, writer views.WireWriter, logf func(string, ...any)) views.Emitter { + return views.Emitter{Owner: assignment.Module, Subject: account.EventSubject, Publish: views.JetStreamPublish(js), Marks: mk, Findings: writer, Logf: logf} } // firstViewsBucket is the assigner's own state of its first views (views.StatusState). var firstViewsBucket = assignment.Module + "_" + views.StatusState -// catalogViews is the assigner's first view of the catalog half of each watched branch (design 47 G §14, §184, §195): asked -// at the start through the seat mesh-catalog's verb `revision` (the trunk's, then each branch a registration names), then -// followed on its event `catalog-revision`, asking once more on a gap; past the bound its finding names the seat and the verb, -// cleared by the answer. Its status is read by assigner_first_views and asked again by assigner_first_view_again. +// catalogViews is the assigner's first view of the catalog half of each watched branch (design 47 G §14, §184, §195, A-26): the +// module's views reader adopts each branch's follower, so it asks the seat mesh-catalog's verb `revision` (the trunk's, then +// each branch a registration names), follows its event `catalog-revision`, asks once more on a gap, and past the bound raises +// the finding naming the seat and the verb, cleared by the answer, in the assigner's findings. Its status is read by +// assigner_first_views and asked again by assigner_first_view_again. type catalogViews struct { - js jetstream.JetStream - account process.Account - call func(string, any) (json.RawMessage, error) - findings assign.Findings - status *process.AwaitedBucket - logf func(string, ...any) - hub *process.EventHub + js jetstream.JetStream + account process.Account + call func(string, any) (json.RawMessage, error) + writer views.WireWriter + status *process.AwaitedBucket + logf func(string, ...any) + hub *process.EventHub h *halves } @@ -149,70 +129,31 @@ type catalogViews struct { // seconds). A test shortens them. var firstViewBound, firstViewWait time.Duration -// start follows the trunk's catalog half and hears the seat's catalog-revision; it answers how many settles to wait for. +// start adopts the trunk's catalog half through the reader and starts the reader; it answers how many settles to wait for. func (iv *catalogViews) start(ctx context.Context, w *world, wake func(), settled chan<- struct{}) int { - finding := func(s views.Status, raised bool) { - if iv.findings.Put == nil { - return - } - f := views.UnansweredFinding(assignment.Module, s, time.Now()) - var err error - if raised { - err = viewFindings(iv.findings).Put(f) - } else { - err = viewFindings(iv.findings).Delete(f) - } - if err != nil { - iv.logf("views: the finding of the first view %s is not written: %v", s.Tool, err) - } + reader := &views.ProcessReader{Module: assignment.Module, Call: iv.call, Logf: iv.logf, Hub: iv.hub, Bound: firstViewBound, Wait: firstViewWait} + if iv.writer != nil { + reader.Findings = func() views.WireWriter { return iv.writer } } - var statusKV jetstream.KeyValue - var mu sync.Mutex - iv.h = &halves{w: w, call: iv.call, wake: wake, logf: iv.logf, ctx: ctx, settled: settled} - iv.h.started = func(c *mcat.Catalog) { - f := c.Follower - f.Bound, f.Wait = firstViewBound, firstViewWait - f.Raise, f.Clear = func(s views.Status) { finding(s, true) }, func(s views.Status) { finding(s, false) } - mu.Lock() - defer mu.Unlock() - if statusKV != nil { - f.Said = views.KeepStatus(ctx, statusKV, iv.logf) - } - } - if iv.status != nil { - iv.status.Ready(func(kv jetstream.KeyValue) { - mu.Lock() - statusKV = kv - mu.Unlock() - var all []*views.Follower - for _, c := range iv.h.followers() { - c.Follower.Said = views.KeepStatus(ctx, kv, iv.logf) - all = append(all, c.Follower) - } - if err := views.HearAgain(ctx, kv, all, iv.logf); err != nil { - iv.logf("views: assigner_first_view_again is not heard (%v) until the assigner starts again", err) - } - }) - } - // The hub says when the bus made its consumer: a picture asked before then is asked again (views.Follower.ConsumerMade). - iv.hub.OnBound(func(created time.Time) { - for _, c := range iv.h.followers() { - c.Follower.ConsumerMade(ctx, created) - } - }) - // Not heard, the catalog-revision is the finding family-not-heard until it binds (ADR 0615 §5): the handler is registered - // on the module's hub, which the assigner starts once every handler is. - _ = notHeardWriter(iv.findings, iv.logf).Bind(process.EventsStream, "the catalog half is the first views' alone until the assigner starts again", func() error { - return iv.hub.Handle(meshcatalog.SubjectCatalogRevisionAll, iv.h.hear) - }) + iv.h = &halves{w: w, call: iv.call, reader: reader, wake: wake, logf: iv.logf, ctx: ctx, settled: settled} iv.h.follow(trunkBranch) + reader.Start(ctx, iv.js, iv.account, iv.status) return 1 } // notHeardWriter writes the assigner's finding of a family it does not hear into its findings (heard.Writer); with no // findings state it is said in the log alone. func notHeardWriter(findings assign.Findings, logf func(string, ...any)) heard.Writer { - w := heard.Writer{Observer: assignment.Module, Logf: logf, Put: findings.Put} + w := heard.Writer{Observer: assignment.Module, Logf: logf} + if findings.Put != nil { + w.Put = func(_ string, value []byte) error { + var f mcseat.Finding + if err := json.Unmarshal(value, &f); err != nil { + return err + } + return findings.Put(f) + } + } if findings.Delete != nil { w.Delete = func(key string) error { if keys, err := findings.Keys(); err == nil && !slices.Contains(keys, key) { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go index a2da5bfb..747b3027 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/wants_tools.go @@ -16,7 +16,6 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" - mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -170,7 +169,7 @@ func (s *wantsTools) doUpgrade(args map[string]any, by string) (any, error) { out["published"] = published(values) } // The assigner's findings, its said conditions as the condition keeper reads them (§15). - if found, err := s.values(mcc.FindingsState); err != nil { + if found, err := s.values(assign.FindingsState); err != nil { out["findings-unread"] = err.Error() } else { keys := []string{} diff --git a/modules/mesh-delivery-assigner/internal/assign/asks_test.go b/modules/mesh-delivery-assigner/internal/assign/asks_test.go index 5264609f..f8c560a9 100644 --- a/modules/mesh-delivery-assigner/internal/assign/asks_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/asks_test.go @@ -16,10 +16,10 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-assigner/asks" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-view/viewstate" - mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + "git.novox.be/novox/mesh-sdk/go/views" ) // 1297/B9 (the assigner's half): the ask about a held fingerprint is a valid ask of the router and of mesh-sdk's layout at @@ -292,11 +292,10 @@ func saidConditionsAreFindings(t *testing.T) { {SeatWaitsCondition([]string{"assignments"}), SeatWaitsCondition(nil), "module.mesh-delivery-assigner.seat-states"}, {BackupCondition("ace", "mail", fNew, "disk full", true), BackupCondition("ace", "mail", fNew, "", false), "module.ace.mail." + strings.TrimPrefix(fNew, meshcatalogclient.SourcePrefix2) + ".backup"}, } { - puts, deletes, said := map[string]mcseat.FindingsByKeyValue{}, []string{}, 0 - f := Findings{Put: func(k string, v []byte) error { - got, err := mcc.DecodeFinding(v) - puts[k] = got - return err + puts, deletes, said := map[string]mcseat.Finding{}, []string{}, 0 + f := Findings{Put: func(f mcseat.Finding) error { + puts[views.FindingKey(f.Scope, f.ID, f.Kind)] = f + return nil }, Delete: func(k string) error { deletes = append(deletes, k); return nil }, Now: func() time.Time { return at }} say := f.Say(func(oc.Condition) error { said++; return nil }) if err := say(c.raised); err != nil { @@ -314,7 +313,7 @@ func saidConditionsAreFindings(t *testing.T) { t.Fatalf("%s: with no findings bucket said %d (%v), want straight once", c.key, said, err) } } - if _, _, _, err := FindingOf(oc.Condition{Key: "another.x", State: occ.StateRaised}, at); err == nil { + if _, _, err := FindingOf(oc.Condition{Key: "another.x", State: occ.StateRaised}, at); err == nil { t.Error("a condition not the assigner's is a finding") } } diff --git a/modules/mesh-delivery-assigner/internal/assign/findings.go b/modules/mesh-delivery-assigner/internal/assign/findings.go index ca58dfc9..de663d5a 100644 --- a/modules/mesh-delivery-assigner/internal/assign/findings.go +++ b/modules/mesh-delivery-assigner/internal/assign/findings.go @@ -11,6 +11,7 @@ import ( mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + "git.novox.be/novox/mesh-sdk/go/views" ) // The assigner's said conditions become findings (novox/hq ADR 0564 §2, issue 1952 WM6, note 16): every condition it says @@ -21,15 +22,66 @@ import ( // assigner no longer says it straight (never both). // FindingsBucket is the assigner's findings. -var FindingsBucket = mcc.FindingsBucket(assignment.Module) +// FindingsState is the assigner's state of its findings, and FindingsBucket its bucket. +const FindingsState = "findings" -// FindingOf is a said condition as a finding: its key `..`, from the condition's own key -// `mesh-delivery-assigner..` (the subject's id when the key names none), its words unchanged. raised is false for -// a clear, whose key is to be deleted; value is nil then. -func FindingOf(c oc.Condition, at time.Time) (key string, value []byte, raised bool, err error) { +var FindingsBucket = assignment.Module + "_" + FindingsState + +// KeyID is an id as the parts of a key: each dot-separated part made safe for a bus key, as the key of a condition is +// (mcc.Key), so a finding's key keeps the shape its condition had. +func KeyID(id string) string { + key := mcc.Key("x", id, "x") + return strings.TrimSuffix(strings.TrimPrefix(key, "x."), ".x") +} + +// Observed is what the assigner observes wrong, before it is the seat mesh-conditions' generated finding. +type Observed struct { + Scope, ID, Kind, Machine string + Also []string + Severity string + Summary, Said string + Headline, Explanation string + Needs, Source string + Actions []mcseat.Action +} + +// Finding is o as the seat's generated finding at at (design 47 G §686, A-22): its id made safe as a key's part, the resolved +// words not a field of it, its observer the assigner; the writer completes its `since`. +func (o Observed) Finding(at time.Time) mcseat.Finding { + at = at.UTC() + id := KeyID(o.ID) + said := o.Said + if said == "" { + said = o.Summary + } + sev := o.Severity + if sev == "" { + sev = mcseat.FindingSeverityWarning + } + return mcseat.Finding{Scope: o.Scope, ID: id, Kind: o.Kind, Severity: sev, Summary: o.Summary, Headline: o.Headline, + Explanation: o.Explanation, Needs: o.Needs, Actions: o.Actions, Source: o.Source, Observer: assignment.Module, At: at, + Evidence: []mcseat.Evidence{{At: at, Said: said}}, + Subject: mcseat.Subject{Scope: o.Scope, ID: id, Machine: o.Machine, Also: o.Also}} +} + +// CallUnanswered is the observation of a kept call to callee (a seat verb's or a tool's key) about id (what the assigner waits +// on: a node, a key), unanswered since since, its last fault said. Its key is `call..call-unanswered`. +func CallUnanswered(id, callee, machine, fault, source string, since time.Time) Observed { + return Observed{Scope: mcc.ScopeCall, ID: id, Kind: mcc.KindCallUnanswered, Machine: machine, Severity: mcseat.FindingSeverityWarning, + Summary: fmt.Sprintf("the call %s about %s is unanswered since %s", callee, id, since.UTC().Format(time.RFC3339)), + Said: fault, Headline: "A call whose answer is kept is not answered", + Explanation: fmt.Sprintf("The call %s is made again on every transport fault and on the callee's event; it has "+ + "not been answered since %s. Its last fault: %s.", callee, since.UTC().Format(time.RFC3339), fault), + Source: source} +} + +// FindingOf is a said condition as a finding: from the condition's own key `mesh-delivery-assigner..` (the +// subject's id when the key names none), its words unchanged but the resolved ones. raised is false for a clear, whose +// finding names only its scope, id and kind. +func FindingOf(c oc.Condition, at time.Time) (f mcseat.Finding, raised bool, err error) { rest, ok := strings.CutPrefix(c.Key, assignment.Module+".") if !ok || rest == "" { - return "", nil, false, fmt.Errorf("the condition %q is not the assigner's", c.Key) + return f, false, fmt.Errorf("the condition %q is not the assigner's", c.Key) } kind, id, _ := strings.Cut(rest, ".") if id == "" { @@ -39,12 +91,11 @@ func FindingOf(c oc.Condition, at time.Time) (key string, value []byte, raised b if scope == "" { scope = mcc.ScopeModule } - key = mcc.Key(scope, id, kind) if c.State == occ.StateCleared { - return key, nil, false, nil + return mcseat.Finding{Scope: scope, ID: KeyID(id), Kind: kind}, false, nil } if c.State != occ.StateRaised { - return "", nil, false, fmt.Errorf("the condition %s is neither raised nor cleared", c.Key) + return f, false, fmt.Errorf("the condition %s is neither raised nor cleared", c.Key) } summary := c.Summary if summary == "" { @@ -54,16 +105,14 @@ func FindingOf(c oc.Condition, at time.Time) (key string, value []byte, raised b for _, a := range c.Actions { actions = append(actions, mcseat.Action{Label: a.Label, Verb: a.Verb, Machine: a.Machine, Arguments: a.Arguments, Level: string(a.Level)}) } - f := mcc.Observation{Scope: scope, ID: id, Kind: kind, Machine: c.Subject.Machine, Severity: string(c.Severity), Summary: summary, - Headline: c.Headline, Explanation: c.Explanation, Resolved: c.Resolved, Needs: c.Needs, Actions: actions, - Source: "assigner." + kind}.Finding(assignment.Module, "", at) - value, err = mcc.EncodeFinding(f) - return key, value, true, err + return Observed{Scope: scope, ID: id, Kind: kind, Machine: c.Subject.Machine, Severity: string(c.Severity), Summary: summary, + Headline: c.Headline, Explanation: c.Explanation, Needs: c.Needs, Actions: actions, Source: "assigner." + kind}.Finding(at), true, nil } -// Findings is where the assigner's findings go: the bucket's put and delete. +// Findings is where the assigner's findings go: its finding writer's write and clear (design 47 G §689, A-29), by the +// finding's key `..`. type Findings struct { - Put func(key string, value []byte) error + Put func(f mcseat.Finding) error Delete func(key string) error Keys func() ([]string, error) Now func() time.Time @@ -80,12 +129,13 @@ func (f Findings) Say(say func(oc.Condition) error) func(oc.Condition) error { if f.Now != nil { now = f.Now } - key, value, raised, err := FindingOf(c, now()) + finding, raised, err := FindingOf(c, now()) + key := views.FindingKey(finding.Scope, finding.ID, finding.Kind) switch { case err != nil: errs = append(errs, err) case raised: - if err := f.Put(key, value); err != nil { + if err := f.Put(finding); err != nil { errs = append(errs, fmt.Errorf("the finding %s cannot be written: %w", key, err)) } default: diff --git a/modules/mesh-delivery-assigner/internal/assign/policy.go b/modules/mesh-delivery-assigner/internal/assign/policy.go index 5a20f309..1e75c117 100644 --- a/modules/mesh-delivery-assigner/internal/assign/policy.go +++ b/modules/mesh-delivery-assigner/internal/assign/policy.go @@ -10,6 +10,7 @@ import ( "time" mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + mcseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" "git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -65,19 +66,17 @@ func ProvisionFailedKey(key string) string { } // ProvisionFailedFinding is the finding of a failed provision, naming its consumer, its provider and why. -func ProvisionFailedFinding(key string, p meshprovisionerclient.Provision, at time.Time) ([]byte, error) { +func ProvisionFailedFinding(key string, p meshprovisionerclient.Provision, at time.Time) (mcseat.Finding, error) { summary := fmt.Sprintf("%s for %s on %s failed at %s on %s: %s, so %s is not moved there until it is provided", p.Provision, p.Consumer.Module, p.Consumer.Node, p.Provider.Module, p.Provider.Node, p.Why, p.Consumer.Module) - return encode(mcc.Observation{Scope: mcc.ScopeProvider, ID: key, Kind: KindProvisionFailed, Machine: p.Consumer.Node, Also: also(p.Provider.Node, p.Consumer.Node), + return encode(Observed{Scope: mcc.ScopeProvider, ID: key, Kind: KindProvisionFailed, Machine: p.Consumer.Node, Also: also(p.Provider.Node, p.Consumer.Node), Summary: summary, Said: p.Why, Headline: clipHeadline(p.Consumer.Module + " waits for a provision that failed"), Explanation: occ.NothingToDo + " " + summary + ". It clears once the provider provides it.", - Resolved: p.Provision + " for " + p.Consumer.Module + " is provided", Source: "assigner." + KindProvisionFailed}, at) + Source: "assigner." + KindProvisionFailed}, at) } -// encode is the observation as the assigner's finding at at, in the seat's layout. -func encode(o mcc.Observation, at time.Time) ([]byte, error) { - return mcc.EncodeFinding(o.Finding(assignment.Module, "", at)) -} +// encode is the observation as the assigner's finding at at. +func encode(o Observed, at time.Time) (mcseat.Finding, error) { return o.Finding(at), nil } func also(provider, consumer string) []string { if provider == "" || provider == consumer { @@ -95,18 +94,18 @@ func ProvisionMissingKey(key string) string { } // ProvisionMissingFinding is the finding of a required provision with no key past bound (key `..`). -func ProvisionMissingFinding(key string, bound time.Duration, at time.Time) ([]byte, error) { +func ProvisionMissingFinding(key string, bound time.Duration, at time.Time) (mcseat.Finding, error) { parts := strings.SplitN(key, ".", 3) if len(parts) != 3 { - return nil, fmt.Errorf("%q is not ..", key) + return mcseat.Finding{}, fmt.Errorf("%q is not ..", key) } node, module, provision := parts[0], parts[1], parts[2] summary := fmt.Sprintf("%s on %s requires %s and mesh-provisioner has composed no key for it in %s, so %s is not moved there: "+ "no provider of it may be placed", module, node, provision, bound, module) - return encode(mcc.Observation{Scope: mcc.ScopeProvider, ID: key, Kind: KindProvisionMissing, Machine: node, + return encode(Observed{Scope: mcc.ScopeProvider, ID: key, Kind: KindProvisionMissing, Machine: node, Summary: summary, Headline: clipHeadline(module + " on " + node + " lacks " + provision), Explanation: occ.NothingToDo + " " + summary + ". It clears once the provision has its key.", - Resolved: provision + " for " + module + " on " + node + " has its key", Source: "assigner." + KindProvisionMissing}, at) + Source: "assigner." + KindProvisionMissing}, at) } // KindUnplaced is, from the lease write, a node that runs but no assignments names (ADR 0571 §6). @@ -118,13 +117,13 @@ func UnplacedKey(node string) string { } // UnplacedFinding is the finding of a node no assignments names since the lease write. -func UnplacedFinding(node string, at time.Time) ([]byte, error) { +func UnplacedFinding(node string, at time.Time) (mcseat.Finding, error) { summary := fmt.Sprintf("%s runs, but no assignments names it: since the controller handed its lease away nothing composes it, "+ "so the assigner writes it no want", node) - return encode(mcc.Observation{Scope: mcc.ScopeMachine, ID: node, Kind: KindUnplaced, Machine: node, Summary: summary, + return encode(Observed{Scope: mcc.ScopeMachine, ID: node, Kind: KindUnplaced, Machine: node, Summary: summary, Headline: clipHeadline(node + " has no assignments"), Needs: "place it with mesh-module-assigner.propose.", Explanation: occ.NeedsYou + "place it with mesh-module-assigner.propose. " + summary + ".", - Resolved: node + " is placed", Source: "assigner." + KindUnplaced}, at) + Source: "assigner." + KindUnplaced}, at) } func clipHeadline(s string) string { diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 986d15dd..0430440d 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -117,8 +117,9 @@ func TestManifest(t *testing.T) { // ADR 0635 §5: its views family is its ordinary events, `views.` among its emits, on no stream of its own. // The seat's four events are said on the seat's own subject (the seat's family), so the module keeps no stream for them // and lists none among its emits. - if len(m.Streams) != 0 { - t.Errorf("streams = %+v, want none: the seat's events ride the shared events stream", m.Streams) + // Its one stream is its own findings' events (design 47 G A-30, §383), which its finding writer says. + if len(m.Streams) != 1 || m.Streams[0].Name != "mesh-delivery-assigner-findings" || !slices.Equal(m.Streams[0].Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) { + t.Errorf("streams = %+v, want its findings' stream alone: the seat's events ride the shared events stream", m.Streams) } for _, e := range meshmoduleassigner.Events() { if slices.Contains(m.Emits, e) { @@ -127,7 +128,7 @@ func TestManifest(t *testing.T) { } // 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; // issue 1966 note 151: and the teller's ask-ended, its subject the one the assigner hears. - if !slices.Equal(m.Consumes, []string{"mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) { + if !slices.Equal(m.Consumes, []string{"mesh-conditions.findings-asked", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", buckets.TellerModule + "." + tellerevents.AskEnded, "mesh-pipeline-runner.pipeline-ended", "mesh-pipeline-runner.group-ended", "node-engine.manifest-reads-changed", "node-engine.running"}) { t.Errorf("consumes = %v, want the catalog half's revisions (design 47 G §195), the teller's ask-ended, the engines' reads changed and what they run", m.Consumes) } if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index e7f8e724..f9d921a5 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -42,6 +42,7 @@ "operator-channel" ], "consumes": [ + "mesh-conditions.findings-asked", "mesh-catalog.catalog-revision", "mesh-catalog.registration-changed", "mesh-catalog.seat-definition-changed", @@ -66,6 +67,14 @@ "seat:node-engine.manifest-reads", "seat:operator-channel.asks" ], + "streams": [ + { + "name": "mesh-delivery-assigner-findings", + "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], + "max-age-days": 7, + "max-mib": 16 + } + ], "emits": [ "assignments-copied", "put",