novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'feat/47-G-findings4' into feat/47-G-dev4-join

Browse Source
This commit is contained in:
jschoubben
2026-11-11 22:25:53 +01:00
parent f06dd01bfe 6a612c58cb
commit 9aae6244d5
60 changed files with 1041 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/artifact-store-tools/cmd/artifact-store-tools/holder.go
+64 -56
View File
@@ -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()}
}
modules/artifact-store-tools/cmd/artifact-store-tools/holder_integration_test.go
+1 -1
View File
@@ -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.
modules/artifact-store-tools/cmd/artifact-store-tools/holder_test.go
+47 -6
View File
@@ -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))
}
modules/artifact-store-tools/cmd/artifact-store-tools/main.go
+19 -2
View File
@@ -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.
modules/artifact-store-tools/cmd/artifact-store-tools/sweepseats.go
+12 -12
View File
@@ -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://<path>`, 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)
}
}
modules/artifact-store-tools/go.mod
-1
View File
@@ -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
modules/claude-code/cmd/claude-code/config_test.go
+1 -1
View File
@@ -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"`) ||
modules/claude-code/cmd/claude-code/main.go
+26 -2
View File
@@ -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")}
modules/claude-code/go.mod
+1 -2
View File
@@ -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
)
modules/claude-code/module.json
+16 -1
View File
@@ -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",
modules/distribution/cmd/distribution-tools/main.go
+27 -18
View File
@@ -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{},
modules/distribution/cmd/distribution/main.go
+21 -2
View File
@@ -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) {
modules/distribution/cmd/distribution/whole_integration_test.go
+6 -3
View File
@@ -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 {
modules/distribution/cmd/distribution/windows.go
-26
View File
@@ -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 {
modules/distribution/go.mod
-1
View File
@@ -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
modules/distribution/internal/holder/findings.go
+80
View File
@@ -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.<seat>.<kind>_<subject>`, 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.<seat>.<subject>.<kind>`) 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
}
modules/distribution/internal/holder/findings_test.go
+44
View File
@@ -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.<seat>.<kind>_<subject>`) 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
}
modules/distribution/internal/holder/holder_test.go
+8 -3
View File
@@ -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.
modules/distribution/module.json
+15 -2
View File
@@ -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
}
]
}
modules/gitea/cmd/package-registry/findings.go
+79
View File
@@ -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.<seat>.<kind>_<subject>`, 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.<seat>.<subject>.<kind>`) 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
}
modules/gitea/cmd/package-registry/findings_test.go
+44
View File
@@ -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.<seat>.<kind>_<subject>`) 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
}
modules/gitea/cmd/package-registry/forge.go
+2 -2
View File
@@ -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")
modules/gitea/cmd/package-registry/main.go
+25
View File
@@ -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)
modules/gitea/cmd/package-registry/seats.go
+1 -20
View File
@@ -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
}
modules/gitea/cmd/package-registry/seats_test.go
+4 -2
View File
@@ -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)
modules/gitea/cmd/package-registry/whole_integration_test.go
+81 -2
View File
@@ -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
modules/gitea/go.mod
-1
View File
@@ -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
modules/gitea/module.json
+10 -1
View File
@@ -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
}
]
}
modules/gitea/packages-account.ts
+1 -1
View File
@@ -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" }
modules/mesh-build-go/cmd/mesh-build-go/compose.go
+46 -7
View File
@@ -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"`
modules/mesh-build-go/cmd/mesh-build-go/compose_test.go
+9 -3
View File
@@ -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
modules/mesh-build-go/cmd/mesh-build-go/main.go
+14
View File
@@ -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).
modules/mesh-build-go/go.mod
+1 -1
View File
@@ -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
modules/mesh-config/cmd/mesh-config/config.go
+3
View File
@@ -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
modules/mesh-config/cmd/mesh-config/findings.go
+50 -44
View File
@@ -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
// (`<module>.<node>.<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)
}
modules/mesh-config/cmd/mesh-config/main.go
+20 -2
View File
@@ -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)
modules/mesh-config/cmd/mesh-config/manifest_test.go
+3 -1
View File
@@ -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, `<module>-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)
}
modules/mesh-config/cmd/mesh-config/ownerviews.go
+29 -68
View File
@@ -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) {
modules/mesh-config/cmd/mesh-config/ownerviews_test.go
+6 -2
View File
@@ -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, "" }
modules/mesh-config/cmd/mesh-config/renames.go
+2 -2
View File
@@ -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))
}
}
}
modules/mesh-config/cmd/mesh-config/renames_test.go
+31 -5
View File
@@ -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))
}
modules/mesh-config/cmd/mesh-config/retiring.go
+2 -2
View File
@@ -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
}
modules/mesh-config/cmd/mesh-config/whole_integration_test.go
+2 -1
View File
@@ -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
}
}
modules/mesh-config/go.mod
-1
View File
@@ -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
modules/mesh-config/module.json
+12 -1
View File
@@ -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
}
]
}
modules/mesh-provisioner/cmd/mesh-provisioner/main.go
+26 -2
View File
@@ -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)
modules/mesh-provisioner/cmd/mesh-provisioner/provisioner_test.go
+8 -3
View File
@@ -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.
modules/mesh-provisioner/go.mod
+1 -2
View File
@@ -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
modules/mesh-provisioner/module.json
+19 -2
View File
@@ -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
}
}
]
}
modules/mesh-vault/cmd/mesh-vault/assignmentsview.go
+19 -61
View File
@@ -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 {
modules/mesh-vault/cmd/mesh-vault/assignmentsview_test.go
+7 -5
View File
@@ -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, "" }
modules/mesh-vault/cmd/mesh-vault/main.go
+19 -5
View File
@@ -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
}
modules/mesh-vault/cmd/mesh-vault/vault_test.go
+1 -1
View File
@@ -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("} {
modules/mesh-vault/go.mod
+1 -2
View File
@@ -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
)
modules/mesh-watcher/cmd/mesh-watcher/main.go
+26 -2
View File
@@ -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) {
modules/mesh-watcher/go.mod
+1 -2
View File
@@ -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
)
modules/minio/cmd/object-store/main.go
+26 -2
View File
@@ -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",
modules/minio/go.mod
+1 -2
View File
@@ -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
)
modules/nfs-server/cmd/nfs-server/main.go
+20 -2
View File
@@ -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) {
modules/nfs-server/go.mod
+1 -2
View File
@@ -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
)