From 952dedd0edb7ccca05d5d34398d700460882c084 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 15:10:36 +0100 Subject: [PATCH] Join item 2: the readers follow through mesh-sdk's views.ProcessReader; sdk/firstview is gone mesh-delivery's sdk/firstview moved whole into mesh-sdk go/views (J3): the builder, checker, watchman, teller and view follow each owner's state with views.ProcessReader.Follow(owner, state, tool, stream), naming the owner's read tool as their invokes name it and its views family as the owner's manifest's `streams` names it ("indexer-views", "assigner-views"); a refused Follow stops the start. Their beds serve the owners with viewstest.Serve and grant with viewstest.Grants by the same names. The view's tools ask the owners' read tools by their literal names. The assigner names its own views family and the indexer's itself (views.AssignerStream and IndexerStream are gone). Every views.StatusTools call takes handacts.Recorder{Ask: stdio.Ask}: a first view asked again by hand is a hand act, recorded through seat:mesh-hand-acts.hand-act, which the builder's, checker's, watchman's, teller's and view's manifests now invoke (a row in each manifest test, each failed first). Row (A2j's written-time switch): TestRunNeedsItsCredential "a want stood since the assigner's bus wrote it": the watchman's want time (createdOr) is the time the owner's bus wrote the key, through ProcessReader's first view (failed with createdOr ignoring the entry's time). sdk/go.mod: nats.go is indirect now. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/mesh-delivery-assigner/views.go | 13 +- .../mesh-delivery-assigner/manifest_test.go | 4 +- .../cmd/mesh-delivery-builder-tools/main.go | 3 +- .../mesh-delivery-builder/integration_test.go | 6 +- .../cmd/mesh-delivery-builder/main.go | 33 +- .../mesh-delivery-builder/manifest_test.go | 2 +- modules/mesh-delivery-builder/module.json | 1 + .../cmd/mesh-delivery-checker-tools/main.go | 3 +- .../mesh-delivery-checker/integration_test.go | 6 +- .../cmd/mesh-delivery-checker/main.go | 36 +- .../mesh-delivery-checker/manifest_test.go | 2 +- modules/mesh-delivery-checker/module.json | 1 + .../cmd/mesh-delivery-teller-tools/main.go | 3 +- .../mesh-delivery-teller/integration_test.go | 6 +- .../cmd/mesh-delivery-teller/main.go | 10 +- modules/mesh-delivery-teller/manifest_test.go | 3 +- modules/mesh-delivery-teller/module.json | 3 +- .../cmd/mesh-delivery-view-tools/main.go | 23 +- .../mesh-delivery-view-tools/owners_test.go | 4 +- .../mesh-delivery-view/integration_test.go | 7 +- .../cmd/mesh-delivery-view/main.go | 28 +- modules/mesh-delivery-view/manifest_test.go | 1 + modules/mesh-delivery-view/module.json | 1 + .../cmd/mesh-delivery-watchman-tools/main.go | 3 +- .../integration_test.go | 6 +- .../cmd/mesh-delivery-watchman/main.go | 19 +- .../cmd/mesh-delivery-watchman/main_test.go | 31 + .../mesh-delivery-watchman/manifest_test.go | 3 +- modules/mesh-delivery-watchman/module.json | 3 +- sdk/firstview/firstview.go | 542 ------------------ sdk/firstview/firstviewtest/owner.go | 109 ---- sdk/go.mod | 6 +- 32 files changed, 178 insertions(+), 743 deletions(-) delete mode 100644 sdk/firstview/firstview.go delete mode 100644 sdk/firstview/firstviewtest/owner.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 008ea532..32da2932 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -10,6 +10,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" @@ -18,8 +19,12 @@ import ( "mesh-delivery-assigner/internal/assign" ) -// viewsStream is the assigner's views family on its old streams section, beside placement (ADR 0619 §2, ADR 0615 §5). -const viewsStream = views.AssignerStream +// viewsStream is the assigner's views family on its old streams section, beside placement (ADR 0619 §2, ADR 0615 §5), +// as its manifest's `streams` names it; indexerViewsStream is the indexer's, as the indexer's manifest names it. +const ( + viewsStream = "assigner-views" + indexerViewsStream = "indexer-views" +) // marksBucket is the assigner's state of what its views family said each state up to. var marksBucket = delivery.AssignerModule + "_" + views.MarksState @@ -67,7 +72,7 @@ func viewTools(getenv func(string) string, state func(string) views.Source, now out = append(out, views.ReadTool(delivery.AssignerModule, t.tool, t.state, t.words, state(t.state), marks, now)) } // Its own first views, of the indexer's manifests and modules (DLV-F3r): read, and asked again by hand. - return append(out, views.StatusTools(delivery.AssignerModule, "assigner_first_views", "assigner_first_view_again", stdio.State(views.StatusState), now)...) + return append(out, views.StatusTools(delivery.AssignerModule, "assigner_first_views", "assigner_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, now)...) } func keptSource(name string) views.Source { return views.KeptSource{State: stdio.State(name)} } @@ -169,7 +174,7 @@ func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settle } }) } - stream := catalogue.StreamBusName(views.IndexerStream) + stream := catalogue.StreamBusName(indexerViewsStream) if cons, err := iv.account.EventConsumer(ctx, iv.js, stream); err != nil { iv.logf("the consumer %s on %s does not exist yet (%v): the indexer's manifests and modules are the first views' alone until the assigner starts again", iv.account.EventConsumerName(), stream, err) diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 4ce9d83e..1fdb4d29 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -102,9 +102,9 @@ func TestManifest(t *testing.T) { } } // ADR 0619 §2, ADR 0615 §5: its views family beside placement on its old streams section. - if len(m.Streams) != 2 || m.Streams[1].Name != views.AssignerStream || !slices.Equal(m.Streams[1].Events, []string{views.EventPattern}) || + if len(m.Streams) != 2 || m.Streams[1].Name != "assigner-views" || !slices.Equal(m.Streams[1].Events, []string{views.EventPattern}) || m.Streams[1].MaxAgeDays != views.StreamDays || m.Streams[1].MaxMiB != views.StreamMiB || m.Streams[1].Retention != "" { - t.Errorf("streams = %+v, want placement and %s", m.Streams, views.AssignerStream) + t.Errorf("streams = %+v, want placement and assigner-views", m.Streams) } if m.Streams[0].Name != delivery.PlacementStream || m.Streams[0].MaxAgeDays != delivery.PlacementStreamDays || !slices.Equal(slices.Sorted(slices.Values(m.Streams[0].Events)), slices.Sorted(slices.Values(delivery.PlacementEvents))) { diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index 110fd3f2..3612b747 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -17,6 +17,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" 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-sdk/go/views" "git.novox.be/novox/mesh-delivery/sdk/enginereads" @@ -403,7 +404,7 @@ func main() { tools = append(tools, enginereads.Tool(EngineReadsTool, "the builder", enginereads.KeptState(stdio.State("engine-reads")), true, time.Now)) // builder_first_views and builder_first_view_again: where the builder's first views of the indexer's and the assigner's // states stand, and asking one again by hand (ADR 0615, DLV-M1). - tools = append(tools, views.StatusTools(delivery.BuilderModule, "builder_first_views", "builder_first_view_again", stdio.State(views.StatusState), time.Now)...) + tools = append(tools, views.StatusTools(delivery.BuilderModule, "builder_first_views", "builder_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index 4c47f301..75b7359f 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -27,12 +27,12 @@ import ( runner "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep" "git.novox.be/novox/mesh-contracts/go/seats/gopackageregistry" "git.novox.be/novox/mesh-contracts/go/seats/ociartifactregistry" - "git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest" "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views/viewstest" "mesh-delivery-builder/internal/build" "mesh-delivery-builder/internal/versions" @@ -291,9 +291,9 @@ func startBedWith(t *testing.T, closed bool) *bed { // (ADR 0615, DLV-M1); n1 and n2 each have their durable consumer. readers := []string{"n1_" + delivery.BuilderModule, "n2_" + delivery.BuilderModule} for _, st := range []string{modulesState, manifestsState, candidatesState} { - firstviewtest.Serve(t, ctx, js, ctl, delivery.IndexerModule, st, readers...) + viewstest.Serve(t, ctx, js, ctl, delivery.IndexerModule, st, delivery.IndexerModule+".indexer_read_"+st, indexerViews, readers...) } - firstviewtest.Serve(t, ctx, js, ctl, delivery.AssignerModule, delivery.PlacementState, readers...) + viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, delivery.PlacementState, delivery.AssignerModule+".assigner_read_placement", assignerViews, readers...) b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory) b.instances = mk(instancesBucket, 1) // The tag asks and their bounds (design 51 *Tagged versions*): the bounds declared key-ttl, so an ended bound leaves the diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index ff5791ff..2adfd338 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -28,8 +28,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" - "git.novox.be/novox/mesh-delivery/sdk/firstview" - "mesh-delivery-builder/internal/build" "mesh-delivery-builder/internal/versions" ) @@ -75,14 +73,26 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } defer calls.Close() - reader := &firstview.Reader{Module: delivery.BuilderModule, Call: calls.Call, Logf: logger.Printf} + reader := &views.ProcessReader{Module: delivery.BuilderModule, Call: calls.Call, Logf: logger.Printf} findingsKV := process.AwaitBucket(ctx, js, conn, findingsBucket, "the builder's findings, a first view unanswered past its bound among them, are said in its log only", logger.Printf) reader.Findings = findingsKV.KV - modules := reader.Follow(delivery.IndexerModule, modulesState) - manifests := reader.Follow(delivery.IndexerModule, manifestsState) + modules, err := reader.Follow(delivery.IndexerModule, modulesState, delivery.IndexerModule+".indexer_read_modules", indexerViews) + if err != nil { + return err + } + manifests, err := reader.Follow(delivery.IndexerModule, manifestsState, delivery.IndexerModule+".indexer_read_manifests", indexerViews) + if err != nil { + return err + } // The candidates' fingerprints are built too (issue 1684). - candidates := reader.Follow(delivery.IndexerModule, candidatesState) - placement := reader.Follow(delivery.AssignerModule, delivery.PlacementState) + candidates, err := reader.Follow(delivery.IndexerModule, candidatesState, delivery.IndexerModule+".indexer_read_candidates", indexerViews) + if err != nil { + return err + } + placement, err := reader.Follow(delivery.AssignerModule, delivery.PlacementState, delivery.AssignerModule+".assigner_read_placement", assignerViews) + if err != nil { + return err + } bound := defaultStepBound if v := strings.TrimSpace(getenv("MESH_BUILDER_STEP_BOUND")); v != "" { if bound, err = time.ParseDuration(v); err != nil { @@ -414,8 +424,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // keeper raises a first view unanswered past its bound (ADR 0615 §6). var firstViewsBucket = delivery.BuilderModule + "_" + views.StatusState -// The indexer's states the builder follows (delivery.ModulesBucket, ManifestsBucket, CandidatesBucket). +// The indexer's states the builder follows (delivery.ModulesBucket, ManifestsBucket, CandidatesBucket), and the owners' +// views families as their manifests' `streams` name them. const ( + indexerViews = "indexer-views" + assignerViews = "assigner-views" modulesState = "modules" manifestsState = "manifests" candidatesState = "candidates" @@ -423,8 +436,8 @@ const ( // buildFromModules builds from the indexer's modules and candidates once the first views of the modules, the manifests and // the candidates are answered (never at the start: an indexer that is down holds no other part of it), and follows them. -func buildFromModules(ctx context.Context, b *build.Builder, modules, manifests, candidates *firstview.Picture) error { - for _, p := range []*firstview.Picture{modules, manifests, candidates} { +func buildFromModules(ctx context.Context, b *build.Builder, modules, manifests, candidates *views.Picture) error { + for _, p := range []*views.Picture{modules, manifests, candidates} { select { case <-ctx.Done(): return nil diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index 6eef4cdc..7efe6b42 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -119,7 +119,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 // Design 51 *The floor*: the floor's three sources, the registry seats' versions, bridge_tags and indexer_requirements. if want := []string{"mesh-delivery-assigner.assigner_read_placement", "mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests", "mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions", - "seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove", + "seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove", "seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 { t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want) } diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index e1fbb2fe..81246c45 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -60,6 +60,7 @@ "seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", + "seat:mesh-hand-acts.hand-act", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove", "seat:npm-package-registry.versions", diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index f01e94b0..3a51d369 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -13,6 +13,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" 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-sdk/go/views" "mesh-delivery-checker/internal/check" @@ -329,7 +330,7 @@ func main() { tools := append(append(append(Tools(r), TryTools(trying)...), SeatTryTools(trying)...), RecoveryTools(r, written)...) // checker_first_views and checker_first_view_again: where the checker's first views of the indexer's and the assigner's // states stand, and asking one again by hand (ADR 0615, DLV-M2). - tools = append(tools, views.StatusTools(delivery.CheckerModule, "checker_first_views", "checker_first_view_again", stdio.State(views.StatusState), time.Now)...) + tools = append(tools, views.StatusTools(delivery.CheckerModule, "checker_first_views", "checker_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, time.Now)...) if err := stdio.Serve("", logged(tools)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index 871a7a31..af0bdedd 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -29,10 +29,10 @@ import ( ne "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-contracts/go/seats/npmpackageregistry" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" - "git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views/viewstest" "mesh-delivery-checker/internal/check" ) @@ -446,9 +446,9 @@ func startBed(t *testing.T) *bed { // The indexer and the assigner as the owners of the states the checker follows: their read tools and views families // (ADR 0615, DLV-M2). for _, st := range []string{candidatesState, modulesState, manifestsState, delivery.PackagesState, delivery.IndexerPredefinedState, seatDefinitionsState} { - firstviewtest.Serve(t, ctx, js, ctl, delivery.IndexerModule, st, "n1_"+delivery.CheckerModule) + viewstest.Serve(t, ctx, js, ctl, delivery.IndexerModule, st, delivery.IndexerModule+".indexer_read_"+strings.ReplaceAll(st, "-", "_"), indexerViews, "n1_"+delivery.CheckerModule) } - firstviewtest.Serve(t, ctx, js, ctl, delivery.AssignerModule, delivery.PlacementState, "n1_"+delivery.CheckerModule) + viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, delivery.PlacementState, delivery.AssignerModule+".assigner_read_placement", assignerViews, "n1_"+delivery.CheckerModule) // 1929/AC1: the indexer's value of the repository's module, with no TypeScript manifest, so the Go build module alone // judges the repository (an unknown repository is judged by both, issue 1887 A12). known, err := delivery.EncodeModule(delivery.Module{Repository: repo, Commit: trunk, Path: "modules/m", None2: "the fingerprint is computed with its chain"}) diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index e0b3de85..059541db 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -8,6 +8,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "log" @@ -26,8 +27,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" - "git.novox.be/novox/mesh-delivery/sdk/firstview" - "mesh-delivery-checker/internal/check" ) @@ -77,8 +76,11 @@ const ( // seatDefinitionsState is the indexer's state of the seat definitions (design 26), keyed `.trunk` and // `.pull..`. seatDefinitionsState = "seat-definitions" - modulesState = "modules" - manifestsState = "manifests" + // The owners' views families, as their manifests' `streams` name them. + indexerViews = "indexer-views" + assignerViews = "assigner-views" + modulesState = "modules" + manifestsState = "manifests" ) // run is the program: it returns when ctx ends or the bus cannot be used. @@ -139,16 +141,26 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // first view of their read tools and then their views families, never their buckets (ADR 0615 §1, ADR 0596, DLV-M2). // An owner that is down never holds the start: the checks start once every first view is answered, and until then the // unanswered first view is the checker's finding. - reader := &firstview.Reader{Module: delivery.CheckerModule, Call: caller.Call, Logf: logger.Printf} + reader := &views.ProcessReader{Module: delivery.CheckerModule, Call: caller.Call, Logf: logger.Printf} reader.Findings = process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf).KV - candidates := reader.Follow(delivery.IndexerModule, candidatesState) - modules := reader.Follow(delivery.IndexerModule, modulesState) - manifests := reader.Follow(delivery.IndexerModule, manifestsState) - packages := reader.Follow(delivery.IndexerModule, delivery.PackagesState) - predefined := reader.Follow(delivery.IndexerModule, delivery.IndexerPredefinedState) - placement := reader.Follow(delivery.AssignerModule, delivery.PlacementState) + // Each by the owner's read tool and views family as the owner's manifest names them. + var followed error + followOwner := func(owner, state, tool, stream string) *views.Picture { + p, err := reader.Follow(owner, state, owner+"."+tool, stream) + followed = errors.Join(followed, err) + return p + } + candidates := followOwner(delivery.IndexerModule, candidatesState, "indexer_read_candidates", indexerViews) + modules := followOwner(delivery.IndexerModule, modulesState, "indexer_read_modules", indexerViews) + manifests := followOwner(delivery.IndexerModule, manifestsState, "indexer_read_manifests", indexerViews) + packages := followOwner(delivery.IndexerModule, delivery.PackagesState, "indexer_read_packages", indexerViews) + predefined := followOwner(delivery.IndexerModule, delivery.IndexerPredefinedState, "indexer_read_predefined", indexerViews) + placement := followOwner(delivery.AssignerModule, delivery.PlacementState, "assigner_read_placement", assignerViews) // The seventh: the indexer's seat definitions, so a pull request whose head's definition does not parse fails. - seatDefs := reader.Follow(delivery.IndexerModule, seatDefinitionsState) + seatDefs := followOwner(delivery.IndexerModule, seatDefinitionsState, "indexer_read_seat_definitions", indexerViews) + if followed != nil { + return followed + } stepBound, err := durationEnv(getenv, "MESH_CHECKER_STEP_BOUND", check.DefaultStepBound) if err != nil { diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 3fdc27ca..93b4e6a9 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -124,7 +124,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 "mesh-delivery-assigner.assigner_read_placement", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests", "mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_read_predefined", "mesh-delivery-indexer.indexer_read_seat_definitions", - "seat:mesh-config.renames", "seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) { + "seat:mesh-config.renames", "seat:mesh-hand-acts.hand-act", "seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) { t.Errorf("invokes = %v, want the forge's reads and compose-candidate", m.Invokes) } if want := []string{"node-build-go", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) { diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 76129b4c..e1ccb9d1 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -82,6 +82,7 @@ "mesh-delivery-indexer.indexer_read_predefined", "mesh-delivery-indexer.indexer_read_seat_definitions", "seat:mesh-config.renames", + "seat:mesh-hand-acts.hand-act", "seat:node-engine.compose-candidate" ], diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go index f5b5e57d..dd9bfa22 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller-tools/main.go @@ -13,6 +13,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-teller/internal/ends" @@ -106,7 +107,7 @@ func main() { // teller_first_views and teller_first_view_again: where the teller's first view of the assigner's asks stands, and // asking it again by hand (ADR 0615, design 25 § 12.2). tools := append(Tools(askEnd(stdio.State(ends.State), 25*time.Second), time.Now), - views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), time.Now)...) + views.StatusTools("mesh-delivery-teller", "teller_first_views", "teller_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-teller-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go index d04702aa..cda389ea 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/integration_test.go @@ -8,7 +8,7 @@ import ( "fmt" occ "git.novox.be/novox/mesh-contracts/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel" - "git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest" + "git.novox.be/novox/mesh-sdk/go/views/viewstest" "os" "path/filepath" "strings" @@ -189,7 +189,7 @@ func fullConf(dir string) string { teller = append(teller, readOf(viewAnswers)...) // issue 1297 B10: the manifest's reads // The assigner's asks by its read tool and its views family (ADR 0615, DLV-M3), and the teller's own first-views and // findings; the bucket itself is no longer the teller's to read. - teller = append(append(append(teller, firstviewtest.Grants(delivery.AssignerModule, "asks", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...) + teller = append(append(append(teller, viewstest.Grants(delivery.AssignerModule, delivery.AssignerModule+".assigner_read_asks", "assigner-views", tellerConsumer)...), keepsOf(firstViewsBucket)...), keepsOf(findingsBucket)...) view := append(append(append(readOf(tellerAnswers), readOf(gateAsksKV)...), readOf(assignerAsks)...), keepsOf(viewAnswers)...) assigner := keepsOf(assignerAsks) tools := append(keepsOf(gateAsksKV), readOf(tellerAnswers)...) @@ -273,7 +273,7 @@ func startFull(t *testing.T) *fullBed { t.Fatal(err) } // The assigner as the owner of its asks: its read tool and its views family (ADR 0615). - firstviewtest.Serve(t, ctx, js, ctl, delivery.AssignerModule, "asks", tellerConsumer) + viewstest.Serve(t, ctx, js, ctl, delivery.AssignerModule, "asks", delivery.AssignerModule+".assigner_read_asks", "assigner-views", tellerConsumer) return &fullBed{srv: srv, ctl: ctl, js: js, ctx: ctx} } diff --git a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go index fd69428a..623f3724 100644 --- a/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go +++ b/modules/mesh-delivery-teller/cmd/mesh-delivery-teller/main.go @@ -24,8 +24,6 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - "git.novox.be/novox/mesh-delivery/sdk/firstview" - "git.novox.be/novox/mesh-sdk/go/delivery" "mesh-delivery-teller/internal/ends" @@ -94,10 +92,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("the teller's connection for its tool calls: %w", err) } defer calls.Close() - reader := &firstview.Reader{Module: delivery.TellerModule, Call: calls.Call, Logf: logger.Printf} + reader := &views.ProcessReader{Module: delivery.TellerModule, Call: calls.Call, Logf: logger.Printf} findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf) reader.Findings = findings.KV - assignerKV := jetstream.KeyValue(reader.Follow(delivery.AssignerModule, assignerAsksState)) + asks, err := reader.Follow(delivery.AssignerModule, assignerAsksState, delivery.AssignerModule+".assigner_read_asks", "assigner-views") + if err != nil { + return err + } + assignerKV := jetstream.KeyValue(asks) reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "teller_first_views reads nothing and teller_first_view_again is not heard", logger.Printf)) viewKV := process.OptionalBucket(ctx, js, delivery.ViewAnswersBucket, teller, logger.Printf) // The assigner's asks are raised on the operator channel whether or not the view's answers can be read (issue 1906). diff --git a/modules/mesh-delivery-teller/manifest_test.go b/modules/mesh-delivery-teller/manifest_test.go index 8de6b9e8..45077315 100644 --- a/modules/mesh-delivery-teller/manifest_test.go +++ b/modules/mesh-delivery-teller/manifest_test.go @@ -96,7 +96,8 @@ func TestManifestDeclarations(t *testing.T) { // 1240/AC1 t.Errorf("reads %s: the indexer's and the assigner's state is read by their tools and views families alone", r) } } - if want := []string{"mesh-delivery-assigner.assigner_read_asks"}; !reflect.DeepEqual(m.Invokes, want) { + // The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools). + if want := []string{"mesh-delivery-assigner.assigner_read_asks", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } if want := []string{"mesh-delivery-assigner.views.asks"}; !reflect.DeepEqual(m.Consumes, want) { diff --git a/modules/mesh-delivery-teller/module.json b/modules/mesh-delivery-teller/module.json index be0af507..6fa9c0dd 100644 --- a/modules/mesh-delivery-teller/module.json +++ b/modules/mesh-delivery-teller/module.json @@ -8,7 +8,8 @@ "teller_first_view_again" ], "invokes": [ - "mesh-delivery-assigner.assigner_read_asks" + "mesh-delivery-assigner.assigner_read_asks", + "seat:mesh-hand-acts.hand-act" ], "consumes": [ "mesh-delivery-assigner.views.asks" diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index 5e83d339..21b91f60 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -13,9 +13,8 @@ import ( "mesh-delivery-view/internal/view" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/handacts" - "git.novox.be/novox/mesh-delivery/sdk/firstview" - "git.novox.be/novox/mesh-delivery/sdk/graph" "git.novox.be/novox/mesh-sdk/go/provisions" "git.novox.be/novox/mesh-sdk/go/views" @@ -394,14 +393,16 @@ func read(state stdio.KeptState) view.StateRead { // by the bundle on each call (ADR 0615 §1, design 25 § 12.2: the view keeps its tools and answers them by asking the // owners), never their buckets. func ownersRead(ask func(key string, body any) (json.RawMessage, error)) view.Reader { - of := func(owner, state string) view.StateRead { return view.OwnerRead(ask, firstview.ReadTool(owner, state)) } - return view.Reader{State: of(delivery.IndexerModule, "modules"), CandidatesState: of(delivery.IndexerModule, "candidates"), - WantsState: of(delivery.AssignerModule, "wants"), DifferencesState: of(delivery.AssignerModule, "differences"), - HeardState: of(delivery.IndexerModule, "heard"), - PredefinedState: of(delivery.IndexerModule, delivery.IndexerPredefinedState), - ConfigCandidatesState: of(delivery.IndexerModule, delivery.IndexerConfigCandidatesState), - PackagesState: of(delivery.IndexerModule, delivery.PackagesState), - GraphState: of(delivery.IndexerModule, graph.State)} + // Each owner's read tool as the view's invokes names it. + indexer := func(tool string) view.StateRead { return view.OwnerRead(ask, delivery.IndexerModule+"."+tool) } + assigner := func(tool string) view.StateRead { return view.OwnerRead(ask, delivery.AssignerModule+"."+tool) } + return view.Reader{State: indexer("indexer_read_modules"), CandidatesState: indexer("indexer_read_candidates"), + WantsState: assigner("assigner_read_wants"), DifferencesState: assigner("assigner_read_differences"), + HeardState: indexer("indexer_read_heard"), + PredefinedState: indexer("indexer_read_predefined"), + ConfigCandidatesState: indexer("indexer_read_config_candidates"), + PackagesState: indexer("indexer_read_packages"), + GraphState: indexer("indexer_read_graph")} } func main() { @@ -423,7 +424,7 @@ func main() { } // view_first_views and view_first_view_again: where the view's process stands with its first views of the indexer's and // the assigner's states (ADR 0615 §6), and a re-read by hand. - tools := append(Tools(r, u), views.StatusTools(delivery.ViewModule, "view_first_views", "view_first_view_again", stdio.State(views.StatusState), time.Now)...) + tools := append(Tools(r, u), views.StatusTools(delivery.ViewModule, "view_first_views", "view_first_view_again", stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, time.Now)...) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-view-tools] %v\n", err) os.Exit(1) diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go index e3b27130..4ce88e98 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go @@ -11,7 +11,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/views" - "git.novox.be/novox/mesh-delivery/sdk/firstview" "git.novox.be/novox/mesh-delivery/sdk/graph" "mesh-delivery-view/internal/view" @@ -68,7 +67,8 @@ func ownersAnswerTheToolsAsBefore(t *testing.T) { tools := map[string]string{} for name := range states { owner, state, _ := strings.Cut(name, ".") - tools[firstview.ReadTool(owner, state)] = name + short := map[string]string{delivery.IndexerModule: "indexer_read_", delivery.AssignerModule: "assigner_read_"}[owner] + tools[owner+"."+short+strings.ReplaceAll(state, "-", "_")] = name } asked := map[string]int{} ask := func(key string, body any) (json.RawMessage, error) { diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go index 4d1f43f4..a3f84601 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -21,7 +21,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" - "git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest" + "git.novox.be/novox/mesh-sdk/go/views/viewstest" "mesh-delivery-view/internal/view" ) @@ -31,9 +31,8 @@ import ( // A bucket the test made is kept; the others are made empty. func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) { t.Helper() - for _, bucket := range ownersBuckets { - owner, state, _ := strings.Cut(bucket, "_") - firstviewtest.Serve(t, ctx, js, nc, owner, state, "n_"+delivery.ViewModule) + for _, o := range ownersFollowed { + viewstest.Serve(t, ctx, js, nc, o.owner, o.state, o.tool, o.stream, "n_"+delivery.ViewModule) } } diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go index ae328646..05b51377 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -27,8 +27,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" - "git.novox.be/novox/mesh-delivery/sdk/firstview" - "mesh-delivery-view/internal/view" ) @@ -118,11 +116,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } defer calls.Close() - reader := &firstview.Reader{Module: delivery.ViewModule, Call: calls.Call, Logf: logger.Printf} + reader := &views.ProcessReader{Module: delivery.ViewModule, Call: calls.Call, Logf: logger.Printf} reader.Findings = process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf).KV - for _, bucket := range ownersBuckets { - owner, state, _ := strings.Cut(bucket, "_") - reader.Follow(owner, state) + for _, o := range ownersFollowed { + if _, err := reader.Follow(o.owner, o.state, o.tool, o.stream); err != nil { + return err + } } open := buckets{js: js, reader: reader} for _, bucket := range []string{delivery.HaveBucket, delivery.PlacementBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket, delivery.TakenBucket} { @@ -180,15 +179,22 @@ var ( findingsBucket = delivery.ViewModule + "_findings" ) -// ownersBuckets are the indexer's and the assigner's states the view's process follows by their first views -// (`_`); the view's tools ask the other six through the owners' read tools on each call. -var ownersBuckets = []string{delivery.PlacementBucket, delivery.AssignerWantsBucket, delivery.AssignerAsksBucket, - delivery.CandidatesBucket, delivery.ModulesBucket, delivery.TakenBucket} +// ownersFollowed are the indexer's and the assigner's states the view's process follows by their first views, each by its +// owner's read tool (as the view's invokes names it) and views family (as the owner's manifest's `streams` names it); the +// view's tools ask the other six through the owners' read tools on each call. +var ownersFollowed = []struct{ owner, state, tool, stream string }{ + {delivery.AssignerModule, delivery.PlacementState, delivery.AssignerModule + ".assigner_read_placement", "assigner-views"}, + {delivery.AssignerModule, "wants", delivery.AssignerModule + ".assigner_read_wants", "assigner-views"}, + {delivery.AssignerModule, "asks", delivery.AssignerModule + ".assigner_read_asks", "assigner-views"}, + {delivery.IndexerModule, "candidates", delivery.IndexerModule + ".indexer_read_candidates", "indexer-views"}, + {delivery.IndexerModule, "modules", delivery.IndexerModule + ".indexer_read_modules", "indexer-views"}, + {delivery.IndexerModule, "taken", delivery.IndexerModule + ".indexer_read_taken", "indexer-views"}, +} // buckets opens what the view watches: an owner's state as its picture (firstview), any other bucket on the bus. type buckets struct { js jetstream.JetStream - reader *firstview.Reader + reader *views.ProcessReader } func (b buckets) KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) { diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index 1ce8fc79..3922b2b5 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -106,6 +106,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 "mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_config_candidates", "mesh-delivery-indexer.indexer_read_graph", "mesh-delivery-indexer.indexer_read_heard", "mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_read_predefined", "mesh-delivery-indexer.indexer_read_taken", + "seat:mesh-hand-acts.hand-act", // the join: its first view asked again by hand is a hand act (views.StatusTools) "seat:node-engine.manifest-reads"}; !reflect.DeepEqual(m.Invokes, want) { // ADR 0618 §3: node_manifest_reads asks the node-engine seat's verb of every node t.Errorf("invokes = %v, want %v", m.Invokes, want) } diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 5fab8e17..37afda1f 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -62,6 +62,7 @@ "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_read_predefined", "mesh-delivery-indexer.indexer_read_taken", + "seat:mesh-hand-acts.hand-act", "seat:node-engine.manifest-reads" ], "reads": [ diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go index 9e3cb313..269c8121 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman-tools/main.go @@ -13,6 +13,7 @@ import ( "time" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-watchman/internal/socket" @@ -163,7 +164,7 @@ func Tools(states func(string) state, ask func(socket.Ask) (json.RawMessage, err Description: "The watchman's findings (novox/hq ADR 0564 §2): each thing it observes wrong now, .., with its " + "severity, subject, evidence and the row that observed it; the condition keeper raises and clears the conditions from " + "them. Give key for one, prefix to narrow. Read-only."}, - }, views.StatusTools(watch.Observer, ToolFirstViews, ToolFirstViewAgain, stdio.State(views.StatusState), time.Now)...) + }, views.StatusTools(watch.Observer, ToolFirstViews, ToolFirstViewAgain, stdio.State(views.StatusState), handacts.Recorder{Ask: stdio.Ask}, time.Now)...) } func main() { diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go index 6ea1b1fd..c509bf7a 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go @@ -17,10 +17,10 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" - "git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest" "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views/viewstest" "mesh-delivery-watchman/internal/deadletters" "mesh-delivery-watchman/internal/socket" @@ -62,8 +62,8 @@ accounts { // families (DLV-M5, ADR 0615): the watchman (n1) follows both, never their buckets. func serveOwners(t *testing.T, ctx context.Context, js jetstream.JetStream, nc *nats.Conn) { t.Helper() - firstviewtest.Serve(t, ctx, js, nc, delivery.AssignerModule, wantsState, "n1_"+delivery.WatchmanModule) - firstviewtest.Serve(t, ctx, js, nc, delivery.IndexerModule, modulesState, "n1_"+delivery.WatchmanModule) + viewstest.Serve(t, ctx, js, nc, delivery.AssignerModule, wantsState, delivery.AssignerModule+".assigner_read_wants", assignerViews, "n1_"+delivery.WatchmanModule) + viewstest.Serve(t, ctx, js, nc, delivery.IndexerModule, modulesState, delivery.IndexerModule+".indexer_read_modules", indexerViews, "n1_"+delivery.WatchmanModule) } // TestWatchmanWhole starts the program the way the node does (its entry, run, with its environment and a broker file) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index dd88ae0d..5a2b9a09 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -20,8 +20,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" - "git.novox.be/novox/mesh-delivery/sdk/firstview" - "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -128,7 +126,7 @@ func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyV // readLate reads an owner's state the watchman follows by its first view (pic), on a goroutine of its own once that is // answered, so an owner that is down never holds the start: every key it holds now, then the start's end for its bucket // (watch.ReadLate), then each change until ctx ends. -func readLate(ctx context.Context, w *watch.Watchman, pic *firstview.Picture, bucket string, each func(jetstream.KeyValueEntry)) error { +func readLate(ctx context.Context, w *watch.Watchman, pic *views.Picture, bucket string, each func(jetstream.KeyValueEntry)) error { select { case <-ctx.Done(): return nil @@ -188,6 +186,9 @@ var firstViewsBucket = delivery.WatchmanModule + "_" + views.StatusState const ( wantsState = "wants" modulesState = "modules" + // The owners' views families, as their manifests' `streams` name them. + indexerViews = "indexer-views" + assignerViews = "assigner-views" ) // run is the program: it returns when ctx ends or the bus cannot be used. @@ -232,10 +233,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return fmt.Errorf("the watchman's connection for its tool calls: %w", err) } defer calls.Close() - reader := &firstview.Reader{Module: delivery.WatchmanModule, Call: calls.Call, Logf: logger.Printf, + reader := &views.ProcessReader{Module: delivery.WatchmanModule, Call: calls.Call, Logf: logger.Printf, Findings: func() jetstream.KeyValue { return kvs.findings }} - wants := reader.Follow(delivery.AssignerModule, wantsState) - modules := reader.Follow(delivery.IndexerModule, modulesState) + wants, err := reader.Follow(delivery.AssignerModule, wantsState, delivery.AssignerModule+".assigner_read_wants", assignerViews) + if err != nil { + return err + } + modules, err := reader.Follow(delivery.IndexerModule, modulesState, delivery.IndexerModule+".indexer_read_modules", indexerViews) + if err != nil { + return err + } w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck, findings: kvs.findings}, time.Now) w.Logf(logger.Printf) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index f4c9e742..f711e98e 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -2,6 +2,7 @@ package main import ( "context" + "encoding/json" "errors" "fmt" "io" @@ -9,7 +10,9 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-watchman/internal/watch" @@ -75,6 +78,34 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") { t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err) } + t.Run("A2j: a want stood since the assigner's bus wrote it, through the first view", func(t *testing.T) { + // The watchman's want wait starts at the entry's time (createdOr): through mesh-sdk's views.ProcessReader it is + // the time the owner's bus wrote the key, never the watchman's start. + written := time.Date(2026, 11, 4, 9, 0, 0, 0, time.UTC) + r := &views.ProcessReader{Module: delivery.WatchmanModule, Call: func(string, any) (json.RawMessage, error) { + return json.Marshal(views.Answer{V: views.Version, Owner: delivery.AssignerModule, State: wantsState, Revision: 3, Created: written, + At: written.Add(time.Hour), Values: map[string]json.RawMessage{"ace.mail": json.RawMessage(`{}`)}, Written: map[string]time.Time{"ace.mail": written}}) + }} + p, err := r.Follow(delivery.AssignerModule, wantsState, delivery.AssignerModule+".assigner_read_wants", assignerViews) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + r.Ask(ctx, nil) + select { + case <-r.Ready(): + case <-ctx.Done(): + t.Fatal("the first view was not taken") + } + e, err := p.Get(ctx, "ace.mail") + if err != nil { + t.Fatal(err) + } + if got := createdOr(e, time.Now()); !got.Equal(written) { + t.Errorf("the want stands since %v, want %v, when the assigner's bus wrote it", got, written) + } + }) } // refusingBus refuses every bucket with err (issue 1945). diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 4e72ada3..417fe3fc 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -103,7 +103,8 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 t.Errorf("reads %s: the indexer's and the assigner's state is read by their tools and views families alone", r) } } - if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_read_modules"}; !reflect.DeepEqual(m.Invokes, want) { + // The join: its first view asked again by hand is a hand act, recorded through the seat (views.StatusTools). + if want := []string{"mesh-delivery-assigner.assigner_read_wants", "mesh-delivery-indexer.indexer_read_modules", "seat:mesh-hand-acts.hand-act"}; !reflect.DeepEqual(m.Invokes, want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) } // DLV-M5 its first views, and DLV-W1 (ADR 0595 §4): the accepted verdicts and the bus module's created and refused, durably. diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index dcac4d8e..baf261f5 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -26,7 +26,8 @@ ], "invokes": [ "mesh-delivery-assigner.assigner_read_wants", - "mesh-delivery-indexer.indexer_read_modules" + "mesh-delivery-indexer.indexer_read_modules", + "seat:mesh-hand-acts.hand-act" ], "consumes": [ "mesh-delivery-assigner.views.wants", diff --git a/sdk/firstview/firstview.go b/sdk/firstview/firstview.go deleted file mode 100644 index 3795dca0..00000000 --- a/sdk/firstview/firstview.go +++ /dev/null @@ -1,542 +0,0 @@ -// Package firstview is a process's pictures of the indexer's and the assigner's states it used to read (novox/hq ADR -// 0615 §1, design 25 § 12.2), for the mesh-delivery parts that run as processes (the builder, the checker, the teller, -// the view, the watchman): each state is asked once through the owner's ordinary read tool with `values` (the first -// view), then kept from the owner's views family, asked again only on a gap or by hand (mesh-sdk go/views Follower). -// -// A Picture reads as the key-value bucket the process used to bind (jetstream.KeyValue: Get, Keys, ListKeys, Watch, -// WatchAll), so a reader's code is unchanged; every write is refused, for the owner's state is the owner's alone. The -// picture is kept in memory: the owner's state is the truth and a restart asks again (ADR 0615 §4). -// -// It is the process side of mesh-sdk go/views Reader (the stdio side) and moves there whole when mesh-sdk takes it. -package firstview - -import ( - "context" - "encoding/json" - "errors" - "fmt" - "slices" - "strings" - "sync" - "time" - - "github.com/nats-io/nats.go/jetstream" - - "git.novox.be/novox/mesh-sdk/go/catalogue" - "git.novox.be/novox/mesh-sdk/go/process" - "git.novox.be/novox/mesh-sdk/go/views" -) - -// ErrReadOnly is every write to a picture: the state is its owner's. -var ErrReadOnly = errors.New("a first view is read-only: the state is its owner's") - -// ReadTool is the owner's ordinary read tool of state, as a reader's `invokes` names it (design 25 § 12.2): -// `._read_`, short the owner's name past its last dash-separated `mesh-delivery-` prefix -// (indexer, assigner) and the state's dashes as underscores. -func ReadTool(owner, state string) string { - short := strings.TrimPrefix(owner, "mesh-delivery-") - return owner + "." + short + "_read_" + strings.ReplaceAll(state, "-", "_") -} - -// Reader is one process's pictures of its owners' states. -type Reader struct { - // Module is the reading module, named in the finding of a first view unanswered past its bound. - Module string - // Call asks a module's tool (process.Conn.Call): the read tools are asked with `values`. - Call func(key string, body any) (json.RawMessage, error) - // Findings is the reader's own state `findings`, where the condition keeper raises and clears the condition of a - // first view unanswered past its bound (ADR 0615 §6); nil says it in the log only. - Findings func() jetstream.KeyValue - Logf func(string, ...any) - // Now, Bound and Wait are for tests: zero are the clock, views.Bound and five seconds. - Now func() time.Time - Bound time.Duration - Wait time.Duration - - mu sync.Mutex - pics map[string]*Picture // by bucket name _ - order []string -} - -func (r *Reader) logf(format string, args ...any) { - if r.Logf != nil { - r.Logf(format, args...) - } -} - -// Follow adds the owner's state to follow through its read tool (ReadTool). Call it before Start. -func (r *Reader) Follow(owner, state string) *Picture { - r.mu.Lock() - defer r.mu.Unlock() - if r.pics == nil { - r.pics = map[string]*Picture{} - } - bucket := owner + "_" + state - if p := r.pics[bucket]; p != nil { - return p - } - tool := ReadTool(owner, state) - p := &Picture{bucket: bucket, ctx: context.Background()} - call := r.Call - ask := views.AskVia(func(key string, body any) (json.RawMessage, error) { - m, _ := body.(map[string]any) - if m == nil { - m = map[string]any{} - } - m["values"] = true - if call == nil { - return nil, errors.New("this reader asks no tool") - } - return call(key, m) - }, tool) - p.f = &views.Follower{Owner: owner, Tool: tool, State: state, Ask: ask, Logf: r.Logf, Now: r.Now, Bound: r.Bound, Wait: r.Wait, - Raise: func(s views.Status) { r.finding(s, true) }, Clear: func(s views.Status) { r.finding(s, false) }, - Changed: p.changed} - r.pics[bucket] = p - r.order = append(r.order, bucket) - return p -} - -// Picture is the picture of the bucket the process used to bind, by its bus name (`_`); nil when not -// followed. -func (r *Reader) Picture(bucket string) *Picture { - r.mu.Lock() - defer r.mu.Unlock() - return r.pics[bucket] -} - -// KeyValue opens a followed state's picture by its bus name, where a process opened the bucket (process.BucketOpener, -// as buildrun.FollowPlacement takes it); any other bucket is refused, for the reader reads its owners' states by their -// first views alone. -func (r *Reader) KeyValue(_ context.Context, bucket string) (jetstream.KeyValue, error) { - if p := r.Picture(bucket); p != nil { - return p, nil - } - return nil, fmt.Errorf("%s reads %s by its owner's first view alone, and does not follow it", r.Module, bucket) -} - -// Followers are the followers, in the order they were added. -func (r *Reader) Followers() []*views.Follower { - r.mu.Lock() - defer r.mu.Unlock() - out := make([]*views.Follower, 0, len(r.order)) - for _, b := range r.order { - out = append(out, r.pics[b].f) - } - return out -} - -func (r *Reader) finding(s views.Status, raised bool) { - if raised { - r.logf("the first view %s of %s's %s is unanswered past its bound: %s", s.Tool, s.Owner, s.State, s.Error) - } - if r.Findings == nil { - return - } - kv := r.Findings() - if kv == nil { - return - } - at := time.Now() - if r.Now != nil { - at = r.Now() - } - key, value, err := views.UnansweredFinding(r.Module, s, at) - if err == nil && raised { - _, err = kv.Put(context.Background(), key, value) - } else if err == nil { - err = kv.Delete(context.Background(), key) - } - if err != nil { - r.logf("views: the finding of the first view %s is not written: %v", s.Tool, err) - } -} - -// Hear is the handler of the owner's views family on the reader's durable consumer (cred.EventConsumer on the owner's -// views stream): each event to the follower of its state. -func (r *Reader) Hear(ctx context.Context, owner string) func(jetstream.Msg) { - fs := map[string]*views.Follower{} - for _, f := range r.Followers() { - if f.Owner == owner { - fs[f.State] = f - } - } - return views.Hear(ctx, fs, r.logf) -} - -// Owners are the owners followed, sorted. -func (r *Reader) Owners() []string { - var out []string - for _, f := range r.Followers() { - if !slices.Contains(out, f.Owner) { - out = append(out, f.Owner) - } - } - slices.Sort(out) - return out -} - -// Ask asks every first view until answered, each on a goroutine of its own, so the process starts at once and waits for -// an owner that is down, saying so (ADR 0615 §6). Said, when set, keeps each status (views.KeepStatus). -func (r *Reader) Ask(ctx context.Context, said func(views.Status)) { - for _, f := range r.Followers() { - if said != nil { - f.Said = said - } - go func() { - if err := f.Start(ctx); err != nil { - r.logf("the first view %s was not answered: %v", f.Tool, err) - } - }() - } -} - -// Ready is closed once every followed state had its first view. -func (r *Reader) Ready() <-chan struct{} { - fs := r.Followers() - done := make(chan struct{}) - go func() { - for _, f := range fs { - <-f.Ready() - } - close(done) - }() - return done -} - -// Picture is one owner's state as the process reads it: a read-only jetstream.KeyValue over a views.Follower. Its -// revisions are the picture's own, one per change it took, so a watcher sees them rise; they are not the owner's. -type Picture struct { - jetstream.KeyValue // never set: every method a reader may call is the picture's own - - bucket string - ctx context.Context - f *views.Follower - - mu sync.Mutex - seq uint64 - revs map[string]uint64 - watchers []*watcher -} - -// Follower is the picture's follower (its status, a gap or a hand ask). -func (p *Picture) Follower() *views.Follower { return p.f } - -func (p *Picture) changed(key string, value json.RawMessage, deleted bool, at time.Time) { - p.mu.Lock() - p.seq++ - if p.revs == nil { - p.revs = map[string]uint64{} - } - p.revs[key] = p.seq - op := jetstream.KeyValuePut - if deleted { - op, value = jetstream.KeyValueDelete, nil - } - e := entry{bucket: p.bucket, key: key, value: value, rev: p.seq, at: at, op: op} - ws := slices.Clone(p.watchers) - p.mu.Unlock() - for _, w := range ws { - w.send(e) - } -} - -// answered is nil once the first view is answered, else the error a read before it gets (never a wait). -func (p *Picture) answered() error { - select { - case <-p.f.Ready(): - return nil - default: - return fmt.Errorf("no first view of %s yet: %s has not answered", p.bucket, p.f.Tool) - } -} - -func (p *Picture) ready(ctx context.Context) error { - select { - case <-p.f.Ready(): - return nil - case <-ctx.Done(): - return fmt.Errorf("no first view of %s yet (%s has not answered): %w", p.bucket, p.f.Tool, ctx.Err()) - } -} - -// Bucket is the bus name of the owner's state the picture stands for. -func (p *Picture) Bucket() string { return p.bucket } - -// Get is one key; before the first view is answered it is an error, never a wait. -func (p *Picture) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { - if err := p.answered(); err != nil { - return nil, err - } - v, ok := p.f.Get(key) - if !ok { - return nil, jetstream.ErrKeyNotFound - } - p.mu.Lock() - rev := p.revs[key] - p.mu.Unlock() - return entry{bucket: p.bucket, key: key, value: v, rev: rev, op: jetstream.KeyValuePut}, nil -} - -// Keys is every key; none is jetstream.ErrNoKeysFound, as the bus says it; before the first view is answered it is an -// error, never a wait. -func (p *Picture) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) { - if err := p.answered(); err != nil { - return nil, err - } - var keys []string - for k := range p.f.Values() { - keys = append(keys, k) - } - if len(keys) == 0 { - return nil, jetstream.ErrNoKeysFound - } - slices.Sort(keys) - return keys, nil -} - -// ListKeys is Keys on a channel. -func (p *Picture) ListKeys(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyLister, error) { - keys, err := p.Keys(ctx, opts...) - if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) { - return nil, err - } - ch := make(chan string, len(keys)) - for _, k := range keys { - ch <- k - } - close(ch) - return lister{ch}, nil -} - -// WatchAll is Watch of every key. -func (p *Picture) WatchAll(ctx context.Context, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { - return p.Watch(ctx, ">", opts...) -} - -// Watch hands, once the first view is answered, every current value of the keys matching (a subject pattern: `*` one -// token, `>` the rest), then nil, then each change, as the bus's watch does. Options are not read: a picture holds no -// history, and deletes are always handed. -func (p *Picture) Watch(ctx context.Context, keys string, _ ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { - w := &watcher{ch: make(chan jetstream.KeyValueEntry, 64), match: keys, wake: make(chan struct{}, 1), stop: make(chan struct{})} - go func() { - defer close(w.ch) - if p.ready(ctx) != nil { - return - } - p.mu.Lock() - p.watchers = append(p.watchers, w) - p.mu.Unlock() - defer p.unwatch(w) - vals := p.f.Values() - names := make([]string, 0, len(vals)) - for k := range vals { - names = append(names, k) - } - slices.Sort(names) - p.mu.Lock() - first := make([]jetstream.KeyValueEntry, 0, len(names)+1) - for _, k := range names { - if matches(keys, k) { - first = append(first, entry{bucket: p.bucket, key: k, value: vals[k], rev: p.revs[k], op: jetstream.KeyValuePut}) - } - } - p.mu.Unlock() - first = append(first, nil) - w.run(ctx, first) - }() - return w, nil -} - -func (p *Picture) unwatch(w *watcher) { - p.mu.Lock() - defer p.mu.Unlock() - p.watchers = slices.DeleteFunc(p.watchers, func(x *watcher) bool { return x == w }) -} - -// The writes and the reads a picture cannot answer are refused, never handed to a bucket. -func (p *Picture) Put(context.Context, string, []byte) (uint64, error) { return 0, ErrReadOnly } -func (p *Picture) PutString(context.Context, string, string) (uint64, error) { return 0, ErrReadOnly } -func (p *Picture) Create(context.Context, string, []byte, ...jetstream.KVCreateOpt) (uint64, error) { - return 0, ErrReadOnly -} -func (p *Picture) Update(context.Context, string, []byte, uint64) (uint64, error) { - return 0, ErrReadOnly -} -func (p *Picture) Delete(context.Context, string, ...jetstream.KVDeleteOpt) error { return ErrReadOnly } -func (p *Picture) Purge(context.Context, string, ...jetstream.KVDeleteOpt) error { return ErrReadOnly } -func (p *Picture) PurgeDeletes(context.Context, ...jetstream.KVPurgeOpt) error { return ErrReadOnly } -func (p *Picture) GetRevision(context.Context, string, uint64) (jetstream.KeyValueEntry, error) { - return nil, errors.New("a first view holds no history") -} -func (p *Picture) History(context.Context, string, ...jetstream.WatchOpt) ([]jetstream.KeyValueEntry, error) { - return nil, errors.New("a first view holds no history") -} -func (p *Picture) WatchFiltered(ctx context.Context, keys []string, opts ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { - if len(keys) != 1 { - return nil, errors.New("a first view watches one pattern") - } - return p.Watch(ctx, keys[0], opts...) -} -func (p *Picture) ListKeysFiltered(ctx context.Context, filters ...string) (jetstream.KeyLister, error) { - keys, err := p.Keys(ctx) - if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) { - return nil, err - } - ch := make(chan string, len(keys)) - for _, k := range keys { - if len(filters) == 0 || slices.ContainsFunc(filters, func(f string) bool { return matches(f, k) }) { - ch <- k - } - } - close(ch) - return lister{ch}, nil -} -func (p *Picture) Status(context.Context) (jetstream.KeyValueStatus, error) { - return nil, errors.New("a first view has no bucket status: the owner's read tool says its fill") -} - -// matches says whether key matches a subject pattern of the bus's watch. -func matches(pattern, key string) bool { - if pattern == "" || pattern == ">" { - return true - } - pt, kt := strings.Split(pattern, "."), strings.Split(key, ".") - for i, t := range pt { - if t == ">" { - return len(kt) > i - } - if i >= len(kt) || (t != "*" && t != kt[i]) { - return false - } - } - return len(pt) == len(kt) -} - -// watcher hands a watch's entries in order: the current values first, then each change heard, never blocking the -// follower (the changes wait in a queue of its own). -type watcher struct { - ch chan jetstream.KeyValueEntry - match string - wake chan struct{} - stop chan struct{} - once sync.Once - - mu sync.Mutex - queue []jetstream.KeyValueEntry -} - -func (w *watcher) send(e entry) { - if !matches(w.match, e.key) { - return - } - w.mu.Lock() - w.queue = append(w.queue, e) - w.mu.Unlock() - select { - case w.wake <- struct{}{}: - default: - } -} - -func (w *watcher) run(ctx context.Context, first []jetstream.KeyValueEntry) { - out := func(e jetstream.KeyValueEntry) bool { - select { - case w.ch <- e: - return true - case <-ctx.Done(): - return false - case <-w.stop: - return false - } - } - for _, e := range first { - if !out(e) { - return - } - } - for { - w.mu.Lock() - q := w.queue - w.queue = nil - w.mu.Unlock() - for _, e := range q { - if !out(e) { - return - } - } - select { - case <-w.wake: - case <-ctx.Done(): - return - case <-w.stop: - return - } - } -} - -func (w *watcher) Updates() <-chan jetstream.KeyValueEntry { return w.ch } - -func (w *watcher) Stop() error { - w.once.Do(func() { close(w.stop) }) - return nil -} - -type lister struct{ ch chan string } - -func (l lister) Keys() <-chan string { return l.ch } -func (l lister) Stop() error { return nil } - -type entry struct { - bucket, key string - value []byte - rev uint64 - at time.Time - op jetstream.KeyValueOp -} - -func (e entry) Bucket() string { return e.bucket } -func (e entry) Key() string { return e.key } -func (e entry) Value() []byte { return e.value } -func (e entry) Revision() uint64 { return e.rev } -func (e entry) Created() time.Time { return e.at } -func (e entry) Delta() uint64 { return 0 } -func (e entry) Operation() jetstream.KeyValueOp { return e.op } - -// ViewsStream is the bus name of owner's views family (views.IndexerStream, views.AssignerStream): `-views` -// as the owner's old `streams` section names it. -func ViewsStream(owner string) string { - return catalogue.StreamBusName(strings.TrimPrefix(owner, "mesh-delivery-") + "-views") -} - -// Start hears each owner's views family on the process's durable consumer of it (its manifest's `consumes` -// `.views.`), keeps each status in the process's state `first-views` and hears the hand asks the tools -// bundle's `_first_view_again` writes there (views.StatusTools), then asks every first view (Ask). A consumer -// the bus does not have yet is said: the picture is then the first view's alone, asked again on a gap or by hand. -func (r *Reader) Start(ctx context.Context, js jetstream.JetStream, cred process.Account, status *process.AwaitedBucket) { - for _, owner := range r.Owners() { - stream := ViewsStream(owner) - cons, err := cred.EventConsumer(ctx, js, stream) - if err == nil { - _, err = process.HearEvents(cons, r.Hear(ctx, owner), r.logf) - } - if err != nil { - r.logf("the consumer %s on %s is not heard (%v): %s's states are the first view's alone, asked again by hand, until the %s starts again", - cred.EventConsumerName(), stream, err, owner, r.Module) - } - } - var said func(views.Status) - if status != nil { - status.Ready(func(kv jetstream.KeyValue) { - if err := views.HearAgain(ctx, kv, r.Followers(), r.logf); err != nil { - r.logf("views: the hand asks of %s's first views are not heard (%v) until it starts again", r.Module, err) - } - }) - said = func(s views.Status) { - if kv := status.KV(); kv != nil { - views.KeepStatus(ctx, kv, r.logf)(s) - } - } - } - r.Ask(ctx, said) -} diff --git a/sdk/firstview/firstviewtest/owner.go b/sdk/firstview/firstviewtest/owner.go deleted file mode 100644 index 761e13ad..00000000 --- a/sdk/firstview/firstviewtest/owner.go +++ /dev/null @@ -1,109 +0,0 @@ -// Package firstviewtest is an owner of a state as a reader's whole-module test needs it (novox/hq ADR 0615): the -// owner's read tool answering the first view from its bucket at its mark, and its views family said by the real -// views emitter onto its views stream, which the reader's durable consumer reads. -package firstviewtest - -import ( - "context" - "encoding/json" - "errors" - "strings" - "testing" - "time" - - "github.com/nats-io/nats.go" - "github.com/nats-io/nats.go/jetstream" - - "git.novox.be/novox/mesh-sdk/go/process" - "git.novox.be/novox/mesh-sdk/go/views" - - "git.novox.be/novox/mesh-delivery/sdk/firstview" -) - -// Serve is owner's state on the bus with the owner's side of it: the bucket `_` (made when absent, history -// 1), the owner's marks, its views stream with a durable consumer per reader (`_`), the emitter saying -// each change, and the read tool (firstview.ReadTool) answering on nc. It returns the bucket the test writes. -func Serve(t testing.TB, ctx context.Context, js jetstream.JetStream, nc *nats.Conn, owner, state string, readers ...string) jetstream.KeyValue { - t.Helper() - bucket := owner + "_" + state - kv, err := js.KeyValue(ctx, bucket) - if err != nil { - if kv, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 1}); err != nil { - t.Fatal(err) - } - } - marks, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: owner + "_" + views.MarksState, History: 1}) - if err != nil { - t.Fatal(err) - } - stream := firstview.ViewsStream(owner) - if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{process.ModuleStreamSubject(owner, views.EventPrefix+">")}}); err != nil { - t.Fatal(err) - } - for _, r := range readers { - if _, err := js.CreateOrUpdateConsumer(ctx, stream, jetstream.ConsumerConfig{Durable: r, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { - t.Fatal(err) - } - } - em := views.Emitter{Owner: owner, Subject: func(e string) string { return process.ModuleStreamSubject(owner, e) }, - Publish: views.JetStreamPublish(js), Marks: marks} - if err := em.Start(ctx, state, kv, nil); err != nil { - t.Fatal(err) - } - tool := strings.TrimPrefix(firstview.ReadTool(owner, state), owner+".") - sub, err := nc.Subscribe("mesh.mod."+owner+".call."+tool+".>", func(m *nats.Msg) { - var in struct { - After string `json:"after"` - } - _ = json.Unmarshal(m.Data, &in) - a, err := views.FirstView(owner, state, Source{ctx, kv}, Source{ctx, marks}, in.After, time.Now()) - out := map[string]any{"result": a, "node": "n0"} - if err != nil { - out = map[string]any{"error": err.Error()} - } - raw, _ := json.Marshal(out) - _ = m.Respond(raw) - }) - if err != nil { - t.Fatal(err) - } - t.Cleanup(func() { _ = sub.Unsubscribe() }) - return kv -} - -// Grants are the subjects a reader's account publishes to follow owner's state (its `invokes` and `consumes`): the -// read tool's call subjects, its durable consumer's info, next and acknowledgements on the owner's views stream. -func Grants(owner, state, durable string) []string { - tool := strings.TrimPrefix(firstview.ReadTool(owner, state), owner+".") - stream := firstview.ViewsStream(owner) - return []string{"mesh.mod." + owner + ".call." + tool + ".>", "mesh.mod." + owner + ".call." + tool, - "$JS.API.CONSUMER.INFO." + stream + "." + durable, "$JS.API.CONSUMER.MSG.NEXT." + stream + "." + durable, - "$JS.ACK." + stream + "." + durable + ".>"} -} - -// Source is a bucket as a views.Source. -type Source struct { - Ctx context.Context - KV jetstream.KeyValue -} - -// Keys is every key; none is no error. -func (s Source) Keys() ([]string, error) { - keys, err := s.KV.Keys(s.Ctx) - if errors.Is(err, jetstream.ErrNoKeysFound) { - return nil, nil - } - return keys, err -} - -// Get is one key's value and revision. -func (s Source) Get(key string) (*views.Entry, bool, error) { - e, err := s.KV.Get(s.Ctx, key) - if errors.Is(err, jetstream.ErrKeyNotFound) { - return nil, false, nil - } - if err != nil { - return nil, false, err - } - return &views.Entry{Value: e.Value(), Revision: e.Revision()}, true, nil -} diff --git a/sdk/go.mod b/sdk/go.mod index 84b39a5d..358b5f55 100644 --- a/sdk/go.mod +++ b/sdk/go.mod @@ -2,13 +2,11 @@ module git.novox.be/novox/mesh-delivery/sdk go 1.26.0 -require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 - github.com/nats-io/nats.go v1.54.0 -) +require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 require ( github.com/klauspost/compress v1.20.0 // indirect + github.com/nats-io/nats.go v1.54.0 // indirect github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect