novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

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

Browse Source 
Builds: design 47 G §689, §687, A-22, A-26
Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-11 21:58:50 +01:00
parent 7310af36e6
commit 0fdf56f4bb
9 changed files with 36 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/findingwriter.go
+5 -10
View File
@@ -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)
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go
+2 -2
View File
@@ -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
}
modules/mesh-delivery-checker/internal/check/checker.go
+1 -2
View File
@@ -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).
modules/mesh-delivery-checker/internal/check/checker_test.go
+3 -4
View File
@@ -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
modules/mesh-delivery-checker/internal/check/couldnotrun.go
+17 -5
View File
@@ -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),
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go
+5 -77
View File
@@ -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.<branch>`.
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 `<module>_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 }
modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go
+1 -1
View File
@@ -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
modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go
+1 -1
View File
@@ -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
modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go
+1 -1
View File
@@ -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