From 0fdf56f4bbd58e0d5bc541cd8161e6496891fd78 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 21:58:50 +0100 Subject: [PATCH] G findings4: the readers' findings go through the part's writer as views.WireWriter; the pipeline runner adopts each branch's catalog view through its reader MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §689, §687, A-22, A-26 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../mesh-delivery-checker/findingwriter.go | 15 ++-- .../cmd/mesh-delivery-checker/main.go | 4 +- .../internal/check/checker.go | 3 +- .../internal/check/checker_test.go | 7 +- .../internal/check/couldnotrun.go | 22 +++-- .../cmd/mesh-delivery-pipeline/views.go | 82 ++----------------- .../cmd/mesh-delivery-teller/main.go | 2 +- .../cmd/mesh-delivery-view/main.go | 2 +- .../cmd/mesh-delivery-watchman/main.go | 2 +- 9 files changed, 36 insertions(+), 103 deletions(-) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/findingwriter.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/findingwriter.go index ee0eeb3e..bb96deb7 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/findingwriter.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/findingwriter.go @@ -27,7 +27,7 @@ type checkerFindings struct { // newCheckerFindings makes and starts the writer; observe says the checks that could not run and whose finding is // written (check.Checker.RaisedCouldNotRun). func newCheckerFindings(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, - observe func(context.Context) ([]views.Finding, error), logf func(string, ...any)) (*checkerFindings, error) { + observe func(context.Context) ([]check.Finding, error), logf func(string, ...any)) (*checkerFindings, error) { c := &checkerFindings{node: cred.Node} w, err := views.NewWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ Module: cred.Module, State: state, Node: cred.Node, JS: js, Conn: conn, Logf: logf, @@ -57,7 +57,7 @@ func newCheckerFindings(ctx context.Context, cred process.Account, conn *nats.Co } // body lays the checker's finding out as the seat's contract type: a finding of the module, about the module on its node. -func (c *checkerFindings) body(f views.Finding) condseat.Finding { +func (c *checkerFindings) body(f check.Finding) condseat.Finding { module, _ := c.w.ModuleNode() return condseat.Finding{Kind: f.Kind, Scope: views.FindingScope, ID: keys.EscapeName(f.ID), At: f.At, Since: f.At, Severity: condseat.FindingSeverityWarning, Source: f.Kind, Summary: f.Summary, Headline: f.Headline, Explanation: f.Explanation, @@ -65,18 +65,13 @@ func (c *checkerFindings) body(f views.Finding) condseat.Finding { } // Put writes one finding; its `since` stays where its key exists. -func (c *checkerFindings) Put(f views.Finding) error { +func (c *checkerFindings) Put(f check.Finding) error { return c.w.Write(context.Background(), c.body(f)) } // Delete clears one finding. -func (c *checkerFindings) Delete(f views.Finding) error { +func (c *checkerFindings) Delete(f check.Finding) error { return c.w.Clear(context.Background(), views.FindingScope, keys.EscapeName(f.ID), f.Kind, "the checker no longer observes it") } -var _ views.FindingWriter = (*checkerFindings)(nil) - -// readerFindings is the views reader's findings through the same writer, as kinds of the reader's own. -func (c *checkerFindings) readerFindings() views.FindingWriter { - return views.WriterFindings{Writer: c.w} -} +var _ check.FindingWriter = (*checkerFindings)(nil) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 61cf7263..1d6548c4 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -164,9 +164,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // The finding writer is made once the checker is (its observe reads the checker's kept checks); until then a first view // unanswered is said in the log alone. var findings atomic.Pointer[checkerFindings] - reader.Findings = func() views.FindingWriter { + reader.Findings = func() views.WireWriter { if f := findings.Load(); f != nil { - return f.readerFindings() + return f.w } return nil } diff --git a/modules/mesh-delivery-checker/internal/check/checker.go b/modules/mesh-delivery-checker/internal/check/checker.go index 02912e41..98f4530d 100644 --- a/modules/mesh-delivery-checker/internal/check/checker.go +++ b/modules/mesh-delivery-checker/internal/check/checker.go @@ -20,7 +20,6 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" "git.novox.be/novox/mesh-delivery/sdk/buildstep" "git.novox.be/novox/mesh-sdk/go/keys" - "git.novox.be/novox/mesh-sdk/go/views" ) // DefaultBound is how long a check may run before it is ended failed naming the step it was stuck at (ADR 0484, issue 1678): @@ -94,7 +93,7 @@ type Checker struct { CouldNotRun KV // Findings writes the checker's finding of a check not asked again within CouldNotRunBound (CouldNotRunDue); nil // writes none. - Findings views.FindingWriter + Findings FindingWriter // CouldNotRunKept is called each time a check that could not run is kept, so the process waits for its bound anew. CouldNotRunKept func() // Bound is how long one check may run; none when 0 (DefaultBound). diff --git a/modules/mesh-delivery-checker/internal/check/checker_test.go b/modules/mesh-delivery-checker/internal/check/checker_test.go index 51e8775f..592a6e5c 100644 --- a/modules/mesh-delivery-checker/internal/check/checker_test.go +++ b/modules/mesh-delivery-checker/internal/check/checker_test.go @@ -5,7 +5,6 @@ import ( "encoding/json" "errors" "fmt" - "git.novox.be/novox/mesh-sdk/go/views" "strings" "sync" "testing" @@ -228,10 +227,10 @@ func TestACheckPastItsBoundFailsNamingTheStuckStep(t *testing.T) { // N6/AC3 } // findingsKept is a FindingWriter that keeps what is written, by ID. -type findingsKept map[string]views.Finding +type findingsKept map[string]Finding -func (f findingsKept) Put(x views.Finding) error { f[x.ID] = x; return nil } -func (f findingsKept) Delete(x views.Finding) error { delete(f, x.ID); return nil } +func (f findingsKept) Put(x Finding) error { f[x.ID] = x; return nil } +func (f findingsKept) Delete(x Finding) error { delete(f, x.ID); return nil } // A check that could not hear a part of the mesh says `could not run`, naming the part, and releases its claim; only the // running of that part asks it again, once per heads and part at one fingerprint; not asked again within an hour it is the diff --git a/modules/mesh-delivery-checker/internal/check/couldnotrun.go b/modules/mesh-delivery-checker/internal/check/couldnotrun.go index 643e06e7..1884aa55 100644 --- a/modules/mesh-delivery-checker/internal/check/couldnotrun.go +++ b/modules/mesh-delivery-checker/internal/check/couldnotrun.go @@ -10,7 +10,6 @@ import ( "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliverycheckerclient" "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshdeliverychecker" - "git.novox.be/novox/mesh-sdk/go/views" ) // A check that failed because a part of the mesh was not heard says `could not run`, naming the part, and releases its @@ -233,9 +232,22 @@ func (ch *Checker) CouldNotRunDue(ctx context.Context, now time.Time) time.Time return next } +// Finding is the checker's finding of a check that could not run, in plain fields: the checker's writer lays it out as the seat +// mesh-conditions' finding (design 47 G §689). +type Finding struct { + Kind, ID, Summary, Headline, Explanation string + At time.Time +} + +// FindingWriter writes and clears the checker's findings. +type FindingWriter interface { + Put(f Finding) error + Delete(f Finding) error +} + // RaisedCouldNotRun is every finding the checker observes now (design 47 G §384, §461): the check that could not run // and whose finding is written. The finding writer states them at its start and when the keeper asks. -func (ch *Checker) RaisedCouldNotRun(ctx context.Context) ([]views.Finding, error) { +func (ch *Checker) RaisedCouldNotRun(ctx context.Context) ([]Finding, error) { if ch.CouldNotRun == nil { return nil, nil } @@ -244,7 +256,7 @@ func (ch *Checker) RaisedCouldNotRun(ctx context.Context) ([]views.Finding, erro return nil, fmt.Errorf("the checks that could not run cannot be listed: %w", err) } sort.Strings(keys) - var out []views.Finding + var out []Finding for _, key := range keys { if r, _, ok := ch.couldNotRunOf(ctx, key); ok && r.Raised { out = append(out, couldNotRunFinding(key, r, r.Since)) @@ -254,8 +266,8 @@ func (ch *Checker) RaisedCouldNotRun(ctx context.Context) ([]views.Finding, erro } // couldNotRunFinding is the finding of the check of key that could not run. -func couldNotRunFinding(key string, r CouldNotRun, at time.Time) views.Finding { - return views.Finding{Observer: CheckerModule, Kind: CouldNotRunKind, ID: key, +func couldNotRunFinding(key string, r CouldNotRun, at time.Time) Finding { + return Finding{Kind: CouldNotRunKind, ID: key, Summary: fmt.Sprintf("the check of %s (heads %s) could not run since %s: %s was not heard, and has not run since", r.Group, r.Heads, r.Since.UTC().Format(time.RFC3339), r.Part), Headline: fmt.Sprintf("The check of %s could not run", r.Group), diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go index d60f9e58..3918bfd5 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go @@ -2,8 +2,6 @@ package main import ( "context" - "sync" - "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -16,9 +14,6 @@ import ( "mesh-delivery-pipeline/internal/pipeline" ) -// catalogRevisions is every watched branch's catalog-revision, `mesh.seat.mesh-catalog.event.catalog-revision.`. -const catalogRevisions = "mesh.seat." + catalog.Seat + ".event." + catalog.EventCatalogRevision + ".>" - // runnerViews are the runner's first views (design 47 G §21, §195, §200): the registrations, then the catalog half of // each watched branch they name, each said in `_first_views` (its state first-views) and, past five minutes // without a view, as view-waits in its findings. @@ -36,55 +31,17 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr return nil, err } status := process.AwaitBucket(ctx, js, conn, cred.Module+"_"+views.StatusState, "the runner's first views are not said in its first-view tool, nor asked again by hand", logf) - writer := func() views.FindingWriter { return views.WriterFindings{Writer: findings} } + writer := func() views.WireWriter { return findings } reader := &views.ProcessReader{Module: cred.Module, Call: call, Logf: logf, Hub: hub, Findings: writer} - // A watched branch first named by a registration gets its view of the catalog half (CT10), its conditions and its - // status kept as every first view's, asked at once. - wire := func(f *views.Follower) { - f.Logf = logf - f.Raise = func(s views.Status) { - putFinding(writer(), views.UnansweredFinding(cred.Module, s, time.Now()), true, logf) - } - f.Clear = func(s views.Status) { - putFinding(writer(), views.UnansweredFinding(cred.Module, s, time.Now()), false, logf) - } - f.RaiseWaits = func(s views.Status) { - logf("waits for %s's view %s since %s", s.Owner, s.WaitsFor, s.WaitingSince.Format(time.RFC3339)) - putFinding(writer(), views.ViewWaitsFinding(cred.Module, s, time.Now()), true, logf) - } - f.ClearWaits = func(s views.Status) { - putFinding(writer(), views.ViewWaitsFinding(cred.Module, s, time.Now()), false, logf) - } - f.Said = func(s views.Status) { - if kv := status.KV(); kv != nil { - views.KeepStatus(ctx, kv, logf)(s) - } - } - status.Ready(func(kv jetstream.KeyValue) { - if err := views.HearAgain(ctx, kv, []*views.Follower{f}, logf); err != nil { - logf("the hand asks of the first view %s are not heard (%v) until the runner starts again", f.Tool, err) - } - }) - } - var ( - mu sync.Mutex - made []*views.Follower - ) + // A watched branch first named by a registration gets its view of the catalog half (CT10), adopted through the reader (design + // 47 G §532, A-26): the reader keeps its status, raises and clears its findings through the module's writer, hears the + // seat's catalog-revision on the module's consumer and asks its first view at once. cv := &pipeline.Views{Logf: logf, Open: func(branch string) (pipeline.BranchView, error) { - c, err := meshcatalogclient.NewCatalog(call, branch) + c, _, err := meshcatalogclient.WireCatalogProcess(reader, call, branch, nil) if err != nil { return nil, err } - wire(c.Follower) - mu.Lock() - made = append(made, c.Follower) - mu.Unlock() - go func() { - if err := c.Start(ctx); err != nil { - logf("the first view of the catalog half of %s was not answered: %v", branch, err) - } - }() return catalogView{c}, nil }} regs, _, err := meshcatalogclient.WireRegistrationsProcess(reader, call, func(reg catalog.Registration, gone bool) { @@ -95,39 +52,10 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr if err != nil { return nil, err } - if err := hub.Handle(catalogRevisions, func(m jetstream.Msg) { - if err := cv.HearRevision(ctx, m.Data()); err != nil { - logf("%s is not taken: %v", m.Subject(), err) - } - _ = m.Ack() - }); err != nil { - return nil, err - } - // A consumer made after a view was asked does not deliver what was said before it: each branch's view asks again. - hub.OnBound(func(created time.Time) { - mu.Lock() - fs := append([]*views.Follower(nil), made...) - mu.Unlock() - for _, f := range fs { - f.ConsumerMade(ctx, created) - } - }) reader.Start(ctx, js, cred, status) return &runnerViews{Registrations: regs, Catalog: cv}, nil } -func putFinding(w views.FindingWriter, f views.Finding, raised bool, logf func(string, ...any)) { - var err error - if raised { - err = w.Put(f) - } else { - err = w.Delete(f) - } - if err != nil { - logf("the finding %s is not written: %v", f.Kind, err) - } -} - // catalogView is a branch's meshcatalogclient.Catalog as the runner reads it. type catalogView struct{ c *meshcatalogclient.Catalog } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index 9ac4db39..3342bd6b 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -108,7 +108,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("the teller's finding writer: %w", err) } defer writer.Close() - reader.Findings = func() views.FindingWriter { return views.WriterFindings{Writer: writer} } + reader.Findings = func() views.WireWriter { return writer } asks, err := reader.Follow(assignerModule, assignerAsksState, assignerModule+".assigner_read_asks") if err != nil { return err diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index b0181359..771b2a5d 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -129,7 +129,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("the view's finding writer: %w", err) } defer writer.Close() - reader.Findings = func() views.FindingWriter { return views.WriterFindings{Writer: writer} } + reader.Findings = func() views.WireWriter { return writer } for _, o := range ownersFollowed { if _, err := reader.Follow(o.owner, o.state, o.tool); err != nil { return err diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 135fd11e..3398eca1 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -263,7 +263,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer held.w.Close() reader := &views.ProcessReader{Module: watch.WatchmanModule, Call: calls.Call, Logf: logger.Printf, Hub: hub, - Findings: func() views.FindingWriter { return views.WriterFindings{Writer: held.w} }} + Findings: func() views.WireWriter { return held.w }} wants, err := reader.Follow(watch.AssignerModule, wantsState, watch.AssignerModule+".assigner_read_wants") if err != nil { return err