diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/door.go b/modules/artifact-store-tools/cmd/artifact-store-tools/door.go index 4335d786..f93d5176 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/door.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/door.go @@ -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() } diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go index ee0c9242..e2c3308a 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/holder.go @@ -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 } diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/main.go b/modules/artifact-store-tools/cmd/artifact-store-tools/main.go index 8eb92fba..3ef95556 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/main.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/main.go @@ -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) } diff --git a/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go b/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go index ea6fdccd..325b522b 100644 --- a/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go +++ b/modules/artifact-store-tools/cmd/artifact-store-tools/store_test.go @@ -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+".")) diff --git a/modules/artifact-store-tools/module.json b/modules/artifact-store-tools/module.json index 54cc5916..5fd62c28 100644 --- a/modules/artifact-store-tools/module.json +++ b/modules/artifact-store-tools/module.json @@ -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": {