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",