diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go index fc598c633..309d4248a 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go @@ -29,12 +29,12 @@ import ( "sync" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" artifactref "git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient/reference" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -45,8 +45,6 @@ const ( HavesRead = "node-engine.have" // ManifestsRead is the indexer's manifests: the images a manifest pins by itself. ManifestsRead = indexerModule + "." + catalog.StateManifests - // FindingsState is the holder's own findings (meshconditions.StateFindings): image-missing. - FindingsState = meshconditions.StateFindings // Module is this module, the observer its findings name. Module = "artifact-store-tools" ) @@ -83,8 +81,10 @@ type Door interface { // Holder is the seat's holder. type Holder struct { - Bus Bus - Door func() (Door, error) + Bus Bus + // Findings is the module's finding writer: S25's and the sweep's findings go through it. + Findings Findings + Door func() (Door, error) // Listed is the store as its seats list it now (Store.View); nil reads none, and then nothing is said gone or collected. Listed func(ctx context.Context) (*View, error) Acts meshhandactsclient.Recorder @@ -95,6 +95,7 @@ type Holder struct { sweep sync.Mutex // one sweep at a time mu sync.Mutex + fmu sync.Mutex // one pass of Observe at a time ended map[string]time.Time // a done key's end, read once from its history } @@ -852,16 +853,60 @@ func inStoreAt(image, address string) string { return r } -// Observe writes S25's findings (image-missing): each want whose build names an image the store does not hold, on the -// want's machine, and deletes each finding that no longer holds. Nothing is said without a listing. +// Observe says S25's findings (image-missing): each want whose build names an image the store does not hold, on the +// want's machine, written once and cleared once it no longer holds. Nothing is said without a listing. What stands is what +// the held writer holds (design 47 G A-32, A-38), so a restart neither drops nor rewrites it. func (h *Holder) Observe(ctx context.Context) error { - j, err := h.judge(ctx) + now, err := h.observed(ctx) if err != nil { return err } - if j.view == nil { + if now == nil { return nil } + h.fmu.Lock() + defer h.fmu.Unlock() + said := map[string]meshconditions.Finding{} + for _, f := range h.Findings.Held() { + if f.Kind == ImageMissing { + said[views.FindingKey(f.Scope, f.ID, f.Kind)] = f + } + } + for k, was := range said { + if _, still := now[k]; !still { + if err := h.Findings.Clear(ctx, was.Scope, was.ID, was.Kind, "the store holds the image again"); err != nil { + return err + } + } + } + for k, f := range now { + if was, ok := said[k]; ok && was.Summary == f.Summary { + continue // written only when what it says changes + } + if err := h.Findings.Write(ctx, f); err != nil { + return err + } + } + return nil +} + +// Findings is the module's held finding writer (views.HeldWriter): the holder writes and clears through it and reads what it +// holds (design 47 G §689, A-29, A-38). +type Findings interface { + Write(ctx context.Context, finding meshconditions.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error + Held() []meshconditions.Finding +} + +// observed is S25's findings now by key, nil without a listing. +func (h *Holder) observed(ctx context.Context) (map[string]meshconditions.Finding, error) { + j, err := h.judge(ctx) + if err != nil { + return nil, err + } + if j.view == nil { + return nil, nil + } address := "" if d, err := h.Door(); err == nil { address = d.Address() @@ -870,7 +915,7 @@ func (h *Holder) Observe(ctx context.Context) error { for _, b := range j.built { builds[b.Module+" "+b.FP] = b } - now := map[string]meshconditions.FindingsByKeyValue{} + now := map[string]meshconditions.Finding{} for _, w := range j.wants { b, ok := builds[w.Module+" "+w.FP] if !ok { @@ -895,63 +940,26 @@ func (h *Holder) Observe(ctx context.Context) error { continue } f := imageMissing(w.Node, w.Module, reference, h.now()) - now[f.Key] = f + now[views.FindingKey(f.Scope, f.ID, f.Kind)] = f break } } - keys, err := h.Bus.Keys(FindingsState) - if err != nil { - return fmt.Errorf("the findings: %w", err) - } - for _, k := range keys { - if k == sweepFindingKey { - continue // the sweep's own finding, which the sweep clears - } - if _, still := now[k]; !still { - // A finding of a first view unanswered is the reader's (views.Reader), cleared by its answer, not by a sweep. - if raw, found, err := h.Bus.Get(FindingsState, k); err == nil && found { - if f, err := meshconditions.DecodeFindingsByKeyValue(raw); err == nil && f.Kind == views.UnansweredKind { - continue - } - } - if err := h.Bus.Delete(FindingsState, k); err != nil { - return err - } - } - } - for k, f := range now { - raw, found, err := h.Bus.Get(FindingsState, k) - if err != nil { - return err - } - if found { - if was, err := meshconditions.DecodeFindingsByKeyValue(raw); err == nil && was.Summary == f.Summary { - continue // written only when what it says changes - } - } - if err := meshconditionsclient.CheckFinding(f); err != nil { - return err - } - if err := h.Bus.Put(FindingsState, k, f); err != nil { - return err - } - } - return nil + return now, nil } -// imageMissing is S25's finding, in the controller's words (cmd/mesh-controller/image_missing.go watchMissingImages). -func imageMissing(node, module, image string, at time.Time) meshconditions.FindingsByKeyValue { - id := module + "." + node - return meshconditions.FindingsByKeyValue{V: meshconditionsclient.FindingVersion, Key: meshconditionsclient.Key(meshconditionsclient.ScopeModule, id, ImageMissing), Kind: ImageMissing, - Subject: meshconditions.Subject{Scope: meshconditionsclient.ScopeModule, ID: id, Machine: node}, - Severity: meshconditionsclient.Warning, Resolver: meshconditionsclient.ResolverOperator, +// imageMissing is S25's finding, in the controller's words (cmd/mesh-controller/image_missing.go watchMissingImages); its id +// is the module's and the machine's, each escaped (design 47 G A-14). +func imageMissing(node, module, image string, at time.Time) meshconditions.Finding { + id := keys.EscapeName(module) + "." + keys.EscapeName(node) + return meshconditions.Finding{Scope: "module", ID: id, Kind: ImageMissing, + Subject: meshconditions.Subject{Scope: "module", ID: id, Machine: node}, + Severity: meshconditions.FindingSeverityWarning, Resolver: "operator", Summary: fmt.Sprintf("%s is left out of every send to %s: its declaration names an image that the artifact store does "+ "not hold (the evidence names it); the running container is untouched", module, node), Headline: capital(module) + " is not updated on " + node, Explanation: fmt.Sprintf("The new version of %s names a container image that the mesh's store does not hold, so it "+ "was not sent to %s. What runs there is untouched, and the rest of %s is sent as usual.", module, node, node), Needs: fmt.Sprintf("build %s again; the details name the image.", module), - Resolved: "Resolved: the store holds the image of " + module + " again", Evidence: []meshconditions.Evidence{{At: at.UTC(), Said: "image " + image + " is not in the store"}}, Source: "S25", Observer: Module, At: at.UTC()} } diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/holder_integration_test.go b/modules/artifact-store-tools/cmd/artifact-store-tools/holder_integration_test.go index 575dfcc00..3963d3749 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/holder_integration_test.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder_integration_test.go @@ -198,7 +198,7 @@ func TestTheSeatSweepsAtABuildsEndOnARealBusWithNoLease(t *testing.T) { } return m.Data, nil } - h := &Holder{Bus: findingsAt{realBus: bus}, Door: mem.Door, Listed: mem.Listed, Seats: ask, Now: time.Now, Logf: t.Logf} + h := &Holder{Bus: findingsAt{realBus: bus}, Findings: findingsThrough{findingsAt{realBus: bus}}, Door: mem.Door, Listed: mem.Listed, Seats: ask, Now: time.Now, Logf: t.Logf} h.Acts = meshhandactsclient.Recorder{Ask: ask} // A build's end heard from the builds bucket, as follow hands it over. diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go b/modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go index 6aa10a50f..0c1a05d98 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go @@ -17,10 +17,10 @@ import ( "testing" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" artifactref "git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient/reference" reg "git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry" @@ -229,7 +229,7 @@ func aMesh(t *testing.T) (*Holder, *memBus, *memDoor, *[]string) { want, _ := json.Marshal(meshwants.WantsByWantNodeModuleValue{V: 1, Fp: fp(1), Self: true}) _ = b.Put(WantsRead, "want.ace.letta", json.RawMessage(want)) var acts []string - h := &Holder{Bus: b, Door: func() (Door, error) { return door, nil }, Seats: door.seat, + h := &Holder{Bus: b, Findings: findingsThrough{b}, Door: func() (Door, error) { return door, nil }, Seats: door.seat, Listed: func(context.Context) (*View, error) { return &View{L: listing}, nil }, Now: func() time.Time { return day.Add(48 * time.Hour) }, Acts: meshhandactsclient.Recorder{Ask: func(key string, body any) (json.RawMessage, error) { @@ -370,7 +370,7 @@ func TestTheSeatKeepsTheFiveNewestAndTheWantedAndSweepsWithNoLease(t *testing.T) if c.letGo[0] == archiveRef("letta", 2) && !slices.Contains(door.calls, "seat:oci-artifact-registry.remove letta/files@"+digestOf(archiveRef("letta", 2))+" "+Retention) { t.Errorf("the seats were called %v", door.calls) } - _, finding, _ := b.Get(FindingsState, sweepFindingKey) + _, finding, _ := b.Get(FindingsState, views.FindingKey("module", Module, SweepRefused)) if finding != (c.refuse != "") { t.Errorf("the sweep's finding is written: %v, want %v", finding, c.refuse != "") } @@ -379,7 +379,7 @@ func TestTheSeatKeepsTheFiveNewestAndTheWantedAndSweepsWithNoLease(t *testing.T) if _, err := h.Sweep(context.Background(), Sweeping{Most: MostPerSweep, By: "again"}); err != nil { t.Fatal(err) } - if _, still, _ := b.Get(FindingsState, sweepFindingKey); still { + if _, still, _ := b.Get(FindingsState, views.FindingKey("module", Module, SweepRefused)); still { t.Error("a clean sweep left the sweep's finding") } return @@ -410,13 +410,16 @@ func TestAWantedImageTheStoreLacksIsAFindingAndImagesNamesWantsAndHaves(t *testi if err := h.Observe(context.Background()); err != nil { t.Fatal(err) } - key := meshconditionsclient.Key(meshconditionsclient.ScopeModule, "letta.ace", ImageMissing) + key := views.FindingKey("module", "letta.ace", ImageMissing) raw, found, _ := b.Get(FindingsState, key) - f, err := meshconditions.DecodeFindingsByKeyValue(raw) + var f meshconditions.Finding + err := json.Unmarshal(raw, &f) if !found || err != nil || f.Subject.Machine != "ace" || f.Source != "S25" || !strings.Contains(f.Evidence[0].Said, imageRef("letta", 1)) { t.Fatalf("the finding is %+v (%v, %v)", f, found, err) } view.L.Repos[repo].Revisions[digest] = true + // A restart: a holder made afresh holds no finding of its own, and clears what the writer holds (design 47 G A-32, A-38). + h = &Holder{Bus: h.Bus, Findings: h.Findings, Door: h.Door, Listed: h.Listed, Seats: h.Seats, Now: h.Now, Acts: h.Acts, Logf: h.Logf} if err := h.Observe(context.Background()); err != nil { t.Fatal(err) } @@ -441,3 +444,41 @@ func TestAWantedImageTheStoreLacksIsAFindingAndImagesNamesWantsAndHaves(t *testi t.Fatalf("the images of ace are %+v: %v", images, err) } } + +// FindingsState is the module's state `findings`, where the tests' writer keeps what the holder writes through it. +const FindingsState = "findings" + +// findingsThrough is the holder's finding writer in a test: it keeps a finding in its key and deletes it as the module's +// writer does (views.Writer), without the bus. +type findingsThrough struct { + bus interface { + Put(state, key string, value any) error + Delete(state, key string) error + } +} + +func (f findingsThrough) Write(_ context.Context, finding meshconditions.Finding) error { + return f.bus.Put(FindingsState, views.FindingKey(finding.Scope, finding.ID, finding.Kind), finding) +} + +// Held is what the writer holds: the findings in the module's state. +func (f findingsThrough) Held() []meshconditions.Finding { + var out []meshconditions.Finding + keys, _ := f.bus.(interface { + Keys(string) ([]string, error) + }).Keys(FindingsState) + for _, k := range keys { + raw, ok, _ := f.bus.(interface { + Get(string, string) ([]byte, bool, error) + }).Get(FindingsState, k) + var m meshconditions.Finding + if ok && json.Unmarshal(raw, &m) == nil { + out = append(out, m) + } + } + return out +} + +func (f findingsThrough) Clear(_ context.Context, scope, id, kind, _ string) error { + return f.bus.Delete(FindingsState, views.FindingKey(scope, id, kind)) +} diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/main.go b/modules/artifact-store-tools/cmd/artifact-store-tools/main.go index 3af7eacc5..acb1d80d5 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/main.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/main.go @@ -13,8 +13,8 @@ import ( "os" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/views" @@ -24,12 +24,29 @@ func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) } func main() { s := &Store{Ask: stdio.Ask} - v := &views.Reader{Module: "artifact-store-tools", Logf: logf, Findings: meshconditionsclient.FindingWriter{State: stdio.State("findings")}, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}} + v := &views.Reader{Module: "artifact-store-tools", Logf: logf, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}} v.Follow(assignerModule, "wants", WantsTool) v.Follow(indexerModule, "manifests", ManifestsTool) h := &Holder{Bus: runtimeBus{views: v}, Door: DoorAt(os.Getenv("MESH_ARTIFACT_STORE_ADDRESS_FILE"), s), Listed: s.View, Seats: stdio.Ask, Now: time.Now, Logf: logf} h.Acts = meshhandactsclient.Recorder{Ask: stdio.Ask} + // The module's one held writer (design 47 G §689, A-16, A-29, A-32): the holder's findings and the reader's go through it. + w, err := views.NewStdioHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: Module, State: stdio.State("findings"), Kinds: []string{ImageMissing, SweepRefused}}) + if err != nil { + logf("%v", err) + os.Exit(1) + } + h.Findings = w + v.Findings = w + // the runtime answers the writer's subscription only once Serve has its channel + go func() { + <-stdio.Served() + if err := w.Start(context.Background()); err != nil { + logf("%v", err) + os.Exit(1) + } + }() v.Start(context.Background()) go follow(h, v) // An empty name serves as the module the runtime names (MESH_SERVED_MODULE): artifact-store-tools. diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/sweepseats.go b/modules/artifact-store-tools/cmd/artifact-store-tools/sweepseats.go index 568b62781..b106f1cf7 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/sweepseats.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/sweepseats.go @@ -10,15 +10,16 @@ package main // sweep's finding, sweep-refused; the next sweep with no refusal clears it. import ( + "context" "encoding/json" "fmt" "strings" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" artifactref "git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient/reference" ociseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/ociartifactregistry" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/keys" reg "git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry" ) @@ -26,9 +27,6 @@ import ( // SweepRefused is the sweep's finding: a seat refused a hold or a remove the sweep asked. const SweepRefused = "sweep-refused" -// sweepFindingKey is the one key of the sweep's finding in the module's findings. -var sweepFindingKey = meshconditionsclient.Key(meshconditionsclient.ScopeModule, Module, SweepRefused) - // canonical is a kept reference in the older form, `artifact-store://`, whichever form it came in; another is as // it was. func canonical(reference string) string { @@ -115,26 +113,28 @@ func (h *Holder) removeOnSeat(reference, why string) error { // sweepFinding writes the sweep's finding: what the seat refused, and why. func (h *Holder) sweepFinding(what string, err error) { at := h.now().UTC() - f := meshconditions.FindingsByKeyValue{V: meshconditionsclient.FindingVersion, Key: sweepFindingKey, Kind: SweepRefused, - Subject: meshconditions.Subject{Scope: meshconditionsclient.ScopeModule, ID: Module}, - Severity: meshconditionsclient.Warning, Resolver: meshconditionsclient.ResolverOperator, + f := meshconditions.Finding{Scope: "module", ID: keys.EscapeName(Module), Kind: SweepRefused, + Subject: meshconditions.Subject{Scope: "module", ID: keys.EscapeName(Module)}, + Severity: meshconditions.FindingSeverityWarning, Resolver: "operator", Summary: fmt.Sprintf("the artifact sweep stopped: a registry seat refused %s (the evidence names why)", what), Headline: "The artifact sweep stopped", Explanation: "The sweep keeps what the mesh still needs and removes the rest through the registry seats. A seat " + "refused one of its asks, so it stopped before anything more was removed; what it kept stays kept.", Needs: "read the seat holder's journal for the refusal; the next sweep asks again.", - Resolved: "Resolved: the artifact sweep ran through without a refusal", Evidence: []meshconditions.Evidence{{At: at.UTC(), Said: fmt.Sprintf("%s: %v", what, err)}}, Source: "CAT-AS1", Observer: Module, At: at} - if werr := h.Bus.Put(FindingsState, sweepFindingKey, f); werr != nil { + if werr := h.Findings.Write(context.Background(), f); werr != nil { h.logf("the sweep's finding (%s: %v) could not be written: %v", what, err, werr) } } -// sweepClean clears the sweep's finding after a sweep no seat refused. +// sweepClean clears the sweep's finding, when the held writer holds it, after a sweep no seat refused. func (h *Holder) sweepClean() { - if _, found, err := h.Bus.Get(FindingsState, sweepFindingKey); err == nil && found { - if err := h.Bus.Delete(FindingsState, sweepFindingKey); err != nil { + for _, was := range h.Findings.Held() { + if was.Kind != SweepRefused { + continue + } + if err := h.Findings.Clear(context.Background(), was.Scope, was.ID, was.Kind, "a sweep ran through without a refusal"); err != nil { h.logf("the sweep's finding could not be cleared: %v", err) } } diff --git a/modules/artifact-store-tools/go.mod b/modules/artifact-store-tools/go.mod index 2e3263e2d..8b71604f0 100644 --- a/modules/artifact-store-tools/go.mod +++ b/modules/artifact-store-tools/go.mod @@ -5,7 +5,6 @@ go 1.26.0 require ( git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 diff --git a/modules/claude-code/cmd/claude-code/config_test.go b/modules/claude-code/cmd/claude-code/config_test.go index 5cdbc2238..948fb3c9b 100644 --- a/modules/claude-code/cmd/claude-code/config_test.go +++ b/modules/claude-code/cmd/claude-code/config_test.go @@ -279,7 +279,7 @@ func TestTheManifestListsEveryToolServed(t *testing.T) { // the controller, and no read of the indexer's or the assigner's state (ADR 0615 §1): their views are consumed. if !slices.Equal(m.Invokes, []string{"seat:mesh-hand-acts.warranted", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.seat-definitions", ModulesTool, ManifestsTool, AssignmentsTool}) || !slices.Equal(m.Reads, []string{"claude-licence-manager.bindings"}) || - !slices.Equal(m.Consumes, []string{"mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests", "mesh-delivery-assigner.views.assignments", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed"}) { + !slices.Equal(m.Consumes, []string{"mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests", "mesh-delivery-assigner.views.assignments", "mesh-catalog.seat-definition-changed", "mesh-catalog.seat-definition-removed", "mesh-conditions.findings-asked"}) { t.Errorf("invokes = %v, reads = %v, consumes = %v", m.Invokes, m.Reads, m.Consumes) } if src, _ := os.ReadFile("proposals.go"); !strings.Contains(string(src), `"seat:mesh-hand-acts.warranted"`) || diff --git a/modules/claude-code/cmd/claude-code/main.go b/modules/claude-code/cmd/claude-code/main.go index 53a18a1a0..df34d25ac 100644 --- a/modules/claude-code/cmd/claude-code/main.go +++ b/modules/claude-code/cmd/claude-code/main.go @@ -22,7 +22,7 @@ import ( "strings" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + condseat "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" stdio "git.novox.be/novox/mesh-sdk/go" @@ -230,12 +230,22 @@ const ( // once of the owner's read tool, then followed on the owner's views family. It reads none of their state. var meshViews *views.Reader +// newFindingsWriter is this module's finding writer (design 47 G §689, A-16): it states the findings of its views reader and +// observes none of its own. +func newFindingsWriter() (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) { + return views.NewStdioWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ + Asked: condseat.SubjectFindingsAsked, Module: "claude-code", State: stdio.State("findings"), + Observe: func(context.Context) (views.Observed[condseat.Finding], error) { + return views.Observed[condseat.Finding]{}, nil + }}) +} + // meshChanged is told, without waiting, that a followed state changed: the instructions' table is read again. var meshChanged = make(chan struct{}, 1) // newMeshViews builds the reader of those three states; Start asks them. func newMeshViews() *views.Reader { - v := &views.Reader{Module: Module, Logf: func(f string, a ...any) { say(f, a...) }, Findings: meshconditionsclient.FindingWriter{State: stdio.State("findings")}, + v := &views.Reader{Module: Module, Logf: func(f string, a ...any) { say(f, a...) }, Acts: handActs{}, Changed: func(string, string, json.RawMessage, bool) { select { @@ -488,6 +498,20 @@ func main() { } } } + findings, err := newFindingsWriter() + if err != nil { + say("%v", err) + os.Exit(1) + } + meshViews.Findings = findings + // the runtime answers the writer's subscription only once Serve has its channel + go func() { + <-stdio.Served() + if err := findings.Start(context.Background()); err != nil { + say("%v", err) + os.Exit(1) + } + }() meshViews.Start(context.Background()) go refreshMeshTools(p) servers := stateOf{stdio.State("servers")} diff --git a/modules/claude-code/go.mod b/modules/claude-code/go.mod index 7b9a5cfc8..9c9f3a901 100644 --- a/modules/claude-code/go.mod +++ b/modules/claude-code/go.mod @@ -7,7 +7,6 @@ require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000 @@ -26,6 +25,6 @@ require ( ) require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect ) diff --git a/modules/claude-code/module.json b/modules/claude-code/module.json index 7bfbd71fd..5a0b83f2f 100644 --- a/modules/claude-code/module.json +++ b/modules/claude-code/module.json @@ -11,6 +11,20 @@ { "name": "gitea-read", "optional": true + }, + { + "name": "finding-events", + "storage": "stream", + "template": "events", + "options": { + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } } ], "binds": { @@ -35,7 +49,8 @@ "mesh-delivery-indexer.views.manifests", "mesh-delivery-assigner.views.assignments", "mesh-catalog.seat-definition-changed", - "mesh-catalog.seat-definition-removed" + "mesh-catalog.seat-definition-removed", + "mesh-conditions.findings-asked" ], "state": [ "servers", diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index 41c949a4a..352dea76f 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -18,12 +18,12 @@ import ( "sync" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" reg "git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry" "git.novox.be/novox/mesh-delivery/sdk/parts" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" "distribution/internal/holder" ) @@ -67,21 +67,6 @@ func say(seat, event string, body holder.Event) error { return stdio.SeatPublish("mesh.seat."+seat+".event."+event, eventBody(event, body), body.ID) } -func finding(kind, subject, words string, raised bool) { - key := meshconditionsclient.Key(meshconditionsclient.ScopeSeat, "distribution", kind+"."+strings.NewReplacer(".", "_", "@", "_", "/", "_").Replace(subject)) - if !raised { - _ = stdio.State(meshconditions.StateFindings).Delete(key) - return - } - f := meshconditions.FindingsByKeyValue{V: meshconditionsclient.FindingVersion, Key: key, Kind: kind, Ref: key[strings.LastIndex(key, ".")+1:], - Subject: meshconditions.Subject{Scope: meshconditionsclient.ScopeSeat, ID: "distribution"}, Severity: meshconditionsclient.Warning, - Summary: words, Headline: words, Explanation: words, Resolved: "cleared once it no longer holds", - Source: kind, Observer: "distribution", At: time.Now().UTC()} - if _, err := stdio.State(meshconditions.StateFindings).Put(key, f); err != nil { - logf("the finding %s could not be written: %v", key, err) - } -} - // learned are the proved callers whose build this process recorded as publishing through the seats. var learned sync.Map @@ -420,7 +405,7 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder. Input: obj(map[string]any{}), Run: func(map[string]any) (any, error) { front, _ := url.Parse("http://127.0.0.1:5100") - return holder.Health(&holder.OCI{Base: front}, finding) + return holder.Health(&holder.OCI{Base: front}, findings.Say) }}, stdio.Tool{Name: "registry_incoming", Description: "Open transfers of incoming and its fill; with drop, drops one by hand (a hand act, why required).", Input: obj(map[string]any{"drop": s("a transfer id to drop"), "why": s("why it is dropped")}), @@ -516,10 +501,34 @@ func tools(v *holder.Verbs, windows kept[holder.Window], callers func() (holder. return out } +// findings keeps the open findings of this bundle for its writer (design 47 G §689). +var findings = &holder.Findings{Logf: logf} + func main() { + // The tools bundle's findings go through the module's held writer (design 47 G §689, A-16, A-29, A-32); it owns the transfers' + // kinds, the program's writer the driver's. + w, err := views.NewStdioHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: holder.Self, State: stdio.State("findings"), Kinds: holder.TransferFindingKinds, Logf: logf}) + if err != nil { + logf("%v", err) + os.Exit(1) + } + findings.Writer = w + // the runtime answers the writer's state and subscription only once Serve has its channel + go func() { + <-stdio.Served() + // the keys of the older form end in no kind the writer owns, so it does not clear them (A-31) + if err := holder.DeleteOldFindings(stdio.State("findings"), holder.Self, holder.TransferFindingKinds); err != nil { + logf("the old findings are not deleted: %v", err) + } + if err := w.Start(context.Background()); err != nil { + logf("%v", err) + os.Exit(1) + } + }() base, _ := url.Parse("http://127.0.0.1:5001") store := &holder.OCI{Base: base, Upstream: &holder.Upstream{}} - h := &holder.Holder{Objects: parts.BusObjects(holder.IncomingObjects), Builder: holder.Reserved{Ask: stdio.Ask}, Store: store, Say: say, Finding: finding, + h := &holder.Holder{Objects: parts.BusObjects(holder.IncomingObjects), Builder: holder.Reserved{Ask: stdio.Ask}, Store: store, Say: say, Finding: findings.Say, Transfers: parts.BusTransfers[holder.Opening](holder.TransfersState), Cap: parts.BusCap(holder.IncomingObjects), Record: recordAct, Idle: parts.BusArm(holder.IdleState)} files := &holder.Files{Container: or(os.Getenv("MESH_REGISTRY_CONTAINER"), "mesh-registry"), Run: holder.ExecRunner, UID: os.Getuid()} v := &holder.Verbs{Holder: h, Catalog: store, Holds: kept[holder.Hold]{state: "holds", key: holder.Hold.Key}, Mirrors: mirrors{}, diff --git a/modules/distribution/cmd/distribution/main.go b/modules/distribution/cmd/distribution/main.go index 8db8da6a4..dfc91918b 100644 --- a/modules/distribution/cmd/distribution/main.go +++ b/modules/distribution/cmd/distribution/main.go @@ -20,7 +20,9 @@ import ( "syscall" "time" + meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" "github.com/nats-io/nats.go/jetstream" "distribution/internal/holder" @@ -62,9 +64,26 @@ func run(ctx context.Context, getenv func(string) string) error { return err } defer calls.Close() - findings := process.OptionalBucket(ctx, js, "distribution_findings", "a window held for seven days is said in the log only", logger.Printf) + // The program's findings go through the module's writer (design 47 G §689); it owns the driver's kinds, the tools + // bundle's writer the transfers'. + findings := &holder.Findings{Logf: logger.Printf} + w, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: holder.Self, Node: cred.Node, State: calls.State("findings"), Kinds: holder.DriverFindingKinds, + JS: js, Conn: conn, Logf: logger.Printf}) + if err != nil { + return err + } + findings.Writer = w + // the keys of the older form end in no kind the writer owns, so it does not clear them (A-31) + if err := holder.DeleteOldFindings(calls.State("findings"), holder.Self, holder.DriverFindingKinds); err != nil { + logger.Printf("the old findings are not deleted: %v", err) + } + if err := w.Start(ctx); err != nil { + return err + } + defer w.Close() go drive(ctx, js, cred, &holder.Driver{Delivery: busDelivery{calls}, State: kvWindows{windows}, Callers: busCallers{calls}.Read, - Closed: sayClosed(js, cred), Finding: findingsKV(findings, logger.Printf)}, logger) + Closed: sayClosed(js, cred), Finding: findings.Say}, logger) } front := &holder.Front{Store: store, Window: &open, Logf: logger.Printf, Published: func(name, tag string) (bool, error) { diff --git a/modules/distribution/cmd/distribution/whole_integration_test.go b/modules/distribution/cmd/distribution/whole_integration_test.go index c60bb57bc..e0ee7b7da 100644 --- a/modules/distribution/cmd/distribution/whole_integration_test.go +++ b/modules/distribution/cmd/distribution/whole_integration_test.go @@ -210,9 +210,12 @@ func TestDistributionWhole(t *testing.T) { i, err := c.Info(ctx) return err == nil && i.NumPending == 0 && i.NumAckPending == 0 && i.Delivered.Consumer >= 2 }) - info, _ := s.Info(ctx) - if info.State.Msgs != 2 { - t.Errorf("distribution.events holds %d events, want old-name-closed and direct-writes-closed once each", info.State.Msgs) + // The stream holds the module's own events once each; the finding writer's (design 47 G A-30) are said beside them. + info, _ := s.Info(ctx, jetstream.WithSubjectFilter("mesh.stream.distribution.event.>")) + for _, ev := range []string{holder.OldNameClosedEvent, holder.DirectWritesClosedEvent} { + if n := info.State.Subjects["mesh.stream.distribution.event."+ev]; n != 1 { + t.Errorf("distribution.events holds %s %d times, want once", ev, n) + } } for _, ev := range []string{holder.OldNameClosedEvent, holder.DirectWritesClosedEvent} { if _, err := s.GetLastMsgForSubject(ctx, "mesh.stream.distribution.event."+ev); err != nil { diff --git a/modules/distribution/cmd/distribution/windows.go b/modules/distribution/cmd/distribution/windows.go index 15b166776..e7a06e185 100644 --- a/modules/distribution/cmd/distribution/windows.go +++ b/modules/distribution/cmd/distribution/windows.go @@ -18,8 +18,6 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" - meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/process" "distribution/internal/holder" @@ -173,30 +171,6 @@ func (k kvWindows) Publishing() (holder.Publishing, error) { return p, k.get(holder.PublishingKey, &p) } -// findingsKV writes the driver's findings into distribution's state findings, as the tools bundle writes its own. -func findingsKV(kv jetstream.KeyValue, logf func(string, ...any)) func(kind, subject, words string, raised bool) { - return func(kind, subject, words string, raised bool) { - if kv == nil { - return - } - key := meshconditionsclient.Key(meshconditionsclient.ScopeSeat, holder.Self, kind+"."+strings.NewReplacer(".", "_", "@", "_", "/", "_").Replace(subject)) - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - defer cancel() - if !raised { - _ = kv.Delete(ctx, key) - return - } - f := meshconditions.FindingsByKeyValue{V: meshconditionsclient.FindingVersion, Key: key, Kind: kind, Ref: key[strings.LastIndex(key, ".")+1:], - Subject: meshconditions.Subject{Scope: meshconditionsclient.ScopeSeat, ID: holder.Self}, Severity: meshconditionsclient.Warning, - Summary: words, Headline: words, Explanation: words + "; registry_old_name or registry_direct_writes drops a retired node by hand", - Resolved: "cleared when the window closes", Source: kind, Observer: holder.Self, At: time.Now().UTC()} - b, _ := json.Marshal(f) - if _, err := kv.Put(ctx, key, b); err != nil { - logf("the finding %s could not be written: %v", key, err) - } - } -} - // sayClosed says a window's closing on distribution's own family distribution.events, once (its id is the window's). func sayClosed(js jetstream.JetStream, cred process.Account) func(holder.Closed) error { return func(c holder.Closed) error { diff --git a/modules/distribution/go.mod b/modules/distribution/go.mod index 4fbf8a54a..d5b6b1f8d 100644 --- a/modules/distribution/go.mod +++ b/modules/distribution/go.mod @@ -3,7 +3,6 @@ module distribution go 1.26.0 require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient v0.0.0-00010101000000-000000000000 diff --git a/modules/distribution/internal/holder/findings.go b/modules/distribution/internal/holder/findings.go new file mode 100644 index 000000000..e14a07b84 --- /dev/null +++ b/modules/distribution/internal/holder/findings.go @@ -0,0 +1,80 @@ +package holder + +import ( + "context" + "strings" + "time" + + meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + keyparts "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// FindingKinds are the kinds the driver says (the program) and the transfers say (the tools bundle): each bundle's writer +// owns its own, since the two share one state `findings` (design 47 G §580, §689). +var ( + DriverFindingKinds = []string{FindingWindowHeld} + TransferFindingKinds = []string{"publish-refused", "incoming-full"} +) + +// FindingWriter is the module's writer (views.Writer) as the keeper writes and clears through it. +type FindingWriter interface { + Write(ctx context.Context, finding meshconditions.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error +} + +// Findings keeps a bundle's open findings and says them through the module's writer (design 47 G §689, A-14): its `Say` is +// the Finding callback of the Driver and the Holder; the writer is the held writer (views.HeldWriter), which holds what is open (A-29, A-32). +type Findings struct { + Writer FindingWriter + Logf func(string, ...any) + Now func() time.Time +} + +// Say raises (raised) or clears the finding of a kind about a subject. Its id is the seat's and the subject's, each escaped +// as a key's name is (design 47 G A-14); the kind stays the bundle's own word. +func (k *Findings) Say(kind, subject, words string, raised bool) { + id := keyparts.EscapeName(Self) + "." + keyparts.EscapeName(subject) + key := views.FindingKey("seat", id, kind) + if !raised { + if err := k.Writer.Clear(context.Background(), "seat", id, kind, "it no longer holds"); err != nil { + k.Logf("the finding %s could not be cleared: %v", key, err) + } + return + } + at := time.Now() + if k.Now != nil { + at = k.Now() + } + f := meshconditions.Finding{Scope: "seat", ID: id, Kind: kind, Subject: meshconditions.Subject{Scope: "seat", ID: id}, + Severity: meshconditions.FindingSeverityWarning, Summary: words, Headline: words, Explanation: words, + Source: kind, Observer: Self, At: at.UTC()} + if err := k.Writer.Write(context.Background(), f); err != nil { + k.Logf("the finding %s could not be written: %v", key, err) + } +} + +// DeleteOldFindings deletes, once at a module's start, the findings the keys of an older form hold (design 47 G A-31): the form +// `seat.._`, whose kind is no kind the writer owns in its key, so the writer's own first statement does not +// clear it. Only a key of one of kinds is deleted; the keys of the writer's form (`seat...`) are left. +func DeleteOldFindings(state views.FindingState, seat string, kinds []string) error { + keys, err := state.Keys() + if err != nil { + return err + } + for _, k := range keys { + rest, ok := strings.CutPrefix(k, "seat."+seat+".") + if !ok { + continue + } + for _, kind := range kinds { + if strings.HasPrefix(rest, kind+"_") { + if err := state.Delete(k); err != nil { + return err + } + break + } + } + } + return nil +} diff --git a/modules/distribution/internal/holder/findings_test.go b/modules/distribution/internal/holder/findings_test.go new file mode 100644 index 000000000..677910d1b --- /dev/null +++ b/modules/distribution/internal/holder/findings_test.go @@ -0,0 +1,44 @@ +package holder + +import ( + "sort" + "strings" + "testing" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// A-31: the keys of the older form (`seat.._`) are deleted once; a key of the writer's form, another seat's +// and another kind's stay. +func TestTheOldFindingKeysAreDeletedOnce(t *testing.T) { + state := &oldKeysState{keys: map[string]bool{}} + for _, k := range []string{ + "seat.distribution.publish-refused_write", "seat.distribution.publish-refused_a_b_c", + "seat.distribution.sub.publish-refused", "seat.other.publish-refused_write", "seat.distribution.unrelated_write"} { + state.keys[k] = true + } + if err := DeleteOldFindings(state, "distribution", []string{"publish-refused", "incoming-full"}); err != nil { + t.Fatal(err) + } + var left []string + for k := range state.keys { + left = append(left, k) + } + sort.Strings(left) + if got, want := strings.Join(left, " "), "seat.distribution.sub.publish-refused seat.distribution.unrelated_write seat.other.publish-refused_write"; got != want { + t.Errorf("left %s, want %s", got, want) + } +} + +type oldKeysState struct{ keys map[string]bool } + +func (s *oldKeysState) Get(string) (*stdio.StateEntry, error) { return nil, nil } +func (s *oldKeysState) Put(string, any) (uint64, error) { return 0, nil } +func (s *oldKeysState) Delete(k string) error { delete(s.keys, k); return nil } +func (s *oldKeysState) Keys() ([]string, error) { + var out []string + for k := range s.keys { + out = append(out, k) + } + return out, nil +} diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index c428e942b..aaf559ff6 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/modules/distribution/internal/holder/holder_test.go @@ -243,8 +243,13 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { if a, _ := b.h.Publish(ArtifactSeat, builder, req("t1")); !reflect.DeepEqual(a.Missing, []string{part.Digest}) { t.Errorf("resumed publish = %+v", a) } - if _, err := b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t1", Digest: part.Digest, Sequence: 0, Data: blob[:1000]}); err == nil { - t.Error("a chunk sent twice was taken: the transfer goes on from the chunks held") + // A chunk 0 again starts the part afresh (it travels whole again, design 51 Publishing); a chunk that is neither the + // cursor nor 0 is refused, naming the cursor. + if _, err := b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t1", Digest: part.Digest, Sequence: 0, Data: blob[:1000]}); err != nil { + t.Errorf("a chunk 0 sent again was refused: %v", err) + } + if _, err := b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t1", Digest: part.Digest, Sequence: 2, Data: blob[:1000]}); err == nil { + t.Error("a chunk out of sequence was taken: the transfer goes on from the chunks held") } a, err = b.h.PublishPart(ArtifactSeat, builder, PartRequest{Transfer: "t1", Digest: part.Digest, Sequence: 1, Last: true, Data: blob[1000:]}) if err != nil || !a.Ended || len(b.said) != 1 || b.said[0].event != EventPublished || b.said[0].body.Reference == "" { @@ -865,7 +870,7 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { if _, ok := claims[OldSeat]; !ok { t.Error("the old seat mesh-artifact-store is no longer claimed while its window is open") } - if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-view.delivered", "node-engine.manifest-reads-changed"}) { + if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-view.delivered", "node-engine.manifest-reads-changed", "mesh-conditions.findings-asked"}) { t.Errorf("consumes = %v, want delivered and manifest-reads-changed (the windows' wakes)", m.Consumes) } // E14: the registry container, on the loopback, is listed before the program, so the port is free when it starts. diff --git a/modules/distribution/module.json b/modules/distribution/module.json index 04b974bbb..82f16fa97 100644 --- a/modules/distribution/module.json +++ b/modules/distribution/module.json @@ -67,7 +67,8 @@ ], "consumes": [ "mesh-delivery-view.delivered", - "node-engine.manifest-reads-changed" + "node-engine.manifest-reads-changed", + "mesh-conditions.findings-asked" ], "config": { "callers": { @@ -278,5 +279,17 @@ } } ] - } + }, + "streams": [ + { + "name": "distribution-findings", + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } + ] } diff --git a/modules/gitea/cmd/package-registry/findings.go b/modules/gitea/cmd/package-registry/findings.go new file mode 100644 index 000000000..ebbce5955 --- /dev/null +++ b/modules/gitea/cmd/package-registry/findings.go @@ -0,0 +1,79 @@ +package main + +import ( + "context" + "strings" + "time" + + meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + keyparts "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" + + "gitea/internal/packageseats" +) + +// findingKinds are the kinds this bundle says: the token refusal and the transfers' (design 47 G §580, §689); the module's +// TypeScript bundle shares the state `findings` and owns its own. +var findingKinds = []string{PackagesTokenRefused, "publish-refused", "incoming-full"} + +// findingWriter is the module's writer (views.Writer) as the keeper writes and clears through it. +type findingWriter interface { + Write(ctx context.Context, finding meshconditions.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error +} + +// Findings keeps a bundle's open findings and says them through the module's writer (design 47 G §689, A-14): its `Say` is +// the Finding callback of the Driver and the Holder; the writer is the held writer (views.HeldWriter), which holds what is open (A-29, A-32). +type Findings struct { + Writer findingWriter + Logf func(string, ...any) + Now func() time.Time +} + +// Say raises (raised) or clears the finding of a kind about a subject. Its id is the seat's and the subject's, each escaped +// as a key's name is (design 47 G A-14); the kind stays the bundle's own word. +func (k *Findings) Say(kind, subject, words string, raised bool) { + id := keyparts.EscapeName(packageseats.Self) + "." + keyparts.EscapeName(subject) + key := views.FindingKey("seat", id, kind) + if !raised { + if err := k.Writer.Clear(context.Background(), "seat", id, kind, "it no longer holds"); err != nil { + k.Logf("the finding %s could not be cleared: %v", key, err) + } + return + } + at := time.Now() + if k.Now != nil { + at = k.Now() + } + f := meshconditions.Finding{Scope: "seat", ID: id, Kind: kind, Subject: meshconditions.Subject{Scope: "seat", ID: id}, + Severity: meshconditions.FindingSeverityWarning, Summary: words, Headline: words, Explanation: words, + Source: kind, Observer: packageseats.Self, At: at.UTC()} + if err := k.Writer.Write(context.Background(), f); err != nil { + k.Logf("the finding %s could not be written: %v", key, err) + } +} + +// DeleteOldFindings deletes, once at a module's start, the findings the keys of an older form hold (design 47 G A-31): the form +// `seat.._`, whose kind is no kind the writer owns in its key, so the writer's own first statement does not +// clear it. Only a key of one of kinds is deleted; the keys of the writer's form (`seat...`) are left. +func DeleteOldFindings(state views.FindingState, seat string, kinds []string) error { + keys, err := state.Keys() + if err != nil { + return err + } + for _, k := range keys { + rest, ok := strings.CutPrefix(k, "seat."+seat+".") + if !ok { + continue + } + for _, kind := range kinds { + if strings.HasPrefix(rest, kind+"_") { + if err := state.Delete(k); err != nil { + return err + } + break + } + } + } + return nil +} diff --git a/modules/gitea/cmd/package-registry/findings_test.go b/modules/gitea/cmd/package-registry/findings_test.go new file mode 100644 index 000000000..df05ad213 --- /dev/null +++ b/modules/gitea/cmd/package-registry/findings_test.go @@ -0,0 +1,44 @@ +package main + +import ( + "sort" + "strings" + "testing" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// A-31: the keys of the older form (`seat.._`) are deleted once; a key of the writer's form, another seat's +// and another kind's stay. +func TestTheOldFindingKeysAreDeletedOnce(t *testing.T) { + state := &oldKeysState{keys: map[string]bool{}} + for _, k := range []string{ + "seat.gitea.packages-token-refused_write", "seat.gitea.packages-token-refused_a_b_c", + "seat.gitea.sub.packages-token-refused", "seat.other.packages-token-refused_write", "seat.gitea.unrelated_write"} { + state.keys[k] = true + } + if err := DeleteOldFindings(state, "gitea", []string{"packages-token-refused", "publish-refused", "incoming-full"}); err != nil { + t.Fatal(err) + } + var left []string + for k := range state.keys { + left = append(left, k) + } + sort.Strings(left) + if got, want := strings.Join(left, " "), "seat.gitea.sub.packages-token-refused seat.gitea.unrelated_write seat.other.packages-token-refused_write"; got != want { + t.Errorf("left %s, want %s", got, want) + } +} + +type oldKeysState struct{ keys map[string]bool } + +func (s *oldKeysState) Get(string) (*stdio.StateEntry, error) { return nil, nil } +func (s *oldKeysState) Put(string, any) (uint64, error) { return 0, nil } +func (s *oldKeysState) Delete(k string) error { delete(s.keys, k); return nil } +func (s *oldKeysState) Keys() ([]string, error) { + var out []string + for k := range s.keys { + out = append(out, k) + } + return out, nil +} diff --git a/modules/gitea/cmd/package-registry/forge.go b/modules/gitea/cmd/package-registry/forge.go index f752a4116..82fb0e69a 100644 --- a/modules/gitea/cmd/package-registry/forge.go +++ b/modules/gitea/cmd/package-registry/forge.go @@ -37,7 +37,7 @@ type Forge struct { // key in the module's findings state (packages-account.ts PACKAGES_TOKEN_REFUSED watches that key). const ( PackagesTokenRefused = "packages-token-refused" - PackagesTokenRefusedKey = "seat.gitea.packages-token-refused_write" + PackagesTokenRefusedKey = "seat.gitea.write.packages-token-refused" ) // wrote is a write's answer, said on Refused: a 401 raises the finding, a write the forge took clears it. @@ -86,7 +86,7 @@ func ForgeFromEnv() (*Forge, error) { } } f := &Forge{Client: c, Writer: &w, Owner: os.Getenv("MESH_NPM_OWNER"), Refused: func(raised bool) { - finding(PackagesTokenRefused, "write", "the forge refused the package seats' write token; the module mints it afresh", raised) + findings.Say(PackagesTokenRefused, "write", "the forge refused the package seats' write token; the module mints it afresh", raised) }} if f.Owner == "" { missing = append(missing, "MESH_NPM_OWNER") diff --git a/modules/gitea/cmd/package-registry/main.go b/modules/gitea/cmd/package-registry/main.go index 0869cfca1..02de1384c 100644 --- a/modules/gitea/cmd/package-registry/main.go +++ b/modules/gitea/cmd/package-registry/main.go @@ -13,7 +13,9 @@ import ( "os" "strings" + meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" "gitea/internal/packageseats" ) @@ -23,7 +25,30 @@ import ( // of the module's seats as their verbs. The README names these two, and a test holds the two together. var ToolNames = []string{"go_packages", "go_retention", "go_incoming", "npm_packages", "npm_retention", "npm_incoming"} +// findings keeps this bundle's open findings for its writer (design 47 G §689). +var findings = &Findings{Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, f+"\n", a...) }} + func main() { + // The bundle's findings go through the module's held writer (design 47 G A-16, A-29, A-32). + w, err := views.NewStdioHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: packageseats.Self, State: stdio.State("findings"), Kinds: findingKinds}) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + findings.Writer = w + // the runtime answers the writer's state and subscription only once Serve has its channel + go func() { + <-stdio.Served() + // the keys of the older form end in no kind the writer owns, so it does not clear them (A-31) + if err := DeleteOldFindings(stdio.State("findings"), packageseats.Self, findingKinds); err != nil { + fmt.Fprintln(os.Stderr, "the old findings are not deleted:", err) + } + if err := w.Start(context.Background()); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + }() forge := func() (*Forge, error) { return ForgeFromEnv() } holder := func() (*packageseats.Holder, error) { return busHolder(forge) } go followIdle(holder) diff --git a/modules/gitea/cmd/package-registry/seats.go b/modules/gitea/cmd/package-registry/seats.go index 4ec5770e8..336e473c7 100644 --- a/modules/gitea/cmd/package-registry/seats.go +++ b/modules/gitea/cmd/package-registry/seats.go @@ -22,14 +22,10 @@ import ( "io" "net/http" "net/url" - "os" "path" "strings" - "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" - meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" goreg "git.novox.be/novox/mesh-delivery/contracts/go/seats/gopackageregistry" npmreg "git.novox.be/novox/mesh-delivery/contracts/go/seats/npmpackageregistry" "git.novox.be/novox/mesh-delivery/sdk/parts" @@ -249,21 +245,6 @@ func sayOn(seat, event string, o packageseats.Outcome) error { return stdio.SeatPublish("mesh.seat."+seat+".event."+event, body, "") } -func finding(kind, subject, words string, raised bool) { - key := meshconditionsclient.Key(meshconditionsclient.ScopeSeat, packageseats.Self, kind+"."+strings.NewReplacer(".", "_", "@", "_", "/", "_").Replace(subject)) - if !raised { - _ = stdio.State(meshconditions.StateFindings).Delete(key) - return - } - f := meshconditions.FindingsByKeyValue{V: meshconditionsclient.FindingVersion, Key: key, Kind: kind, Ref: key[strings.LastIndex(key, ".")+1:], - Subject: meshconditions.Subject{Scope: meshconditionsclient.ScopeSeat, ID: packageseats.Self}, Severity: meshconditionsclient.Warning, - Summary: words, Headline: words, Explanation: words, Resolved: "cleared once it no longer holds", - Source: kind, Observer: packageseats.Self, At: time.Now().UTC()} - if _, err := stdio.State(meshconditions.StateFindings).Put(key, f); err != nil { - fmt.Fprintf(os.Stderr, "the finding %s could not be written: %v\n", key, err) - } -} - // recordAct records a hand act through the seat mesh-hand-acts alone. func recordAct(q meshhandactsclient.Request) error { _, err := meshhandactsclient.Recorder{Ask: stdio.Ask}.Record(q) @@ -277,7 +258,7 @@ var boundCap = parts.BusCap(packageseats.IncomingObjects) func busHolder(forge func() (*Forge, error)) (*packageseats.Holder, error) { return &packageseats.Holder{Objects: parts.BusObjects(packageseats.IncomingObjects), Transfers: parts.BusTransfers[packageseats.Opening](packageseats.TransfersState), - Builder: packageseats.Reserved{Ask: stdio.Ask}, Store: seatStore{forge}, Say: say, Finding: finding, + Builder: packageseats.Reserved{Ask: stdio.Ask}, Store: seatStore{forge}, Say: say, Finding: findings.Say, Cap: boundCap, Record: recordAct, Idle: parts.BusArm(packageseats.IdleState)}, nil } diff --git a/modules/gitea/cmd/package-registry/seats_test.go b/modules/gitea/cmd/package-registry/seats_test.go index 6deb1ab47..13d4fbd0f 100644 --- a/modules/gitea/cmd/package-registry/seats_test.go +++ b/modules/gitea/cmd/package-registry/seats_test.go @@ -19,8 +19,10 @@ import ( "testing" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + keyparts "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" "gitea/internal/forgehttp" + "gitea/internal/packageseats" ) @@ -133,7 +135,7 @@ func TestThePackageSeatsStoreIntoTheForge(t *testing.T) { // and the next write the forge takes clears it (no timer looks at the token). var refused []bool forge.Refused = func(raised bool) { refused = append(refused, raised) } - if k := meshconditionsclient.Key(meshconditionsclient.ScopeSeat, packageseats.Self, PackagesTokenRefused+".write"); k != PackagesTokenRefusedKey { + if k := views.FindingKey("seat", keyparts.EscapeName(packageseats.Self)+".write", PackagesTokenRefused); k != PackagesTokenRefusedKey { t.Errorf("the refusal's key is %s, the watcher watches %s", k, PackagesTokenRefusedKey) } _ = os.WriteFile(writer.TokenFile, []byte("stale\n"), 0o600) diff --git a/modules/gitea/cmd/package-registry/whole_integration_test.go b/modules/gitea/cmd/package-registry/whole_integration_test.go index 5a52ce7a4..b7b76057f 100644 --- a/modules/gitea/cmd/package-registry/whole_integration_test.go +++ b/modules/gitea/cmd/package-registry/whole_integration_test.go @@ -19,7 +19,9 @@ import ( "bytes" "context" "crypto/rand" + "crypto/sha256" "encoding/base64" + "encoding/hex" "encoding/json" "errors" "fmt" @@ -37,6 +39,7 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/testbus" "gitea/internal/packageseats" @@ -60,6 +63,8 @@ type player struct { in io.Writer next int answers map[string]chan wire + puts map[string][]byte // the put sessions open, by call + gets map[string][]byte // the get sessions open, by call } func (p *player) send(m map[string]any) { @@ -127,8 +132,15 @@ func (p *player) handle(m wire) { p.answer(m.ID, map[string]any{"module": "m", "version": q.Version, "artifacts": []map[string]any{{"artifact": "m:a", "digest": d}}}, nil) case m.Method == "mesh/seat.publish": p.answer(m.ID, map[string]any{}, p.nc.Publish(a.Subject, a.Body)) - case m.Method == "mesh/publish": + case m.Method == "mesh/capabilities": + p.answer(m.ID, map[string]any{"words": []string{"put-sessions", "finding-calls"}}, nil) + case m.Method == "mesh/publish", m.Method == "mesh/subscribe": + // the module's finding writer emits its events and subscribes to findings-asked (design 47 G A-16); none is played p.answer(m.ID, map[string]any{}, nil) + case m.Method == buscalls.ObjectsPutChunk: + p.putChunk(m) + case m.Method == buscalls.ObjectsGetChunk: + p.getChunk(m) case strings.HasPrefix(m.Method, "mesh/objects."): p.objects(m, a.Objects, a.Name, a.Data) case strings.HasPrefix(m.Method, "mesh/state."): @@ -138,6 +150,59 @@ func (p *player) handle(m wire) { } } +// putChunk plays the runtime's put session (design 47 G §518, A-21): the chunks of one call in order, each answered as taken, the +// last put whole and answered as the put's. +func (p *player) putChunk(m wire) { + var c buscalls.ObjectsPutChunkRequest + _ = json.Unmarshal(m.Params, &c) + p.mu.Lock() + if p.puts == nil { + p.puts = map[string][]byte{} + } + if c.Seq == 0 || (len(p.puts[c.Call]) == 0 && c.Seq != 1) { + p.mu.Unlock() + p.answer(m.ID, buscalls.ObjectsChunkTaken{Seq: c.Seq, Error: "a put session begins at chunk 1"}, nil) + return + } + p.puts[c.Call] = append(p.puts[c.Call], c.Data...) + whole := p.puts[c.Call] + if c.Last { + delete(p.puts, c.Call) + } + p.mu.Unlock() + if !c.Last { + p.answer(m.ID, buscalls.ObjectsChunkTaken{Seq: c.Seq}, nil) + return + } + os, err := p.js.ObjectStore(p.ctx, "gitea_"+c.Objects) + if err != nil { + p.answer(m.ID, nil, err) + return + } + info, err := os.PutBytes(p.ctx, c.Name, whole) + if err != nil { + p.answer(m.ID, nil, err) + return + } + p.answer(m.ID, map[string]any{"name": info.Name, "size": info.Size, "digest": info.Digest, "seq": c.Seq}, nil) +} + +// getChunk plays the runtime's get session: the chunk of the session's value. +func (p *player) getChunk(m wire) { + var c buscalls.ObjectsGetChunkRequest + _ = json.Unmarshal(m.Params, &c) + p.mu.Lock() + whole := p.gets[c.Call] + p.mu.Unlock() + at := int(c.Seq-1) * buscalls.ChunkBytes + if whole == nil || c.Seq == 0 || at >= len(whole) { + p.answer(m.ID, buscalls.ObjectsChunk{Seq: c.Seq, Error: "no such chunk of the session"}, nil) + return + } + end := min(at+buscalls.ChunkBytes, len(whole)) + p.answer(m.ID, buscalls.ObjectsChunk{Seq: c.Seq, Last: end == len(whole), Data: whole[at:end]}, nil) +} + func (p *player) objects(m wire, local, name string, data []byte) { os, err := p.js.ObjectStore(p.ctx, "gitea_"+local) if err != nil { @@ -154,6 +219,20 @@ func (p *player) objects(m wire, local, name string, data []byte) { p.answer(m.ID, map[string]any{"name": info.Name, "size": info.Size, "digest": info.Digest}, nil) case "mesh/objects.get": got, err := os.GetBytes(p.ctx, name) + if err == nil && len(got) > buscalls.ChunkBytes { + // a value over one chunk is read by a session (design 47 G §490, §518) + sum := sha256.Sum256(got) + call := fmt.Sprintf("get-%s-%d", name, len(got)) + p.mu.Lock() + if p.gets == nil { + p.gets = map[string][]byte{} + } + p.gets[call] = got + p.mu.Unlock() + p.answer(m.ID, buscalls.ObjectsChunked{Chunked: json.RawMessage("true"), Size: uint64(len(got)), + Digest: "sha256:" + hex.EncodeToString(sum[:]), Call: call}, nil) + break + } p.answer(m.ID, map[string]any{"info": map[string]any{"name": name, "size": len(got)}, "data": got}, err) case "mesh/objects.delete": err := os.Delete(p.ctx, name) @@ -307,7 +386,7 @@ func TestThePackageSeatsWhole(t *testing.T) { return file } cmd := exec.CommandContext(ctx, bin) - cmd.Env = append(os.Environ(), "MESH_GITEA_URL="+srv.URL, "MESH_NPM_OWNER=acme", "MESH_GITEA_TOOLS_USER=mesh-tools", + cmd.Env = append(os.Environ(), "MESH_NODE=novox", "MESH_GITEA_URL="+srv.URL, "MESH_NPM_OWNER=acme", "MESH_GITEA_TOOLS_USER=mesh-tools", "MESH_GITEA_TOKEN_FILE="+kept("token", "reader", "read:package,read:user,write:issue,write:repository for mesh-tools"), "MESH_GITEA_PACKAGES_TOKEN_FILE="+kept("packages-token", "writer", "write:package for mesh-packages")) cmd.Stderr = os.Stderr diff --git a/modules/gitea/go.mod b/modules/gitea/go.mod index d84ba67cf..11e94c655 100644 --- a/modules/gitea/go.mod +++ b/modules/gitea/go.mod @@ -3,7 +3,6 @@ module gitea go 1.26.0 require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/provisioning/go/clients/meshprovisionerclient v0.0.0-00010101000000-000000000000 diff --git a/modules/gitea/module.json b/modules/gitea/module.json index 5255a55f8..d7a18c757 100644 --- a/modules/gitea/module.json +++ b/modules/gitea/module.json @@ -657,6 +657,15 @@ "git.branch-moved", "git.pull-updated", "git.pull-closed", - "git.pull-merged" + "git.pull-merged", + "mesh-conditions.findings-asked" + ], + "streams": [ + { + "name": "gitea-findings", + "events": ["finding-written.*", "finding-cleared.*", "findings-stated.*"], + "max-age-days": 7, + "max-mib": 16 + } ] } diff --git a/modules/gitea/packages-account.ts b/modules/gitea/packages-account.ts index ffbfab7b9..ac0ec296b 100644 --- a/modules/gitea/packages-account.ts +++ b/modules/gitea/packages-account.ts @@ -89,7 +89,7 @@ export async function keepPackagesToken(source: MintedToken, url: string, owner } /** The finding the Go bundle raises in the module's findings state when the forge refuses the write token (cmd/package-registry PackagesTokenRefusedKey). */ -export const PACKAGES_TOKEN_REFUSED = "seat.gitea.packages-token-refused_write"; +export const PACKAGES_TOKEN_REFUSED = "seat.gitea.write.packages-token-refused"; /** One change of the findings state, as its watch hands it. */ export interface FindingChange { readonly key: string; readonly op: "put" | "delete" } diff --git a/modules/mesh-build-go/cmd/mesh-build-go/compose.go b/modules/mesh-build-go/cmd/mesh-build-go/compose.go index a28671458..35afca719 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/compose.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/compose.go @@ -25,11 +25,12 @@ import ( "time" "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" "git.novox.be/novox/mesh-delivery/sdk/composer" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -37,6 +38,9 @@ import ( // module invokes it, module.json invokes). A manifest by its digest never changes and a compose ask is rare, so the // module asks for the one key it needs when it needs it, and keeps no picture of the indexer's whole state (ADR 0615). // indexerModule is the module that holds the seat mesh-catalog, whose manifests state a compose ask reads. +// findingsModule is the module whose findings these are. +const findingsModule = "mesh-build-go" + const indexerModule = "mesh-delivery-indexer" const ManifestsTool = indexerModule + ".indexer_read_manifests" @@ -83,18 +87,53 @@ func (m *Module) compose(ctx context.Context, fetch buildstep.StepFunc, a sdkste // askManifest asks the indexer's read tool for one manifest by its key: nil when it keeps none. A call the indexer does not // answer is a finding of this module's own, which the condition keeper raises (ADR 0615 §6) and the next answer clears. func askManifest(key string) (json.RawMessage, error) { - return askManifestVia(stdio.Ask, meshconditionsclient.FindingWriter{State: stdio.State("findings")}, time.Now, key) + return askManifestVia(stdio.Ask, manifestFindings, time.Now, key) } -func askManifestVia(ask func(string, any) (json.RawMessage, error), findings views.FindingWriter, now func() time.Time, key string) (json.RawMessage, error) { - st := views.Status{Owner: indexerModule, Tool: ManifestsTool, State: "manifests", Asked: now().UTC()} +// manifestFindings is where askManifest writes its finding: the module's writer once main has started it (design 47 G §689, +// A-16); outside a launch by the runtime, which has no node and no state, nothing is written. +var manifestFindings findingWriter = noFindings{} + +// findingWriter is the module's writer (views.Writer) as askManifestVia writes and clears through it. +type findingWriter interface { + Write(ctx context.Context, finding seat.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error +} + +type noFindings struct{} + +func (noFindings) Write(context.Context, seat.Finding) error { return nil } +func (noFindings) Clear(context.Context, string, string, string, string) error { return nil } + +// unansweredManifest is the finding of an ask the indexer did not answer, laid out as mesh-sdk lays out a views reader's +// first-view-unanswered (design 47 G §687, A-28): the id is the view's name, escaped as a key's name is. +func unansweredManifest(err string, asked, at time.Time) seat.Finding { + id := keys.EscapeName(findingsModule + "." + indexerModule + ".manifests") + return seat.Finding{Kind: views.UnansweredKind, Scope: "module", ID: id, Severity: seat.FindingSeverityWarning, Source: "views", + Subject: seat.Subject{Scope: "module", ID: findingsModule, Machine: os.Getenv("MESH_NODE")}, + Summary: fmt.Sprintf("%s has no first view of %s's manifests: %s has not answered since %s (%s)", findingsModule, indexerModule, ManifestsTool, + asked.Format(time.RFC3339), strings.TrimSpace(err)), + Headline: fmt.Sprintf("%s waits for %s", findingsModule, indexerModule), At: at.UTC()} +} + +// newFindingsWriter is this module's finding writer: it states the finding of askManifest and observes none of its own. +func newFindingsWriter() (*views.Writer[seat.Finding, seat.FindingCleared, seat.FindingsStated, seat.FindingsAskedEvent], error) { + return views.NewStdioWriter(views.WriterConfig[seat.Finding, seat.FindingCleared, seat.FindingsStated, seat.FindingsAskedEvent]{ + Asked: seat.SubjectFindingsAsked, Module: findingsModule, State: stdio.State("findings"), + Observe: func(context.Context) (views.Observed[seat.Finding], error) { + return views.Observed[seat.Finding]{}, nil + }}) +} + +func askManifestVia(ask func(string, any) (json.RawMessage, error), findings findingWriter, now func() time.Time, key string) (json.RawMessage, error) { + asked := now().UTC() raw, err := ask(ManifestsTool, map[string]any{"key": key}) + f := unansweredManifest("", asked, now()) if err != nil { - st.Error = err.Error() - _ = findings.Put(views.UnansweredFinding("mesh-build-go", st, now())) + _ = findings.Write(context.Background(), unansweredManifest(err.Error(), asked, now())) return nil, err } - _ = findings.Delete(views.UnansweredFinding("mesh-build-go", st, now())) + _ = findings.Clear(context.Background(), f.Scope, f.ID, f.Kind, "the indexer answered") var a struct { Found bool `json:"found"` Value json.RawMessage `json:"value"` diff --git a/modules/mesh-build-go/cmd/mesh-build-go/compose_test.go b/modules/mesh-build-go/cmd/mesh-build-go/compose_test.go index 722b5abc7..332f60d72 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/compose_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/compose_test.go @@ -11,12 +11,12 @@ import ( "testing" "time" + seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" catalogclient "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" meshcat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" sdkstep "git.novox.be/novox/mesh-delivery/sdk/buildstep" "git.novox.be/novox/mesh-delivery/sdk/composer" "git.novox.be/novox/mesh-sdk/go/forgekey" - "git.novox.be/novox/mesh-sdk/go/views" ) // nodeEngineComposer is the composer command of the node-engine's repository, whose Go module is at its root (ADR 0593 §7): @@ -187,8 +187,14 @@ func TestAnIndexerThatDoesNotAnswerIsAFinding(t *testing.T) { type mapFindings map[string][]byte -func (m mapFindings) Put(f views.Finding) error { m[f.ID+"."+f.Kind] = []byte(f.Summary); return nil } -func (m mapFindings) Delete(f views.Finding) error { delete(m, f.ID+"."+f.Kind); return nil } +func (m mapFindings) Write(_ context.Context, f seat.Finding) error { + m[f.ID+"."+f.Kind] = []byte(f.Summary) + return nil +} +func (m mapFindings) Clear(_ context.Context, _, id, kind, _ string) error { + delete(m, id+"."+kind) + return nil +} // ADR 0593 §7 (C9): a compose ask naming the node-engine's repository builds its composer command in the Go module at the // repository's root, under a key of that repository, commit and package, and runs the program over the request with its diff --git a/modules/mesh-build-go/cmd/mesh-build-go/main.go b/modules/mesh-build-go/cmd/mesh-build-go/main.go index 317911181..8817fc799 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/main.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/main.go @@ -71,6 +71,20 @@ func main() { // Outside a launch by the runtime — a build, a check — there is no node and no state to write. if m.Node != "" { go m.Start(ctx) + findings, err := newFindingsWriter() + if err != nil { + logf("the finding writer: %v", err) + os.Exit(1) + } + manifestFindings = findings + // the runtime answers the writer's subscription only once Serve has its channel + go func() { + <-stdio.Served() + if err := findings.Start(ctx); err != nil { + logf("the finding writer: %v", err) + os.Exit(1) + } + }() } go startKeeper(ctx) // Every state it keeps, said on the bus before it serves (ADR 0495 §2; issue 1790). diff --git a/modules/mesh-build-go/go.mod b/modules/mesh-build-go/go.mod index 87a97efea..a3591514d 100644 --- a/modules/mesh-build-go/go.mod +++ b/modules/mesh-build-go/go.mod @@ -101,7 +101,7 @@ require git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys v0.0.0-000101 replace git.novox.be/novox/mesh-catalog/modules/ssh-client/sshkeys => ../ssh-client/sshkeys require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-delivery/contracts/go/clients/gopackageregistryclient v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-delivery/contracts/go/clients/npmpackageregistryclient v0.0.0-00010101000000-000000000000 // indirect diff --git a/modules/mesh-config/cmd/mesh-config/config.go b/modules/mesh-config/cmd/mesh-config/config.go index a745180e4..1708f57a8 100644 --- a/modules/mesh-config/cmd/mesh-config/config.go +++ b/modules/mesh-config/cmd/mesh-config/config.go @@ -86,6 +86,9 @@ type Config struct { Logf func(format string, a ...any) mu sync.Mutex + // Findings is the module's finding writer (design 47 G §689); nil, no finding is said. findMu serializes a pass over what it holds. + Findings FindingsWriter + findMu sync.Mutex // withheld is why the newest list of predefined keys was not published, empty when it was (publish). withheld string diff --git a/modules/mesh-config/cmd/mesh-config/findings.go b/modules/mesh-config/cmd/mesh-config/findings.go index 3f9c1482d..09531271c 100644 --- a/modules/mesh-config/cmd/mesh-config/findings.go +++ b/modules/mesh-config/cmd/mesh-config/findings.go @@ -5,17 +5,19 @@ package main // is gone. The keeper raises and clears the condition of the same key (mesh-conditions' Observers and Homes). import ( + "context" "fmt" "slices" "strings" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" ) // FindingsState is the state, meshconditions.StateFindings. -const FindingsState = meshconditions.StateFindings +const FindingsState = "findings" // The kinds of finding mesh-config writes. const ( @@ -31,89 +33,95 @@ const ( FindingRetireHeld = "retirement-held" ) -// findingKey is the key of a finding of kind about a module on a node: the condition's key. -func findingKey(kind, module, node, ref string) string { - id := module +// findingID is the id of a finding about a module on a node: the module, the node when there is one, and the ref +// (`..`), each escaped as a key's name is, since the kind is a fixed word (design 47 G A-14). +func findingID(module, node, ref string) string { + id := keys.EscapeName(module) if node != "" { - id += "." + node + id += "." + keys.EscapeName(node) } - return meshconditionsclient.Key(meshconditionsclient.ScopeModule, id, kind+"-"+ref) + return id + "." + keys.EscapeName(ref) } -func finding(kind, module, node, ref, summary, headline, explanation, needs, resolved, said string, at time.Time) meshconditions.FindingsByKeyValue { - id := module - if node != "" { - id += "." + node - } - return meshconditions.FindingsByKeyValue{V: meshconditionsclient.FindingVersion, Key: findingKey(kind, module, node, ref), Kind: kind, Ref: kind + "-" + ref, - Subject: meshconditions.Subject{Scope: meshconditionsclient.ScopeModule, ID: id, Machine: node}, - Severity: meshconditionsclient.Warning, Resolver: meshconditionsclient.ResolverOperator, Summary: summary, Headline: headline, - Explanation: explanation, Needs: needs, Resolved: resolved, +func finding(kind, module, node, ref, summary, headline, explanation, needs, said string, at time.Time) meshconditions.Finding { + id := findingID(module, node, ref) + return meshconditions.Finding{Scope: "module", ID: id, Kind: kind, + Subject: meshconditions.Subject{Scope: "module", ID: id, Machine: node}, + Severity: meshconditions.FindingSeverityWarning, Resolver: "operator", Summary: summary, Headline: headline, + Explanation: explanation, Needs: needs, Evidence: []meshconditions.Evidence{{At: at.UTC(), Said: said}}, Source: "config", Observer: Module, At: at.UTC()} } // renameRefused is the finding of a move mesh-config refused. -func renameRefused(module, to, from, why string, at time.Time) meshconditions.FindingsByKeyValue { +func renameRefused(module, to, from, why string, at time.Time) meshconditions.Finding { return finding(FindingRenameRefused, module, "", to, fmt.Sprintf("the rename of %s's config value %s to %s is refused: %s", module, from, to, why), fmt.Sprintf("%s's value %s is not moved to %s", module, from, to), fmt.Sprintf("The new version of %s names %s as the new name of its config value %s, and mesh-config refused to move the value: %s. Nothing was moved.", module, to, from, why), fmt.Sprintf("put the cause right, then move it with config_rename_again (module %s, to %s); the details name it.", module, to), - fmt.Sprintf("Resolved: %s's value %s is under %s", module, from, to), why, at) + why, at) } // windowHeld is the finding of a rename window held open past HeldFinding by a node. -func windowHeld(r Rename, node string, at time.Time) meshconditions.FindingsByKeyValue { +func windowHeld(r Rename, node string, at time.Time) meshconditions.Finding { return finding(FindingWindowHeld, r.Module, node, r.To, fmt.Sprintf("%s on %s has held the old config name %s open for %d days", r.Module, node, r.From, int(HeldFinding.Hours()/24)), fmt.Sprintf("%s on %s still runs the old config name", r.Module, node), fmt.Sprintf("The rename of %s's config value %s to %s is not complete on %s: its node-engine has not reported a build that reads %s (it may be away).", r.Module, r.From, r.To, node, r.To), fmt.Sprintf("bring %s back, or drop it from the window with config_rename_close.", node), - fmt.Sprintf("Resolved: %s reads %s on %s", r.Module, r.To, node), fmt.Sprintf("the window opened %s and %s has not reported a build reading %s", r.Nodes[node].Since.UTC().Format(time.RFC3339), node, r.To), at) } -// setFinding writes a finding when it differs from the one kept. -func (c *Config) setFinding(f meshconditions.FindingsByKeyValue) { - if err := meshconditionsclient.CheckFinding(f); err != nil { +// FindingsWriter is the module's held finding writer (views.HeldWriter), through which mesh-config writes and clears its +// findings and reads what it holds (design 47 G §689, A-29, A-38): the module keeps no set of its own. +type FindingsWriter interface { + Write(ctx context.Context, finding meshconditions.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error + Held() []meshconditions.Finding +} + +// setFinding writes a finding when it differs from the one held; the module and ref are the parts of its id. +func (c *Config) setFinding(module, ref string, f meshconditions.Finding) { + if c.Findings == nil { + return + } + if err := views.CheckFinding(f); err != nil { c.logf("[mesh-config] a finding is not valid: %v", err) return } - if e, err := c.Store.Get(FindingsState, f.Key); err == nil && e != nil { - if old, derr := meshconditions.DecodeFindingsByKeyValue(e.Value); derr == nil && old.Summary == f.Summary { + key := views.FindingKey(f.Scope, f.ID, f.Kind) + c.findMu.Lock() + defer c.findMu.Unlock() + for _, old := range c.Findings.Held() { + if views.FindingKey(old.Scope, old.ID, old.Kind) == key && old.Summary == f.Summary { return // what it says has not changed } } - if err := c.Store.Put(FindingsState, f.Key, f); err != nil { - c.logf("[mesh-config] the finding %s is not written: %v", f.Key, err) + if err := c.Findings.Write(context.Background(), f); err != nil { + c.logf("[mesh-config] the finding %s is not written: %v", key, err) } } -// clearFinding deletes the findings of a kind and ref about a module, on whatever node but those in keep. +// clearFinding clears the findings of a kind and ref about a module, on whatever node but those in keep. func (c *Config) clearFinding(kind, module, ref string, keep ...string) { - keys, err := c.Store.Keys(FindingsState) - if err != nil { + if c.Findings == nil { return } - want := kind + "-" + ref - for _, k := range keys { - e, err := c.Store.Get(FindingsState, k) - if err != nil || e == nil { - continue - } - f, err := meshconditions.DecodeFindingsByKeyValue(e.Value) - if err != nil || f.Ref != want || !(f.Subject.ID == module || strings.HasPrefix(f.Subject.ID, module+".")) || + c.findMu.Lock() + defer c.findMu.Unlock() + for _, f := range c.Findings.Held() { + if f.Kind != kind || f.ID != findingID(module, f.Subject.Machine, ref) || (f.Subject.Machine != "" && slices.Contains(keep, f.Subject.Machine)) { continue } - if err := c.Store.Delete(FindingsState, k); err != nil { - c.logf("[mesh-config] the finding %s is not cleared: %v", k, err) + if err := c.Findings.Clear(context.Background(), f.Scope, f.ID, f.Kind, "the fault is gone"); err != nil { + c.logf("[mesh-config] the finding %s is not cleared: %v", views.FindingKey(f.Scope, f.ID, f.Kind), err) } } } // retirementHeld is the finding of a retirement that waits seven days for nodes that still run a build reading the value. -func retirementHeld(e Retiring, at time.Time) meshconditions.FindingsByKeyValue { +func retirementHeld(e Retiring, at time.Time) meshconditions.Finding { node := "" if len(e.Nodes) > 0 { node = e.Nodes[0] @@ -123,17 +131,15 @@ func retirementHeld(e Retiring, at time.Time) meshconditions.FindingsByKeyValue fmt.Sprintf("%s's %s cannot be retired yet", e.Module, e.Key), fmt.Sprintf("No manifest reads %s's %s any more, but %s still run a build that does, so the operator is not asked to remove it.", e.Module, e.Key, strings.Join(e.Nodes, ", ")), "bring those nodes up to date, or cancel the retirement with config_retiring.", - fmt.Sprintf("Resolved: no node reads %s's %s", e.Module, e.Key), "retiring since "+e.Since.UTC().Format(time.RFC3339)+", still read on "+strings.Join(e.Nodes, ", "), at) } // retireUnanswered is the finding of a retirement asked twice and not answered. -func retireUnanswered(e Retiring, at time.Time) meshconditions.FindingsByKeyValue { +func retireUnanswered(e Retiring, at time.Time) meshconditions.Finding { return finding(FindingRetireUnanswered, e.Module, "", e.Key, fmt.Sprintf("the operator did not answer the ask to remove %s's %s, asked twice", e.Module, e.Key), fmt.Sprintf("%s's %s waits for an answer", e.Module, e.Key), fmt.Sprintf("No manifest reads %s's %s any more. The operator was asked twice whether to remove it and did not answer, so it stays.", e.Module, e.Key), "answer the ask again with config_retiring ask-again, or cancel it.", - fmt.Sprintf("Resolved: %s's %s is answered or read again", e.Module, e.Key), "asked "+e.Asked.UTC().Format(time.RFC3339)+" and once more", at) } diff --git a/modules/mesh-config/cmd/mesh-config/main.go b/modules/mesh-config/cmd/mesh-config/main.go index 16a839755..d094be9b8 100644 --- a/modules/mesh-config/cmd/mesh-config/main.go +++ b/modules/mesh-config/cmd/mesh-config/main.go @@ -18,10 +18,11 @@ import ( "strings" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" ) // Seat is the role this module holds, once for the mesh, and Verb its one verb. @@ -120,13 +121,30 @@ func (runtimeAsks) Warranted(ask string) error { func main() { c := &Config{Store: runtimeStore{}, Logf: logf, Asks: runtimeAsks{}, Forge: toolForge{ask: stdio.Ask}, Delivery: toolDelivery{ask: stdio.Ask}, Vault: toolVault{ask: stdio.Ask, recorder: recorder}} + // The module's one held writer (design 47 G §689, A-16, A-29, A-32): the renames' and retirements' findings and the views reader's go through it. + w, err := views.NewStdioHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: Module, State: stdio.State(FindingsState), + Kinds: []string{FindingRenameRefused, FindingWindowHeld, FindingRetireUnanswered, FindingRetireHeld}}) + if err != nil { + logf("%v", err) + os.Exit(1) + } + c.Findings = w moved := make(chan struct{}, 1) theViews = newOwnerViews(stdio.Ask, func() { select { case moved <- struct{}{}: default: } - }, meshconditionsclient.FindingWriter{State: stdio.State(FindingsState)}) + }, w) + // the runtime answers the writer's subscription only once Serve has its channel + go func() { + <-stdio.Served() + if err := w.Start(context.Background()); err != nil { + logf("%v", err) + os.Exit(1) + } + }() theViews.Start(context.Background()) go resolving(c, moved) go hearing(c) diff --git a/modules/mesh-config/cmd/mesh-config/manifest_test.go b/modules/mesh-config/cmd/mesh-config/manifest_test.go index 9a0d91269..79b0f0144 100644 --- a/modules/mesh-config/cmd/mesh-config/manifest_test.go +++ b/modules/mesh-config/cmd/mesh-config/manifest_test.go @@ -63,7 +63,9 @@ func TestEveryStateAndActHasItsTool(t *testing.T) { if !slices.Equal(dm.Emits, []string{ConfigChanged}) { t.Errorf("emits %v, want only %s: the stream carries the others", dm.Emits, ConfigChanged) } - if len(dm.Streams) != 1 || dm.Streams[0].Name != StreamName || !dm.Streams[0].KeepsHistory() || !dm.Streams[0].Carries(EventRenamed) || + // beside the findings' stream of a foundation part, `-findings` (design 47 G A-30, §529) + if len(dm.Streams) != 2 || dm.Streams[1].Name != Module+"-findings" || !dm.Streams[1].Carries("finding-written.x") || + dm.Streams[0].Name != StreamName || !dm.Streams[0].KeepsHistory() || !dm.Streams[0].Carries(EventRenamed) || !dm.Streams[0].Carries(EventRetiring) || !dm.Streams[0].Carries(EventRetired) || dm.Streams[0].Carries(ConfigChanged) { t.Errorf("streams %+v: want %s, keeping its history, carrying renamed, retiring and retired but not config-changed", dm.Streams, StreamName) } diff --git a/modules/mesh-config/cmd/mesh-config/ownerviews.go b/modules/mesh-config/cmd/mesh-config/ownerviews.go index 1eb15cf1f..623739cda 100644 --- a/modules/mesh-config/cmd/mesh-config/ownerviews.go +++ b/modules/mesh-config/cmd/mesh-config/ownerviews.go @@ -18,10 +18,8 @@ import ( catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" "sort" "strings" - "time" stdio "git.novox.be/novox/mesh-sdk/go" - "git.novox.be/novox/mesh-sdk/go/views" ) @@ -49,83 +47,46 @@ var followed = []struct{ Read, Owner, State, Tool string }{ {WantsRead, assignerModule, "wants", AssignerWantsTool}, } -// ownerViews are the pictures, by the read name. +// ownerViews are the pictures, by the read name, followed through one views.Reader holding the module's writer (design 47 G +// A-28): the reader raises and clears the first views' findings and hears each owner's views event, and asks the first view +// again on a gap. type ownerViews struct { - by map[string]*views.Follower - // findings keeps the findings a first view unanswered writes. - findings views.FindingWriter + by map[string]*views.Follower + reader *views.Reader } -// newOwnerViews makes a follower for each state; changed is told when any picture changes. -func newOwnerViews(ask func(key string, body any) (json.RawMessage, error), changed func(), findings views.FindingWriter) *ownerViews { - o := &ownerViews{by: map[string]*views.Follower{}, findings: findings} +// newOwnerViews makes a follower for each state; changed is told when any picture changes. findings is the module's writer, +// which the reader writes the first views' findings through. +func newOwnerViews(ask func(key string, body any) (json.RawMessage, error), changed func(), findings views.WireWriter) *ownerViews { + o := &ownerViews{by: map[string]*views.Follower{}} + o.reader = &views.Reader{Module: Module, Findings: findings, Logf: logf, + Changed: func(string, string, json.RawMessage, bool) { changed() }} for _, f := range followed { f := f - fol := &views.Follower{Owner: f.Owner, Tool: f.Owner + "." + f.Tool, State: f.State, Logf: logf, - Ask: func(_ context.Context, after string) (views.Answer, error) { - body := map[string]any{"values": true} - if after != "" { - body["after"] = after - } - raw, err := ask(f.Owner+"."+f.Tool, body) - if err != nil { - return views.Answer{}, err - } - answer, output, ok := unwrapAnswer(raw) - if !ok { - return views.Answer{}, fmt.Errorf("%s refused: %s", f.Tool, output) - } - return views.DecodeAnswer(answer) - }, - Changed: func(string, json.RawMessage, bool, time.Time) { changed() }} - fol.Raise = func(s views.Status) { - _ = findings.Put(views.UnansweredFinding(Module, s, time.Now())) - } - fol.Clear = func(s views.Status) { - _ = findings.Delete(views.UnansweredFinding(Module, s, time.Now())) + fol := o.reader.Follow(f.Owner, f.State, f.Owner+"."+f.Tool) + // the owner's ordinary read tool answers wrapped as a tool's answer, or refused with its output + fol.Ask = func(_ context.Context, after string) (views.Answer, error) { + body := map[string]any{"values": true} + if after != "" { + body["after"] = after + } + raw, err := ask(f.Owner+"."+f.Tool, body) + if err != nil { + return views.Answer{}, err + } + answer, output, ok := unwrapAnswer(raw) + if !ok { + return views.Answer{}, fmt.Errorf("%s refused: %s", f.Tool, output) + } + return views.DecodeAnswer(answer) } o.by[f.Read] = fol } return o } -// Start asks every first view until answered (each on its goroutine) and subscribes to the owners' views events. -func (o *ownerViews) Start(ctx context.Context) { - for _, fol := range o.by { - go func(f *views.Follower) { - if err := f.Start(ctx); err != nil { - logf("[mesh-config] the first view of %s's %s ended unanswered: %v", f.Owner, f.State, err) - } - }(fol) - } - for _, f := range followed { - f := f - pattern := f.Owner + "." + views.EventName(f.State) - go func() { - <-stdio.Served() - for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - err := stdio.Subscribe(pattern, func(e stdio.Envelope) error { - ev, err := views.DecodeEvent(e.Body) - if err != nil { - logf("[mesh-config] %s is not a views event: %v", pattern, err) - return nil - } - fol := o.by[f.Read] - if gap, why := fol.Apply(ev); gap { - logf("[mesh-config] a gap in %s's %s (%s): its first view is asked once more", f.Owner, f.State, why) - go func() { _ = fol.Start(ctx) }() - } - return nil - }) - if err == nil { - return - } - logf("[mesh-config] does not hear %s yet (%v); asking again in %s", pattern, err, wait) - time.Sleep(wait) - } - }() - } -} +// Start subscribes the owners' views events and asks every first view until answered, each on its goroutine. +func (o *ownerViews) Start(ctx context.Context) { o.reader.Start(ctx) } // read is the picture of a read name, and whether the name is one followed. func (o *ownerViews) read(state string) (*views.Follower, bool) { diff --git a/modules/mesh-config/cmd/mesh-config/ownerviews_test.go b/modules/mesh-config/cmd/mesh-config/ownerviews_test.go index d7611930e..9b0579144 100644 --- a/modules/mesh-config/cmd/mesh-config/ownerviews_test.go +++ b/modules/mesh-config/cmd/mesh-config/ownerviews_test.go @@ -36,6 +36,9 @@ func TestAFollowedStateIsReadFromItsOwnersFirstViewAndEvents(t *testing.T) { if _, followed, err := o.get(WantsRead, "want.g14.dunst"); !followed || err == nil || !strings.Contains(err.Error(), "not answered yet") { t.Fatalf("before the first view: followed %v, %v", followed, err) } + if !o.reader.Follows("mesh-delivery-assigner.wants") { + t.Error("the wants view is not one of the reader's (A-28)") + } if _, followed, _ := o.get(BuildsRead, "x"); followed { t.Error("the builder's builds is followed, and it has no owner tool") } @@ -78,5 +81,6 @@ func TestAFollowedStateIsReadFromItsOwnersFirstViewAndEvents(t *testing.T) { // nopFindings takes a finding and keeps none. type nopFindings struct{} -func (nopFindings) Put(views.Finding) error { return nil } -func (nopFindings) Delete(views.Finding) error { return nil } +func (nopFindings) WriteWire(context.Context, map[string]any) error { return nil } +func (nopFindings) Clear(_ context.Context, _, _, _, _ string) error { return nil } +func (nopFindings) ModuleNode() (string, string) { return Module, "" } diff --git a/modules/mesh-config/cmd/mesh-config/renames.go b/modules/mesh-config/cmd/mesh-config/renames.go index a22de1506..f7338a38a 100644 --- a/modules/mesh-config/cmd/mesh-config/renames.go +++ b/modules/mesh-config/cmd/mesh-config/renames.go @@ -155,7 +155,7 @@ func (c *Config) rename(s *Sources, held map[string]Rename, trunk ModuleDeclarat nodes := nodesPlaced(s, module) var said []string refuse := func(why string) []string { - c.setFinding(renameRefused(module, to, from, why, now)) + c.setFinding(module, to, renameRefused(module, to, from, why, now)) if r, ok := held[key]; ok { s.Renames[key] = r } @@ -488,7 +488,7 @@ func (c *Config) sweepRenameFindings(s *Sources, now time.Time) { } c.clearFinding(FindingWindowHeld, r.Module, r.To, held...) for _, node := range held { - c.setFinding(windowHeld(r, node, now)) + c.setFinding(r.Module, r.To, windowHeld(r, node, now)) } } } diff --git a/modules/mesh-config/cmd/mesh-config/renames_test.go b/modules/mesh-config/cmd/mesh-config/renames_test.go index e4b5d3a27..4655fa358 100644 --- a/modules/mesh-config/cmd/mesh-config/renames_test.go +++ b/modules/mesh-config/cmd/mesh-config/renames_test.go @@ -1,6 +1,7 @@ package main import ( + "context" "encoding/json" "errors" "fmt" @@ -12,6 +13,7 @@ import ( meshconditions "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" ) // fakeDelivery is the delivery parts as mesh-config asks them: each node's have, whether a node-engine states what it @@ -78,20 +80,20 @@ func aRenamingMesh(t *testing.T) (*Config, *memStore, *fakeDelivery, *time.Time) } clock := time.Date(2026, 11, 4, 10, 0, 0, 0, time.UTC) d := &fakeDelivery{haves: map[string]string{"g14.player": fp1, "novox.player": fp1}, notSays: map[string]bool{}} - c := &Config{Store: m, Now: func() time.Time { return clock }, Logf: t.Logf, Delivery: d} + c := &Config{Store: m, Now: func() time.Time { return clock }, Logf: t.Logf, Delivery: d, Findings: m} if _, err := c.Reconcile(); err != nil { t.Fatal(err) } return c, m, d, &clock } -func findingsOf(t *testing.T, m *memStore, kind string) []meshconditions.FindingsByKeyValue { +func findingsOf(t *testing.T, m *memStore, kind string) []meshconditions.Finding { t.Helper() - var out []meshconditions.FindingsByKeyValue + var out []meshconditions.Finding for _, k := range m.keysOf(FindingsState) { e, _ := m.Get(FindingsState, k) - f, err := meshconditions.DecodeFindingsByKeyValue(e.Value) - if err != nil { + var f meshconditions.Finding + if err := json.Unmarshal(e.Value, &f); err != nil { t.Fatalf("finding %s: %v", k, err) } if f.Kind == kind { @@ -272,6 +274,8 @@ func TestAWindowHeldSevenDaysIsAFindingAndAHandCloseIsAHandAct(t *testing.T) { t.Error("a node holding no open window was dropped") } m.handActs = nil + // A restart in between: the module holds no finding of its own, the writer holds it (design 47 G A-32, A-38). + c = &Config{Store: c.Store, Now: c.Now, Logf: c.Logf, Delivery: c.Delivery, Findings: c.Findings} if _, err := c.RenameClose("player", "level", "novox", "novox is retired for good"); err != nil { t.Fatal(err) } @@ -435,3 +439,25 @@ func TestARetirementDeclinedUnansweredCancelledOrASecret(t *testing.T) { } _ = fmt.Sprint } + +// Write and Clear are the module's finding writer in a test: a finding is kept in its key and deleted as views.Writer does. +func (m *memStore) Write(_ context.Context, f meshconditions.Finding) error { + return m.Put(FindingsState, views.FindingKey(f.Scope, f.ID, f.Kind), f) +} + +// Held is what the writer holds: the findings in the module's state. +func (m *memStore) Held() []meshconditions.Finding { + var out []meshconditions.Finding + for _, k := range m.keysOf(FindingsState) { + e, _ := m.Get(FindingsState, k) + var f meshconditions.Finding + if e != nil && json.Unmarshal(e.Value, &f) == nil { + out = append(out, f) + } + } + return out +} + +func (m *memStore) Clear(_ context.Context, scope, id, kind, _ string) error { + return m.Delete(FindingsState, views.FindingKey(scope, id, kind)) +} diff --git a/modules/mesh-config/cmd/mesh-config/retiring.go b/modules/mesh-config/cmd/mesh-config/retiring.go index dc010dd74..6f6bd2c16 100644 --- a/modules/mesh-config/cmd/mesh-config/retiring.go +++ b/modules/mesh-config/cmd/mesh-config/retiring.go @@ -235,7 +235,7 @@ func (c *Config) advanceRetiring(s Sources, e *Retiring) []string { return []string{fmt.Sprintf("the retirement %s is not kept: %v", key, err)} } if now.Sub(e.Since) >= HeldFinding { - c.setFinding(retirementHeld(*e, now)) + c.setFinding(e.Module, e.Key, retirementHeld(*e, now)) } return []string{fmt.Sprintf("%s's %s is not read by any manifest, and %s still run a build that reads it", e.Module, e.Key, strings.Join(e.Nodes, ", "))} @@ -410,7 +410,7 @@ func (c *Config) decideRetirement(e *Retiring, now time.Time) []string { if w.Outcome != occ.OutcomeChosen { out := keep(RetireUnanswered, "the ask "+string(w.Outcome)+"; nothing is removed") if e.AskedAgain { - c.setFinding(retireUnanswered(*e, now)) + c.setFinding(e.Module, e.Key, retireUnanswered(*e, now)) } return out } diff --git a/modules/mesh-config/cmd/mesh-config/whole_integration_test.go b/modules/mesh-config/cmd/mesh-config/whole_integration_test.go index a8d253d6e..407535096 100644 --- a/modules/mesh-config/cmd/mesh-config/whole_integration_test.go +++ b/modules/mesh-config/cmd/mesh-config/whole_integration_test.go @@ -726,7 +726,8 @@ func TestConfigWholeRenamesAndRetiresAgainstTheBus(t *testing.T) { keys, _ := (func() ([]string, error) { kv, _ := p.kv(FindingsState); return kv.Keys(p.ctx) })() for _, k := range keys { raw, _ := p.get(FindingsState, k) - if f, err := meshconditions.DecodeFindingsByKeyValue(raw); err == nil && f.Kind == FindingRenameRefused && strings.Contains(f.Summary, "dock") { + var f meshconditions.Finding + if err := json.Unmarshal(raw, &f); err == nil && f.Kind == FindingRenameRefused && strings.Contains(f.Summary, "dock") { return true } } diff --git a/modules/mesh-config/go.mod b/modules/mesh-config/go.mod index fc6a03485..ced63103d 100644 --- a/modules/mesh-config/go.mod +++ b/modules/mesh-config/go.mod @@ -9,7 +9,6 @@ require ( git.novox.be/novox/mesh-contracts/go/seats/meshconfig v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-config/module.json b/modules/mesh-config/module.json index 5e65b3fc8..b139a8421 100644 --- a/modules/mesh-config/module.json +++ b/modules/mesh-config/module.json @@ -33,7 +33,8 @@ "mesh-delivery-indexer.views.predefined", "mesh-delivery-indexer.views.config-candidates", "mesh-delivery-assigner.views.assignments", - "mesh-delivery-assigner.views.wants" + "mesh-delivery-assigner.views.wants", + "mesh-conditions.findings-asked" ], "invokes": [ "seat:mesh-hand-acts.hand-act", @@ -123,6 +124,16 @@ "retired" ], "retention": "history" + }, + { + "name": "mesh-config-findings", + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 } ] } diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go index 68708baa2..83f513caf 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/main.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/main.go @@ -16,7 +16,7 @@ import ( "sync/atomic" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" @@ -185,7 +185,7 @@ const ( // newViews is the provisioner's pictures of the assigner's assignments and wants and the indexer's modules and manifests: // each asked once of its owner's read tool, then followed by the owner's events (ADR 0615). func newViews() *views.Reader { - v := &views.Reader{Module: Module, Logf: logf, Findings: meshconditionsclient.FindingWriter{State: stdio.State("findings")}, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}} + v := &views.Reader{Module: Module, Logf: logf, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}} v.Follow(assignerModule, "assignments", AssignmentsTool) v.Follow(assignerModule, "wants", WantsTool) v.Follow(indexerModule, "modules", ModulesTool) @@ -215,9 +215,25 @@ func watchedOver(v *views.Reader) func(string, func(stdio.StateChange) error) er } } +// newFindingsWriter is this module's finding writer (design 47 G §689, A-16): it states the findings of its views reader and +// observes none of its own. +func newFindingsWriter() (*views.Writer[seat.Finding, seat.FindingCleared, seat.FindingsStated, seat.FindingsAskedEvent], error) { + return views.NewStdioWriter(views.WriterConfig[seat.Finding, seat.FindingCleared, seat.FindingsStated, seat.FindingsAskedEvent]{ + Asked: seat.SubjectFindingsAsked, Module: "mesh-provisioner", State: stdio.State("findings"), + Observe: func(context.Context) (views.Observed[seat.Finding], error) { + return views.Observed[seat.Finding]{}, nil + }}) +} + func main() { s := runtimeStore{} v := newViews() + findings, err := newFindingsWriter() + if err != nil { + logf("%v", err) + os.Exit(1) + } + v.Findings = findings // One copy of what it reads, each state by its watch (ADR 0580 clause 2); the indexer's and the assigner's by their // pictures, a manifest by its digest from the picture too. ws := stdio.WatchedStatesOver(watchedOver(v), func(state, key string) ([]byte, bool, error) { @@ -230,6 +246,14 @@ func main() { } return e.Value, true, nil }) + // the runtime answers the writer's subscription only once Serve has its channel + go func() { + <-stdio.Served() + if err := findings.Start(context.Background()); err != nil { + logf("%v", err) + os.Exit(1) + } + }() v.Start(context.Background()) h := &Holder{Store: Watching(s, ws), Asks: runtimeAsks{}, Now: time.Now, Logf: logf} h.Compose = FromCompositions(h.Store) // the node-engines' published compositions, from the copy (ADR 0593 §8) diff --git a/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go b/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go index 0b1837325..fd6cbc2d8 100644 --- a/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go +++ b/modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go @@ -11,7 +11,6 @@ import ( "sync" "testing" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -291,6 +290,12 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) { if err != nil { t.Fatal(err) } + // the writer's three events are declared (design 47 G A-30) + for _, ev := range []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"} { + if !strings.Contains(string(raw), `"`+ev+`"`) { + t.Errorf("the manifest does not declare %s", ev) + } + } var m struct { Module string `json:"module"` RunsAs string `json:"runs-as"` @@ -322,7 +327,7 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) { t.Errorf("the claim: %+v", m.Claims) } // needs stays declared this release so its first start can empty it (ADR 0570 clause 1). - if want := []string{Registry, Needs, Applied, Collect, ProvisionsState, PlansState, RefusedState, meshconditionsclient.FindingsState}; !slices.Equal(m.State, want) { + if want := []string{Registry, Needs, Applied, Collect, ProvisionsState, PlansState, RefusedState, "findings"}; !slices.Equal(m.State, want) { t.Errorf("the state is %v, want %v", m.State, want) } // Every state and act has its tool (ADR 0570 Consequences): the verb reads the clients' states, the holder's tools @@ -365,7 +370,7 @@ func TestTheManifestDeclaresWhatTheCodeKeeps(t *testing.T) { } if !slices.Equal(m.Invokes, []string{AssignmentsTool, WantsTool, ModulesTool, ManifestsTool, "seat:mesh-hand-acts.hand-act"}) || !slices.Equal(m.Consumes, []string{"mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants", - "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests"}) { + "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.manifests", "mesh-conditions.findings-asked"}) { t.Errorf("invokes %v, consumes %v", m.Invokes, m.Consumes) } // It composes nothing (ADR 0593 §8): no config value is read to make a plan. diff --git a/modules/mesh-provisioner/go.mod b/modules/mesh-provisioner/go.mod index 00b8dd54e..f02f70bb8 100644 --- a/modules/mesh-provisioner/go.mod +++ b/modules/mesh-provisioner/go.mod @@ -6,7 +6,6 @@ require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000 @@ -36,7 +35,7 @@ require ( require ( git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 // indirect - git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-delivery/contracts/go/clients/containerimageregistryclient v0.0.0-00010101000000-000000000000 // indirect git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry v0.0.0-00010101000000-000000000000 // indirect diff --git a/modules/mesh-provisioner/module.json b/modules/mesh-provisioner/module.json index c35a14b64..fb3326069 100644 --- a/modules/mesh-provisioner/module.json +++ b/modules/mesh-provisioner/module.json @@ -40,7 +40,8 @@ "mesh-delivery-assigner.views.assignments", "mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.views.modules", - "mesh-delivery-indexer.views.manifests" + "mesh-delivery-indexer.views.manifests", + "mesh-conditions.findings-asked" ], "tools": [ "provisioner_provisions", @@ -105,5 +106,21 @@ ] } ] - } + }, + "requires": [ + { + "name": "finding-events", + "storage": "stream", + "template": "events", + "options": { + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 + } + } + ] } diff --git a/modules/mesh-vault/cmd/mesh-vault/assignmentsview.go b/modules/mesh-vault/cmd/mesh-vault/assignmentsview.go index 644029f0f..90e1094e4 100644 --- a/modules/mesh-vault/cmd/mesh-vault/assignmentsview.go +++ b/modules/mesh-vault/cmd/mesh-vault/assignmentsview.go @@ -12,9 +12,6 @@ import ( "fmt" "sort" "strings" - "time" - - stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/views" ) @@ -39,33 +36,12 @@ var assignmentsView *views.Follower // newAssignmentsView makes the follower: ask calls the tool, gone is told of a deleted key, changed of every change after // it is held (a first view's keys too). func newAssignmentsView(ask func(key string, body any) (json.RawMessage, error), gone func(key string), changed func(), - findings views.FindingWriter) *views.Follower { - return &views.Follower{Owner: assignerModule, Tool: AssignmentsTool, State: "assignments", Logf: say, - Ask: func(_ context.Context, after string) (views.Answer, error) { - body := map[string]any{"values": true} - if after != "" { - body["after"] = after - } - raw, err := ask(AssignmentsTool, body) - if err != nil { - return views.Answer{}, err - } - return views.DecodeAnswer(unwrapToolAnswer(raw)) - }, - // Past its bound the vault's finding names the owner and the tool, and the condition keeper raises it (the - // answer deletes it); the vault keeps asking meanwhile. - Raise: func(s views.Status) { - say("the vault has no first view of the assigner's assignments: %s has not answered since %s (%s); it keeps asking", s.Tool, - s.Asked.Format(time.RFC3339), s.Error) - f := views.UnansweredFinding(Seat, s, time.Now()) - if err := findings.Put(f); err != nil { - say("the finding %s is not written: %v", f.ID, err) - } - }, - Clear: func(s views.Status) { - _ = findings.Delete(views.UnansweredFinding(Seat, s, time.Now())) - }, - Changed: func(key string, _ json.RawMessage, deleted bool, _ time.Time) { + findings views.WireWriter) (*views.Follower, *views.Reader) { + // A views.Reader holding the vault's writer follows it (design 47 G A-28), so the reader writes and clears its findings + // (past its bound the finding names the owner and the tool, and the answer clears it; no secret) and hears the owner's + // views event, asking again on a gap. + r := &views.Reader{Module: Seat, Findings: findings, Logf: say, + Changed: func(_, key string, _ json.RawMessage, deleted bool) { if deleted { gone(key) } @@ -73,6 +49,19 @@ func newAssignmentsView(ask func(key string, body any) (json.RawMessage, error), changed() } }} + f := r.Follow(assignerModule, "assignments", AssignmentsTool) + f.Ask = func(_ context.Context, after string) (views.Answer, error) { + body := map[string]any{"values": true} + if after != "" { + body["after"] = after + } + raw, err := ask(AssignmentsTool, body) + if err != nil { + return views.Answer{}, err + } + return views.DecodeAnswer(unwrapToolAnswer(raw)) + } + return f, r } // unwrapToolAnswer takes a tool's answer out of the protocol's content list when it comes in one. @@ -88,37 +77,6 @@ func unwrapToolAnswer(raw json.RawMessage) []byte { return raw } -// beginAssignmentsView asks the first view until answered and hears the owner's events. -func beginAssignmentsView(f *views.Follower) { - go func() { - if err := f.Start(context.Background()); err != nil { - say("the first view of the assigner's assignments ended unanswered: %v", err) - } - }() - go func() { - <-stdio.Served() - for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { - err := stdio.Subscribe(AssignmentsViewsEvent, func(e stdio.Envelope) error { - ev, err := views.DecodeEvent(e.Body) - if err != nil { - say("%s is not a views event: %v", AssignmentsViewsEvent, err) - return nil - } - if gap, why := f.Apply(ev); gap { - say("a gap in the assigner's assignments (%s): its first view is asked once more", why) - go func() { _ = f.Start(context.Background()) }() - } - return nil - }) - if err == nil { - return - } - say("does not hear %s yet (%v); asking again in %s", AssignmentsViewsEvent, err, wait) - time.Sleep(wait) - } - }() -} - // viewAssignments is the assignments reader over the picture: a key asked before the first view is answered is an error, so // the caller waits (the "not yet" of the vault's other reads), never a read of the owner's bucket. func viewAssignments(f func() *views.Follower) AssignerReader { diff --git a/modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go b/modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go index f259e36e1..ae5118c49 100644 --- a/modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go +++ b/modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "sync" "testing" "time" @@ -28,7 +29,7 @@ func TestAnUnansweredFirstViewOfTheAssignmentsIsAFindingTheAnswerClears(t *testi Values: map[string]json.RawMessage{"g14.node": json.RawMessage(`{}`)}}) return raw, nil } - f := newAssignmentsView(ask, func(string) {}, nil, heldFindings{&mu, held}) + f, _ := newAssignmentsView(ask, func(string) {}, nil, heldFindings{&mu, held}) f.Bound, f.Wait = 30*time.Millisecond, 10*time.Millisecond reader := viewAssignments(func() *views.Follower { return f }) if _, err := reader.Keys(); err == nil { @@ -62,15 +63,16 @@ type heldFindings struct { held map[string]json.RawMessage } -func (h heldFindings) Put(f views.Finding) error { +func (h heldFindings) WriteWire(_ context.Context, f map[string]any) error { h.mu.Lock() defer h.mu.Unlock() - h.held[f.ID+"."+f.Kind] = json.RawMessage(f.Summary) + h.held[fmt.Sprint(f["id"], ".", f["kind"])] = nil return nil } -func (h heldFindings) Delete(f views.Finding) error { +func (h heldFindings) Clear(_ context.Context, _, id, kind, _ string) error { h.mu.Lock() defer h.mu.Unlock() - delete(h.held, f.ID+"."+f.Kind) + delete(h.held, id+"."+kind) return nil } +func (h heldFindings) ModuleNode() (string, string) { return Seat, "" } diff --git a/modules/mesh-vault/cmd/mesh-vault/main.go b/modules/mesh-vault/cmd/mesh-vault/main.go index c6dc697f9..c113a05b9 100644 --- a/modules/mesh-vault/cmd/mesh-vault/main.go +++ b/modules/mesh-vault/cmd/mesh-vault/main.go @@ -8,8 +8,9 @@ package main import ( - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + meshconditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "context" "encoding/json" "errors" @@ -118,13 +119,26 @@ func watchLeavers(env *Env) bool { return true } if assignmentsView == nil { - assignmentsView = newAssignmentsView(stdio.Ask, func(key string) { + // The vault's finding writer (design 47 G §689, A-16): it states the assignments view's findings and observes none of + // its own, and names no secret. + w, err := views.NewStdioWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: Seat, State: stdio.State(FindingsState), + Observe: func(context.Context) (views.Observed[meshconditions.Finding], error) { + return views.Observed[meshconditions.Finding]{}, nil + }}) + if err == nil { + err = w.Start(context.Background()) + } + if err != nil { + return false + } + f, r := newAssignmentsView(stdio.Ask, func(key string) { if node, ok := leaverKey(key); ok { env.Vault.NodeLeft(node) } - }, env.Vault.AssignmentsChanged, - meshconditionsclient.FindingWriter{State: stdio.State(FindingsState)}) - beginAssignmentsView(assignmentsView) + }, env.Vault.AssignmentsChanged, w) + assignmentsView = f + r.Start(context.Background()) } return true } diff --git a/modules/mesh-vault/cmd/mesh-vault/vault_test.go b/modules/mesh-vault/cmd/mesh-vault/vault_test.go index 68def3bd5..d953d15db 100644 --- a/modules/mesh-vault/cmd/mesh-vault/vault_test.go +++ b/modules/mesh-vault/cmd/mesh-vault/vault_test.go @@ -198,7 +198,7 @@ func TestNoBucketHoldsASecret(t *testing.T) { code = []byte(strings.Replace(string(code), "stdio.State(WantedRead).Watch(", "", 1)) code = []byte(strings.Replace(string(code), "stdio.State(WantedRead).Get(", "", 1)) // And its findings (ADR 0615 §6): the finding of a first view unanswered, written and deleted, never a secret. - code = []byte(strings.Replace(string(code), "meshconditionsclient.FindingWriter{State: stdio.State(FindingsState)}", "", 1)) + code = []byte(strings.Replace(string(code), "State: stdio.State(FindingsState)", "", 1)) // And its outcome entries (novox/hq ADR 0570 clauses 3, 6): a value's name and generation, never a value; and // one read of the seat's provisions, which hold none. for _, call := range []string{"Get(", "Put(", "Delete(", "Keys("} { diff --git a/modules/mesh-vault/go.mod b/modules/mesh-vault/go.mod index b09f9aeb8..b9bd08189 100644 --- a/modules/mesh-vault/go.mod +++ b/modules/mesh-vault/go.mod @@ -38,11 +38,10 @@ require ( ) require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-networking/modules/mesh-wireguard/go v0.0.0-00010101000000-000000000000 ) require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-networking/modules/mesh-connectivity-contracts/go/seats/theprivatenetwork v0.0.0-00010101000000-000000000000 // indirect ) diff --git a/modules/mesh-watcher/cmd/mesh-watcher/main.go b/modules/mesh-watcher/cmd/mesh-watcher/main.go index ecded8338..9e659b913 100644 --- a/modules/mesh-watcher/cmd/mesh-watcher/main.go +++ b/modules/mesh-watcher/cmd/mesh-watcher/main.go @@ -17,8 +17,8 @@ import ( "sync" "time" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" meshselfcheck "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshselfcheck" stdio "git.novox.be/novox/mesh-sdk/go" @@ -76,8 +76,22 @@ func main() { node := os.Getenv("MESH_NODE") w := NewWatcher(tg, time.Now, logf, node) l := &listening{now: "not yet: starting"} - assignments := &views.Reader{Module: "mesh-watcher", Logf: logf, Findings: meshconditionsclient.FindingWriter{State: stdio.State("findings")}, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}} + assignments := &views.Reader{Module: "mesh-watcher", Logf: logf, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}} + findings, err := newFindingsWriter() + if err != nil { + logf("%v", err) + os.Exit(1) + } + assignments.Findings = findings assignments.Follow(assignerModule, mesh.StateSeatHolders, SeatHoldersTool) + // the runtime answers the writer's subscription only once Serve has its channel + go func() { + <-stdio.Served() + if err := findings.Start(context.Background()); err != nil { + logf("%v", err) + os.Exit(1) + } + }() go run(w, l, node, assignments) if err := stdio.Serve("", append(tools(w, l), assignments.Tools("watcher")...)); err != nil { logf("%v", err) @@ -85,6 +99,16 @@ func main() { } } +// newFindingsWriter is this module's finding writer (design 47 G §689, A-16): it states the findings of its views reader and +// observes none of its own. +func newFindingsWriter() (*views.Writer[seat.Finding, seat.FindingCleared, seat.FindingsStated, seat.FindingsAskedEvent], error) { + return views.NewStdioWriter(views.WriterConfig[seat.Finding, seat.FindingCleared, seat.FindingsStated, seat.FindingsAskedEvent]{ + Asked: seat.SubjectFindingsAsked, Module: "mesh-watcher", State: stdio.State("findings"), + Observe: func(context.Context) (views.Observed[seat.Finding], error) { + return views.Observed[seat.Finding]{}, nil + }}) +} + // run keeps time, makes the round trips, reads where the controller is, and listens for heartbeats. // The clock runs whatever the bus does: it is what notices the bus is gone. func run(w *Watcher, l *listening, node string, assignments *views.Reader) { diff --git a/modules/mesh-watcher/go.mod b/modules/mesh-watcher/go.mod index 4453cc660..0970045fa 100644 --- a/modules/mesh-watcher/go.mod +++ b/modules/mesh-watcher/go.mod @@ -7,7 +7,6 @@ require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 github.com/klauspost/compress v1.20.1 // indirect github.com/nats-io/nats.go v1.54.0 // indirect @@ -20,6 +19,6 @@ require ( require git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshselfcheck v0.0.0-00010101000000-000000000000 require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect ) diff --git a/modules/minio/cmd/object-store/main.go b/modules/minio/cmd/object-store/main.go index 575a383d9..e043b6639 100644 --- a/modules/minio/cmd/object-store/main.go +++ b/modules/minio/cmd/object-store/main.go @@ -15,7 +15,7 @@ import ( mesh "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "os" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" @@ -30,7 +30,13 @@ const ( ) func main() { - seats := &views.Reader{Module: "minio", Findings: meshconditionsclient.FindingWriter{State: stdio.State("findings")}, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}, Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[object-store] "+f+"\n", a...) }} + seats := &views.Reader{Module: "minio", Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}, Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[object-store] "+f+"\n", a...) }} + findings, err := newFindingsWriter() + if err != nil { + fmt.Fprintf(os.Stderr, "[object-store] %v\n", err) + os.Exit(1) + } + seats.Findings = findings seats.Follow(assignerModule, mesh.StateSeatHolders, SeatHoldersTool) s, err := newServer(os.Getenv(ConfigWord), os.Getenv(LinksWord), os.Getenv(ReceivesWord), func() (map[string][]byte, error) { return seats.Values(assignerModule + "." + mesh.StateSeatHolders) }) @@ -38,6 +44,14 @@ func main() { fmt.Fprintf(os.Stderr, "[object-store] %v\n", err) os.Exit(1) } + // the runtime answers the writer's subscription only once Serve has its channel + go func() { + <-stdio.Served() + if err := findings.Start(context.Background()); err != nil { + fmt.Fprintf(os.Stderr, "[object-store] %v\n", err) + os.Exit(1) + } + }() seats.Start(context.Background()) if err := stdio.Serve("", append(tools(s), seats.Tools("object_store")...)); err != nil { fmt.Fprintf(os.Stderr, "[object-store] %v\n", err) @@ -52,6 +66,16 @@ const assignerModule = "mesh-delivery-assigner" const SeatHoldersTool = assignerModule + ".assigner_read_seat_holders" +// newFindingsWriter is this module's finding writer (design 47 G §689, A-16): it states the findings of its views reader and +// observes none of its own. +func newFindingsWriter() (*views.Writer[seat.Finding, seat.FindingCleared, seat.FindingsStated, seat.FindingsAskedEvent], error) { + return views.NewStdioWriter(views.WriterConfig[seat.Finding, seat.FindingCleared, seat.FindingsStated, seat.FindingsAskedEvent]{ + Asked: seat.SubjectFindingsAsked, Module: "minio", State: stdio.State("findings"), + Observe: func(context.Context) (views.Observed[seat.Finding], error) { + return views.Observed[seat.Finding]{}, nil + }}) +} + func tools(s *server) []stdio.Tool { return []stdio.Tool{{ Name: "object-store.link", diff --git a/modules/minio/go.mod b/modules/minio/go.mod index d398b5ea2..390a722a6 100644 --- a/modules/minio/go.mod +++ b/modules/minio/go.mod @@ -7,7 +7,6 @@ require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 github.com/klauspost/compress v1.20.1 // indirect github.com/nats-io/nats.go v1.54.0 // indirect @@ -18,6 +17,6 @@ require ( ) require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect ) diff --git a/modules/nfs-server/cmd/nfs-server/main.go b/modules/nfs-server/cmd/nfs-server/main.go index 153d5c186..6b4d106df 100644 --- a/modules/nfs-server/cmd/nfs-server/main.go +++ b/modules/nfs-server/cmd/nfs-server/main.go @@ -22,8 +22,8 @@ import ( "time" "git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings" - "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + seat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/views" @@ -51,8 +51,26 @@ func main() { return } m := Here() + w, err := views.NewStdioWriter(views.WriterConfig[seat.Finding, seat.FindingCleared, seat.FindingsStated, seat.FindingsAskedEvent]{ + Asked: seat.SubjectFindingsAsked, Module: "nfs-server", State: stdio.State("findings"), + Observe: func(context.Context) (views.Observed[seat.Finding], error) { + return views.Observed[seat.Finding]{}, nil + }}) + if err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + seats.Findings = w seats.Follow(assignerModule, mesh.StateSeatHolders, SeatHoldersTool) seats.Start(context.Background()) + // the runtime answers the writer's subscription only once Serve has its channel + go func() { + <-stdio.Served() + if err := w.Start(context.Background()); err != nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(1) + } + }() go ShareExports(m) if err := stdio.Serve("", append(Tools(m), seats.Tools("nfs_server")...)); err != nil { fmt.Fprintln(os.Stderr, err) @@ -132,7 +150,7 @@ const SeatHoldersTool = assignerModule + ".assigner_read_seat_holders" // seats is the picture of the assignments seat's state of who holds which seat: asked of the assigner once and followed by // its events, never read from its state (ADR 0615). -var seats = &views.Reader{Module: "nfs-server", Findings: meshconditionsclient.FindingWriter{State: stdio.State("findings")}, Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}, Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "nfs-server: "+f+"\n", a...) }} +var seats = &views.Reader{Module: "nfs-server", Acts: meshhandactsclient.Recorder{Ask: stdio.Ask}, Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "nfs-server: "+f+"\n", a...) }} // seatHolderKeys is the assignments seat's state of who holds which seat, every key. func seatHolderKeys() (map[string][]byte, error) { diff --git a/modules/nfs-server/go.mod b/modules/nfs-server/go.mod index 510814cc0..5c20efeb4 100644 --- a/modules/nfs-server/go.mod +++ b/modules/nfs-server/go.mod @@ -7,7 +7,6 @@ require git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 require ( git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000 - git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 github.com/klauspost/compress v1.20.1 // indirect github.com/nats-io/nats.go v1.54.0 // indirect @@ -22,6 +21,6 @@ require git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings v0.0.0-00 replace git.novox.be/novox/mesh-catalog/modules/disk-load/diskreadings => ../disk-load/diskreadings require ( - git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 // indirect + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshhandacts v0.0.0-00010101000000-000000000000 // indirect )