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