From 8afd34525d33256f467f6b0f3a4c0e1b8679cbc9 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 14:24:17 +0100 Subject: [PATCH] DLV-M4, DLV-V1: the view reads the indexer's and the assigner's twelve states through their read tools, never their buckets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ADR 0615 §1, design 25 § 12.2: the manifest drops its twelve reads of the two owners' states and invokes their twelve read tools. The process follows the six it watches (placement, wants, asks, candidates, modules, taken) by the first view and then the views families (consumes), through firstview pictures, so its watch code is unchanged; an owner that is down holds no other watch, and the unanswered first view is the view's finding (states first-views and findings; tools view_first_views, view_first_view_again). The tools keep their names, arguments and answers and ask the owners' read tools on each call (view.OwnerRead; a read of many keys is one paged values read). DLV-V1: a table subtest answers each tool through the owner's own Read over a fake state and matches what reading the state answered. The owners' read tools answer no history of a key: indexer_module_history, indexer_heard and view_tiers' times say so (an error, never none) until the read tool takes a history argument (a request to the join). Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../cmd/mesh-delivery-view-tools/main.go | 43 ++++-- .../cmd/mesh-delivery-view-tools/main_test.go | 1 + .../mesh-delivery-view-tools/owners_test.go | 131 ++++++++++++++++++ .../mesh-delivery-view/integration_test.go | 28 +++- .../cmd/mesh-delivery-view/main.go | 54 +++++++- .../mesh-delivery-view/internal/view/owner.go | 111 +++++++++++++++ .../mesh-delivery-view/internal/view/read.go | 13 ++ .../internal/view/status.go | 3 + modules/mesh-delivery-view/manifest_test.go | 23 ++- modules/mesh-delivery-view/module.json | 42 ++++-- 10 files changed, 410 insertions(+), 39 deletions(-) create mode 100644 modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go create mode 100644 modules/mesh-delivery-view/internal/view/owner.go 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 746dc902..af1a071e 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 @@ -14,8 +14,10 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" + "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" stdio "git.novox.be/novox/mesh-sdk/go" ) @@ -311,20 +313,30 @@ func read(state stdio.KeptState) view.StateRead { }} } +// ownersRead is the view's reading of the indexer's and the assigner's states: each through its owner's read tool, asked +// 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)} +} + func main() { - r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates")), - WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have")), - ReachState: read(stdio.State("reach")), - HeardState: read(stdio.State("mesh-delivery-indexer.heard")), - SignersState: read(stdio.State("mesh-vault.signers")), - ConditionsState: read(stdio.State(view.StatusConditionsState)), AttemptsState: read(stdio.State(view.StatusAttemptsState)), - PresenceState: read(stdio.State(view.StatusPresenceState)), StuckState: read(stdio.State(view.StatusStuckState)), - PredefinedState: read(stdio.State(delivery.IndexerModule + "." + delivery.IndexerPredefinedState)), - ConfigCandidatesState: read(stdio.State(delivery.IndexerModule + "." + delivery.IndexerConfigCandidatesState)), - PackagesState: read(stdio.State(delivery.IndexerModule + "." + delivery.PackagesState)), - GraphState: read(stdio.State(graph.Read)), SourceState: read(stdio.State(delivery.GateModule + ".source")), - ProvisionsState: read(stdio.State(provisions.ProvisionsRead)), - BuildLog: stdio.Log(delivery.BuildLogRead).Since} + r := ownersRead(stdio.Ask) + r.Builds, r.HavesState = read(stdio.State("mesh-delivery-builder.builds")), read(stdio.State("node-engine.have")) + r.ReachState = read(stdio.State("reach")) + r.SignersState = read(stdio.State("mesh-vault.signers")) + r.ConditionsState, r.AttemptsState = read(stdio.State(view.StatusConditionsState)), read(stdio.State(view.StatusAttemptsState)) + r.PresenceState, r.StuckState = read(stdio.State(view.StatusPresenceState)), read(stdio.State(view.StatusStuckState)) + r.SourceState = read(stdio.State(delivery.GateModule + ".source")) + r.ProvisionsState = read(stdio.State(provisions.ProvisionsRead)) + r.BuildLog = stdio.Log(delivery.BuildLogRead).Since u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")), Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())} for _, lang := range delivery.AskLanguages { @@ -332,7 +344,10 @@ func main() { u.Steps[lang] = read(stdio.State(module + "." + delivery.StepsState)) u.Pauses[lang] = read(stdio.State(module + "." + delivery.PausesState)) } - if err := stdio.Serve("", Tools(r, u)); err != nil { + // 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)...) + 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/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go index 1520dfc2..217e3425 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go @@ -56,6 +56,7 @@ const ( func TestModuleToolAnswersTheStoredValue(t *testing.T) { t.Run("indexer_packages answers the published modules", packagesToolAnswersThePublishedModules) t.Run("view_tiers names each module's wait", tiersToolSaysEachModulesWait) + t.Run("each tool answers through the owners' read tools as before", ownersAnswerTheToolsAsBefore) r := fake{"a": withFP2, "b": withNone}.reader() for key, want := range map[string]string{"a": withFP2, "b": withNone} { out, err := run(t, r, view.ModuleReadTool, map[string]any{"module": key}) 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 new file mode 100644 index 00000000..e3b27130 --- /dev/null +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/owners_test.go @@ -0,0 +1,131 @@ +package main + +import ( + "encoding/json" + "fmt" + "sort" + "strings" + "testing" + "time" + + "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" +) + +// source is a fake owner state as a views.Source: what the owner's read tool answers from. +type source map[string]string + +func (s source) Keys() ([]string, error) { + var out []string + for k := range s { + out = append(out, k) + } + sort.Strings(out) + return out, nil +} + +func (s source) Get(key string) (*views.Entry, bool, error) { + v, ok := s[key] + if !ok { + return nil, false, nil + } + return &views.Entry{Value: json.RawMessage(v), Revision: uint64(len(key))}, true, nil +} + +// ownersAnswerTheToolsAsBefore is DLV-V1: every view tool over the indexer's and the assigner's states answers through +// the owners' read tools (the real owner side, views.Read, over a fake state) what it answered reading the states. +func ownersAnswerTheToolsAsBefore(t *testing.T) { + cand := `{"v":1,"group":"feat/x","base":3,"heads":"h","pulls":[{"modules":["zsh"],"fps":{"zsh":{"fp2":"` + fp2B + `","manifest":"m"}}}]}` + pkgKey, err := delivery.PackagesKey("git.novox.be/novox/mesh-sdk/go") + if err != nil { + t.Fatal(err) + } + states := map[string]source{ + delivery.IndexerModule + ".modules": {"a": withFP2, "b": withNone}, + delivery.IndexerModule + ".candidates": {"feat/x": cand, "fix": `{"v":1,"group":"fix","base":3,"heads":"h","pulls":[{"modules":["netcheck"]}]}`}, + delivery.AssignerModule + ".wants": {"want.ace.zsh": `{"fp":"` + fp2A + `"}`, "want.novox.zsh": `{"fp":"` + fp2B + `"}`, + "held." + keyB: `{"node":"ace","words":"failed"}`}, + delivery.AssignerModule + ".differences": {"ace.zsh": `{"unexplained":true}`, "novox.zsh": `{"rule":3}`}, + delivery.IndexerModule + "." + delivery.IndexerPredefinedState: {delivery.PredefinedListKey: `{"v":1,"keys":["dir:state"]}`}, + delivery.IndexerModule + "." + delivery.IndexerConfigCandidatesState: {"novox.r.7": `{"v":1,"pull":"novox/r#7"}`}, + delivery.IndexerModule + "." + delivery.PackagesState: {pkgKey: `{"v":1,"module":"git.novox.be/novox/mesh-sdk/go","commit":"c1"}`}, + delivery.IndexerModule + "." + graph.State: {}, + } + // The direct reading, as the tools read the states before (the reads the manifest no longer has). + st := func(name string) view.StateRead { return fake(states[name]).reader().State } + direct := view.Reader{State: st(delivery.IndexerModule + ".modules"), CandidatesState: st(delivery.IndexerModule + ".candidates"), + WantsState: st(delivery.AssignerModule + ".wants"), DifferencesState: st(delivery.AssignerModule + ".differences"), + PredefinedState: st(delivery.IndexerModule + "." + delivery.IndexerPredefinedState), + ConfigCandidatesState: st(delivery.IndexerModule + "." + delivery.IndexerConfigCandidatesState), + PackagesState: st(delivery.IndexerModule + "." + delivery.PackagesState), + GraphState: st(delivery.IndexerModule + "." + graph.State)} + // The owners' read tools, answered by the owner's own Read over the same state; any other address is refused. + tools := map[string]string{} + for name := range states { + owner, state, _ := strings.Cut(name, ".") + tools[firstview.ReadTool(owner, state)] = name + } + asked := map[string]int{} + ask := func(key string, body any) (json.RawMessage, error) { + name, ok := tools[key] + if !ok { + return nil, fmt.Errorf("no tool %s", key) + } + asked[key]++ + args, _ := body.(map[string]any) + k, _ := args["key"].(string) + all, _ := args["values"].(bool) + after, _ := args["after"].(string) + owner, state, _ := strings.Cut(name, ".") + a, err := views.Read(owner, state, states[name], nil, k, all, after, time.Unix(0, 0)) + if err != nil { + return nil, err + } + return json.Marshal(a) + } + owned := ownersRead(ask) + for _, row := range []struct { + tool string + args map[string]any + }{ + {view.ModuleReadTool, map[string]any{"module": "a"}}, + {view.ModuleReadTool, map[string]any{"module": "gone"}}, + {view.ModulesReadTool, map[string]any{}}, + {view.ModulesReadTool, map[string]any{"none2": true}}, + {view.CandidateReadTool, map[string]any{"group": "feat/x"}}, + {view.CandidatesReadTool, map[string]any{}}, + {view.CandidatesReadTool, map[string]any{"module": "zsh"}}, + {view.WantsReadTool, map[string]any{}}, + {view.WantsReadTool, map[string]any{"node": "ace"}}, + {view.WantReadTool, map[string]any{"node": "ace", "module": "zsh"}}, + {view.DifferencesReadTool, map[string]any{"node": "novox"}}, + {view.PredefinedReadTool, map[string]any{}}, + {view.ConfigCandidatesReadTool, map[string]any{}}, + {view.ConfigCandidatesReadTool, map[string]any{"key": "novox.r.7"}}, + {view.PackagesReadTool, map[string]any{"module": "git.novox.be/novox/mesh-sdk/go"}}, + {view.PackagesReadTool, nil}, + } { + want, werr := run(t, direct, row.tool, row.args) + got, gerr := run(t, owned, row.tool, row.args) + wj, _ := json.Marshal(want) + gj, _ := json.Marshal(got) + if string(wj) != string(gj) || fmt.Sprint(werr) != fmt.Sprint(gerr) { + t.Errorf("%s %v through the owner's read tool = %s, %v; reading the state it was %s, %v", row.tool, row.args, gj, gerr, wj, werr) + } + } + // Each owner state the tools read was asked of its owner's read tool. + for tool, name := range tools { + if name != delivery.IndexerModule+"."+graph.State && asked[tool] == 0 { + t.Errorf("%s was never asked", tool) + } + } + // A history the owner's read tool does not answer is said, never answered as none. + if _, err := run(t, owned, view.ModuleHistoryTool, map[string]any{"module": "a"}); err == nil || !strings.Contains(err.Error(), "no history argument") { + t.Errorf("indexer_module_history with an owner answering no history: %v", err) + } +} 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 3cbf53e8..44cb4fac 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 @@ -20,9 +20,22 @@ 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" + "mesh-delivery-view/internal/view" ) +// serveOwners serves the six states of the indexer and the assigner the view's process follows by their first views +// (ADR 0615 §1, DLV-M4): each owner's read tool and views family, with the view's durable consumer (`n_mesh-delivery-view`). +// 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) + } +} + // TestProcessWholeSeesTheBuckets starts the program the way the node does (its entry, run, with its environment and a // broker file) against an embedded bus holding the three buckets, and asks its socket. func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8 @@ -99,6 +112,7 @@ func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8 env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) + serveOwners(t, ctx, js, nc) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) @@ -198,6 +212,7 @@ func TestProcessWholeShowsWhatTheWatchmanSays(t *testing.T) { // 1168/AC5 env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) + serveOwners(t, ctx, js, nc) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) @@ -264,6 +279,7 @@ func TestProcessWholeShowsTheIndexersBucketsOnceTheyExist(t *testing.T) { // 124 env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) + serveOwners(t, ctx, js, nc) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) @@ -273,11 +289,11 @@ func TestProcessWholeShowsTheIndexersBucketsOnceTheyExist(t *testing.T) { // 124 break } } - cands, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.CandidatesBucket, History: delivery.CandidatesHistory}) + cands, err := js.KeyValue(ctx, delivery.CandidatesBucket) if err != nil { t.Fatal(err) } - mods, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ModulesBucket, History: delivery.ModulesHistory}) + mods, err := js.KeyValue(ctx, delivery.ModulesBucket) if err != nil { t.Fatal(err) } @@ -350,6 +366,7 @@ func TestProcessWholeShowsTheOpenAsksAndDropsTheAnsweredOnes(t *testing.T) { // env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) + serveOwners(t, ctx, js, nc) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) @@ -440,10 +457,11 @@ func TestProcessWholeListsEveryDifference(t *testing.T) { // 1297/B8 AC env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) + serveOwners(t, ctx, js, nc) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) - kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerWantsBucket, History: delivery.AssignerWantsHistory}) + kv, err := js.KeyValue(ctx, delivery.AssignerWantsBucket) if err != nil { t.Fatal(err) } @@ -537,6 +555,7 @@ func TestProcessWholeWritesTheAnswerToAnAskNeedingNoProof(t *testing.T) { // 129 env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) + serveOwners(t, ctx, js, nc) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) @@ -688,6 +707,7 @@ func TestProcessWholeShowsTheBacklogOfBuildAsks(t *testing.T) { // 1887/A8a AC env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) + serveOwners(t, ctx, js, nc) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) @@ -755,6 +775,7 @@ func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) { env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": filepath.Join(dir, "run", "cli.sock"), "MESH_DELIVERY_WEB": "127.0.0.1:0"} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) + serveOwners(t, ctx, js, nc) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) @@ -851,6 +872,7 @@ func TestProcessWholeSaysModulesInMotion(t *testing.T) { env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": filepath.Join(dir, "run", "cli.sock"), "MESH_DELIVERY_WEB": "127.0.0.1:0"} runCtx, stop := context.WithCancel(ctx) done := make(chan error, 1) + serveOwners(t, ctx, js, nc) go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }() t.Cleanup(func() { stop(); <-done }) time.Sleep(2 * time.Second) // the view has read what stood: nothing 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 a9953e7a..ae328646 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -25,6 +25,9 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "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" ) @@ -106,13 +109,31 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err }) model.SetBacklog(view.BacklogOver(js, 5*time.Second)) + // The indexer's and the assigner's states are their owners': read by the first view of the owner's read tool and then + // its views family, never their buckets (ADR 0615 §1, DLV-M4). An owner that is down never holds the start: a watch of + // its picture waits for the first view on its own goroutine, and until then the unanswered first view is the view's + // finding (ADR 0615 §6). + calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) + if err != nil { + return err + } + defer calls.Close() + reader := &firstview.Reader{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) + } + 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} { - go follow(ctx, js, bucket, false, model, logger) + go follow(ctx, open, bucket, false, model, logger) } // The modules' and the source's history give each fingerprint's first commit and each head's source revision (ADR 0556). + // The modules' picture holds no history: from its first view on, each change it takes is the history (finding of DLV-M4). for _, bucket := range []string{delivery.ModulesBucket, delivery.SourceBucket} { - go follow(ctx, js, bucket, true, model, logger) + go follow(ctx, open, bucket, true, model, logger) } + reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "view_first_views reads nothing and view_first_view_again is not heard", logger.Printf)) select { case <-ctx.Done(): case err := <-webUp: @@ -152,9 +173,34 @@ func putReach(ctx context.Context, js jetstream.JetStream, w view.ReachWrite) er return err } +// The view's own states of its first views (views.StatusState) and of its findings, where the condition keeper raises a +// first view unanswered past its bound (ADR 0615 §6). +var ( + firstViewsBucket = delivery.ViewModule + "_" + views.StatusState + 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} + +// 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 +} + +func (b buckets) KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) { + if p := b.reader.Picture(bucket); p != nil { + return p, nil + } + return b.js.KeyValue(ctx, bucket) +} + // follow watches a bucket with its last values first and then each change, for as long as ctx lives: a bucket that is // not there yet, or a watch that ends, is tried again. -func follow(ctx context.Context, js jetstream.JetStream, bucket string, history bool, model *view.Model, logger *log.Logger) { +func follow(ctx context.Context, js process.BucketOpener, bucket string, history bool, model *view.Model, logger *log.Logger) { for ctx.Err() == nil { if err := watchOnce(ctx, js, bucket, history, model); err != nil && ctx.Err() == nil { logger.Printf("cannot watch %s, trying again: %v", bucket, err) @@ -166,7 +212,7 @@ func follow(ctx context.Context, js jetstream.JetStream, bucket string, history } } -func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, history bool, model *view.Model) error { +func watchOnce(ctx context.Context, js process.BucketOpener, bucket string, history bool, model *view.Model) error { kv, err := js.KeyValue(ctx, bucket) if err != nil { return err diff --git a/modules/mesh-delivery-view/internal/view/owner.go b/modules/mesh-delivery-view/internal/view/owner.go new file mode 100644 index 00000000..938dd138 --- /dev/null +++ b/modules/mesh-delivery-view/internal/view/owner.go @@ -0,0 +1,111 @@ +package view + +import ( + "encoding/json" + "fmt" + "time" +) + +// OwnerRead reads an owner's state through the owner's ordinary read tool (design 25 § 12.2, ADR 0615 §1, DLV-M4): the +// view's tools answer the indexer's and the assigner's states by asking it, never by reading their buckets. ask is the +// bundle's ask of another module's tool (mesh-sdk stdio Ask), tool the read tool as the manifest's `invokes` names it +// (`._read_`). With no argument the tool answers the keys; with key, the key's value and revision; +// with values, the values whole by key in pages; with key and history, the values it keeps for the key with their times. +func OwnerRead(ask func(key string, body any) (json.RawMessage, error), tool string) StateRead { + entry := func(key string) (json.RawMessage, uint64, error) { + raw, err := ask(tool, map[string]any{"key": key}) + if err != nil { + return nil, 0, fmt.Errorf("%s: %w", tool, err) + } + var a struct { + Found bool `json:"found"` + Value json.RawMessage `json:"value"` + Revision uint64 `json:"revision"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return nil, 0, fmt.Errorf("%s answered what is not a key's value: %w", tool, err) + } + if !a.Found || len(a.Value) == 0 { + return nil, 0, nil + } + return a.Value, a.Revision, nil + } + return StateRead{ + Keys: func() ([]string, error) { + raw, err := ask(tool, map[string]any{}) + if err != nil { + return nil, fmt.Errorf("%s: %w", tool, err) + } + var a struct { + Keys []string `json:"keys"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return nil, fmt.Errorf("%s answered what is not a state's keys: %w", tool, err) + } + return a.Keys, nil + }, + GetEntry: entry, + Get: func(key string) (json.RawMessage, error) { + v, _, err := entry(key) + return v, err + }, + All: func() (map[string]json.RawMessage, error) { + out := map[string]json.RawMessage{} + for after, pages := "", 0; ; pages++ { + if pages == maxOwnerPages { + return nil, fmt.Errorf("%s answered more than %d pages", tool, maxOwnerPages) + } + body := map[string]any{"values": true} + if after != "" { + body["after"] = after + } + raw, err := ask(tool, body) + if err != nil { + return nil, fmt.Errorf("%s: %w", tool, err) + } + var a struct { + Values map[string]json.RawMessage `json:"values"` + Next string `json:"next"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return nil, fmt.Errorf("%s answered what is not a page of values: %w", tool, err) + } + for k, v := range a.Values { + out[k] = v + } + if a.Next == "" || a.Next == after { + return out, nil + } + after = a.Next + } + }, + History: func(key string) ([]Revision, error) { + raw, err := ask(tool, map[string]any{"key": key, "history": true}) + if err != nil { + return nil, fmt.Errorf("%s: %w", tool, err) + } + var a struct { + History *[]struct { + Revision uint64 `json:"revision"` + At time.Time `json:"at"` + Op string `json:"op"` + Value json.RawMessage `json:"value"` + } `json:"history"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return nil, fmt.Errorf("%s answered what is not a key's history: %w", tool, err) + } + if a.History == nil { + return nil, fmt.Errorf("%s answers no history of a key: its owner's read tool takes no history argument yet", tool) + } + out := make([]Revision, 0, len(*a.History)) + for _, h := range *a.History { + out = append(out, Revision{Revision: h.Revision, At: h.At, Op: h.Op, Value: h.Value}) + } + return out, nil + }, + } +} + +// maxOwnerPages bounds one read of a state's values page by page (a page is at most 768 KiB, mesh-sdk views PageBytes). +const maxOwnerPages = 256 diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go index 11c31b07..d76ee9ca 100644 --- a/modules/mesh-delivery-view/internal/view/read.go +++ b/modules/mesh-delivery-view/internal/view/read.go @@ -33,6 +33,9 @@ type StateRead struct { GetEntry func(key string) (json.RawMessage, uint64, error) // History, when set, gives every value the state keeps for a key with its time (indexer_module_history, indexer_heard). History func(key string) ([]Revision, error) + // All, when set, gives every value by key in one read (an owner's read tool's pages, OwnerRead), so a read of many keys + // is not a call per key. + All func() (map[string]json.RawMessage, error) } // Reader answers the read tools from a modules state. @@ -579,6 +582,16 @@ func parallel(n int, read func(i int) error) error { // values reads the stored value of each key, in parallel, in the order of keys (nil for a key with none). func (s StateRead) values(keys []string) ([]json.RawMessage, error) { out := make([]json.RawMessage, len(keys)) + if s.All != nil { + all, err := s.All() + if err != nil { + return nil, err + } + for i, k := range keys { + out[i] = all[k] + } + return out, nil + } err := parallel(len(keys), func(i int) error { raw, err := s.Get(keys[i]) out[i] = raw diff --git a/modules/mesh-delivery-view/internal/view/status.go b/modules/mesh-delivery-view/internal/view/status.go index d8a08373..edc4d1ae 100644 --- a/modules/mesh-delivery-view/internal/view/status.go +++ b/modules/mesh-delivery-view/internal/view/status.go @@ -81,6 +81,9 @@ type StatusMoving struct { } func values(s StateRead) (map[string]json.RawMessage, error) { + if s.All != nil { + return s.All() + } if s.Keys == nil || s.Get == nil { return nil, fmt.Errorf("not read here") } diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index 4e65b88f..7ccdba13 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -51,6 +51,8 @@ type manifest struct { Version string `json:"version"` Slug string `json:"slug"` Reads []string `json:"reads"` + Consumes []string `json:"consumes"` + Invokes []string `json:"invokes"` Uses []string `json:"uses"` Tools []string `json:"tools"` Emits []string `json:"emits"` @@ -87,17 +89,29 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if m.Module != "mesh-delivery-view" { t.Errorf("module = %q", m.Module) } - want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-conditions.conditions", "mesh-config.resolved", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.placement", "mesh-delivery-assigner.wants", "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.config-candidates", "mesh-delivery-indexer.graph", "mesh-delivery-indexer.heard", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.packages", "mesh-delivery-indexer.predefined", "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-healers.attempts", "mesh-provisioner.provisions", "mesh-vault.signers", "node-engine.have"} // view_tiers reads the module graph and the provisioner's keys (ADR 0603 §6); view_status reads the keeper's conditions and the healers' attempts (1952/V1); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 + want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-conditions.conditions", "mesh-config.resolved", "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "mesh-healers.attempts", "mesh-provisioner.provisions", "mesh-vault.signers", "node-engine.have"} // no reads of the indexer's or the assigner's states (ADR 0615 §1, DLV-M4); view_tiers reads the module graph and the provisioner's keys (ADR 0603 §6); view_status reads the keeper's conditions and the healers' attempts (1952/V1); build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 got := slices.Clone(m.Reads) slices.Sort(got) if !reflect.DeepEqual(got, want) { t.Errorf("reads = %v, want exactly %v", got, want) } - for _, k := range []string{"consumes", "provides", "claims"} { + for _, k := range []string{"provides", "claims"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the view writes only its answers", k) } } + // The indexer's and the assigner's states are asked of their owners' read tools (design 25 § 12.2), the six the process + // follows then heard on their views families (ADR 0615 §1, §5; DLV-M4). + if want := []string{"mesh-delivery-assigner.assigner_read_asks", "mesh-delivery-assigner.assigner_read_differences", "mesh-delivery-assigner.assigner_read_placement", + "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"}; !reflect.DeepEqual(m.Invokes, want) { + t.Errorf("invokes = %v, want %v", m.Invokes, want) + } + if want := []string{"mesh-delivery-assigner.views.asks", "mesh-delivery-assigner.views.placement", "mesh-delivery-assigner.views.wants", + "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.modules", "mesh-delivery-indexer.views.taken"}; !reflect.DeepEqual(m.Consumes, want) { + t.Errorf("consumes = %v, want %v", m.Consumes, want) + } // The one event the view says: a source revision is delivered (1906/D9, ADR 0539 §17), derived by it from the wants and the haves. if !reflect.DeepEqual(m.Emits, []string{"delivered", "moving", "settled", "stuck"}) { t.Errorf("emits = %v, want exactly delivered, moving, settled and stuck (1906/D9, D10)", m.Emits) @@ -108,7 +122,8 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 } // The states the view writes: the operator's answers to the asks that need no proof (ADR 0528 clause 6, 1297/B9), which the // teller reads, and how far each repository reached (ADR 0556 clause 2, issue 1280), which the gate reads. - if want := []state{{Name: "answers", History: 5}, {Name: "reach", History: 1}}; !reflect.DeepEqual(m.State, want) { + // Its first views of the owners' states and the finding of one unanswered past its bound (ADR 0615 §6). + if want := []state{{Name: "answers", History: 5}, {Name: "reach", History: 1}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -119,7 +134,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 } // The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer // itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads. - if want := []string{"view_status", "indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "indexer_predefined", "indexer_config_candidates", "indexer_packages", "builder_build", "builder_builds", "builder_history", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "node_trusts", "build_usage", "view_reach", "indexer_module_history", "indexer_heard", "view_tiers"}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"view_status", "indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "indexer_predefined", "indexer_config_candidates", "indexer_packages", "builder_build", "builder_builds", "builder_history", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "node_trusts", "build_usage", "view_reach", "indexer_module_history", "indexer_heard", "view_tiers", "view_first_views", "view_first_view_again"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } for _, a := range m.Build.Artifacts { diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 55bbbfb6..6ff8b9ec 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -24,7 +24,9 @@ "view_reach", "indexer_module_history", "indexer_heard", - "view_tiers" + "view_tiers", + "view_first_views", + "view_first_view_again" ], "emits": [ "delivered", @@ -34,7 +36,31 @@ ], "state": [ { "name": "answers", "history": 5 }, - { "name": "reach", "history": 1 } + { "name": "reach", "history": 1 }, + { "name": "first-views", "history": 1 }, + { "name": "findings", "history": 1 } + ], + "consumes": [ + "mesh-delivery-assigner.views.asks", + "mesh-delivery-assigner.views.placement", + "mesh-delivery-assigner.views.wants", + "mesh-delivery-indexer.views.candidates", + "mesh-delivery-indexer.views.modules", + "mesh-delivery-indexer.views.taken" + ], + "invokes": [ + "mesh-delivery-assigner.assigner_read_asks", + "mesh-delivery-assigner.assigner_read_differences", + "mesh-delivery-assigner.assigner_read_placement", + "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" ], "reads": [ "mesh-build-go.pauses", @@ -44,23 +70,11 @@ "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", - "mesh-delivery-assigner.asks", - "mesh-delivery-assigner.differences", - "mesh-delivery-assigner.placement", - "mesh-delivery-assigner.wants", "mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", - "mesh-delivery-indexer.candidates", - "mesh-delivery-indexer.config-candidates", - "mesh-delivery-indexer.graph", - "mesh-delivery-indexer.heard", - "mesh-delivery-indexer.modules", - "mesh-delivery-indexer.packages", - "mesh-delivery-indexer.predefined", - "mesh-delivery-indexer.taken", "mesh-delivery-teller.answers", "mesh-conditions.conditions", "mesh-healers.attempts",