CAT-RD1: artifact-store-tools asks the assigner's wants and the indexer's manifests once and follows their events, no reads

This commit is contained in:
2026-11-04 12:09:12 +01:00
parent 6128a570a2
commit 2bb964b000
5 changed files with 94 additions and 22 deletions
@@ -8,10 +8,13 @@ import (
"fmt"
"os"
"strings"
"sync/atomic"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-delivery/sdk/views"
"artifactstoretools/internal/registry"
)
@@ -53,10 +56,22 @@ func DoorAt(file string) func() (Door, error) {
}
// runtimeBus is the holder's bus through the runtime: its reads, its findings and its seat's event.
type runtimeBus struct{}
type runtimeBus struct{ views *views.Reader }
func (runtimeBus) Keys(state string) ([]string, error) { return stdio.State(state).Keys() }
func (runtimeBus) Get(state, key string) ([]byte, bool, error) {
// followed says whether the state is the indexer's or the assigner's, which the holder asks once and follows by events
// (ADR 0615) rather than reads.
func (b runtimeBus) followed(state string) bool { return b.views != nil && b.views.Follows(state) }
func (b runtimeBus) Keys(state string) ([]string, error) {
if b.followed(state) {
return b.views.Keys(state)
}
return stdio.State(state).Keys()
}
func (b runtimeBus) Get(state, key string) ([]byte, bool, error) {
if b.followed(state) {
return b.views.Get(state, key)
}
e, err := stdio.State(state).Get(key)
if err != nil || e == nil {
return nil, false, err
@@ -90,7 +105,7 @@ func (runtimeBus) Publish(subject string, body any) error {
// follow watches the builder's done keys and the wants: a build's end sweeps and
// observes again; a want changed observes again. Event-driven: nothing reads them on a timer.
func follow(h *Holder) {
func follow(h *Holder, v *views.Reader) {
<-stdio.Served()
watch := func(state string, handle func(stdio.StateChange) error) {
for {
@@ -119,11 +134,22 @@ func follow(h *Holder) {
}
return nil
})
watch(WantsRead, func(c stdio.StateChange) error {
if !c.Current {
go observe()
// The wants are the assigner's: the first view then its events (ADR 0615). What Watch hands over before it returns
// is the first view, not a change.
var current atomic.Bool
current.Store(true)
for {
err := v.Watch(WantsRead, func(string, []byte, bool) {
if !current.Load() {
go observe()
}
})
if err == nil {
break
}
return nil
})
h.logf("watching %s: %v; asked again in 10s", WantsRead, err)
time.Sleep(10 * time.Second)
}
current.Store(false)
observe()
}
@@ -29,6 +29,8 @@ import (
"git.novox.be/novox/mesh-sdk/go/handacts"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-delivery/sdk/views"
"artifactstoretools/internal/registry"
)
@@ -900,6 +902,12 @@ func (h *Holder) Observe(ctx context.Context) error {
}
for _, k := range keys {
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 := health.DecodeFinding(raw); err == nil && f.Kind == views.UnansweredKind {
continue
}
}
if err := h.Bus.Delete(FindingsState, k); err != nil {
return err
}
@@ -10,12 +10,17 @@
package main
import (
"context"
"encoding/json"
"fmt"
"os"
"time"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/handacts"
"git.novox.be/novox/mesh-delivery/sdk/views"
)
func logf(format string, a ...any) { fmt.Fprintf(os.Stderr, format+"\n", a...) }
@@ -26,13 +31,32 @@ func main() {
Run: ExecRunner,
UID: os.Getuid(),
}
h := &Holder{Bus: runtimeBus{}, Door: DoorAt(os.Getenv("MESH_ARTIFACT_STORE_ADDRESS_FILE")), Listed: s.View,
v := &views.Reader{Module: "artifact-store-tools", Logf: logf, Findings: viewFindings{}}
v.Follow(delivery.AssignerModule, "wants", WantsTool)
v.Follow(delivery.IndexerModule, "manifests", ManifestsTool)
h := &Holder{Bus: runtimeBus{views: v}, Door: DoorAt(os.Getenv("MESH_ARTIFACT_STORE_ADDRESS_FILE")), Listed: s.View,
Now: time.Now, Logf: logf}
h.Acts = handacts.Recorder{Ask: stdio.Ask}
go follow(h)
v.Start(context.Background())
go follow(h, v)
// An empty name serves as the module the runtime names (MESH_SERVED_MODULE): artifact-store-tools.
if err := stdio.Serve("", append(Tools(s, h, &Catalogue{Ask: stdio.Ask}), SeatTools(h, s)...)); err != nil {
if err := stdio.Serve("", append(append(Tools(s, h, &Catalogue{Ask: stdio.Ask}), SeatTools(h, s)...), v.Tools("artifact_store")...)); err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
}
// The owners' read tools of the states the holder follows, which the manifest invokes (ADR 0615).
const (
WantsTool = delivery.AssignerModule + ".assigner_read_wants"
ManifestsTool = delivery.IndexerModule + ".indexer_read_manifests"
)
// viewFindings writes the holder's finding of a first view unanswered into its own state findings.
type viewFindings struct{}
func (viewFindings) Put(key string, value []byte) error {
_, err := stdio.State(FindingsState).Put(key, json.RawMessage(value))
return err
}
func (viewFindings) Delete(key string) error { return stdio.State(FindingsState).Delete(key) }
@@ -1,6 +1,7 @@
package main
import (
"git.novox.be/novox/mesh-delivery/sdk/views"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"context"
@@ -8,6 +9,7 @@ import (
"fmt"
"os"
"reflect"
"slices"
"sort"
"strconv"
"strings"
@@ -443,10 +445,11 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) {
t.Fatal(err)
}
var m struct {
Tools []string `json:"tools"`
Invokes []string `json:"invokes"`
Reads []string `json:"reads"`
Claims []struct {
Tools []string `json:"tools"`
Invokes []string `json:"invokes"`
Reads []string `json:"reads"`
Consumes []string `json:"consumes"`
Claims []struct {
Name string `json:"name"`
Serves []string `json:"serves"`
} `json:"claims"`
@@ -455,7 +458,7 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) {
t.Fatal(err)
}
var served []string
for _, tl := range Tools(&Store{}, recordsAsked{&asked{}}, &Catalogue{}) {
for _, tl := range append(Tools(&Store{}, recordsAsked{&asked{}}, &Catalogue{}), (&views.Reader{}).Tools("artifact_store")...) {
if !strings.HasPrefix(tl.Name, "artifact_store_") || tl.Description == "" || tl.Run == nil {
t.Errorf("tool %q", tl.Name)
}
@@ -469,10 +472,15 @@ func TestTheToolsServedAreTheToolsTheManifestNames(t *testing.T) {
}
// ADR 0571 clause 2: it holds the seat mesh-artifact-sweep and serves its verbs, and calls no verb of the controller's;
// it reads no controller state, the lease neither (issue 1126).
want := []string{"seat:mesh-hand-acts.hand-act", CatalogueModules, CatalogueModule}
want := []string{"seat:mesh-hand-acts.hand-act", CatalogueModules, CatalogueModule, WantsTool, ManifestsTool}
if !reflect.DeepEqual(m.Invokes, want) {
t.Errorf("invokes %v", m.Invokes)
}
// ADR 0615: the assigner's wants and the indexer's manifests are asked once and followed by their events, not read.
if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.views.manifests"}) ||
slices.Contains(m.Reads, WantsRead) || slices.Contains(m.Reads, ManifestsRead) {
t.Errorf("consumes %v, reads %v", m.Consumes, m.Reads)
}
var verbs []string
for _, tl := range SeatTools(&Holder{}, &Store{}) {
verbs = append(verbs, strings.TrimPrefix(tl.Name, catalogue.ArtifactSweepSeat+"."))
+10 -4
View File
@@ -20,15 +20,19 @@
"invokes": [
"seat:mesh-hand-acts.hand-act",
"seat:mesh-catalog.modules",
"seat:mesh-catalog.module"
"seat:mesh-catalog.module",
"mesh-delivery-assigner.assigner_read_wants",
"mesh-delivery-indexer.indexer_read_manifests"
],
"reads": [
"mesh-delivery-assigner.wants",
"mesh-delivery-builder.builds",
"mesh-delivery-builder.build-log",
"mesh-delivery-indexer.manifests",
"node-engine.have"
],
"consumes": [
"mesh-delivery-assigner.views.wants",
"mesh-delivery-indexer.views.manifests"
],
"state": [
"findings"
],
@@ -36,7 +40,9 @@
"artifact_store_repositories",
"artifact_store_usage",
"artifact_store_references",
"artifact_store_collect"
"artifact_store_collect",
"artifact_store_first_views",
"artifact_store_first_view_again"
],
"config": {
"older-store-names": {