diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go index baa62b17..7361c41b 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/main.go @@ -192,7 +192,7 @@ func heldCounts(s stdio.KeptState) func() (index.OldSectionsCount, error) { } func main() { - if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), catalogTools()...)); err != nil { + if err := stdio.Serve("", append(Tools(askTakeIn(stdio.State(index.TakeInsState), 25*time.Second), time.Now, stdio.State(index.VerdictsState), heldCounts(stdio.State(index.OldSectionsState))), append(catalogTools(), viewTools(os.Getenv, keptSource, time.Now)...)...)); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-indexer-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go new file mode 100644 index 00000000..cf164ff7 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer-tools/views.go @@ -0,0 +1,31 @@ +package main + +import ( + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-delivery/sdk/views" + + "mesh-delivery-indexer/internal/index" +) + +// viewTools are the indexer's first-view tools (ADR 0621 §1, ADR 0622 §1), each answering only the callers its manifest +// writes into its env value (views.CallersEnv), and its ordinary read tools (design 25 § 12.2). Every one reads only the +// indexer's own states, so it answers as soon as the runtime of its own is connected (ADR 0621 §3). +func viewTools(getenv func(string) string, state func(string) views.Source, now func() time.Time) []stdio.Tool { + marks := state(views.MarksState) + var out []stdio.Tool + for _, t := range index.FirstViewTools { + out = append(out, views.FirstViewTool(delivery.IndexerModule, t.Tool, t.State, t.Words, + views.ParseCallers(getenv(views.CallersEnv(t.Tool))), state(t.State), marks, now)) + } + for _, t := range index.ReadTools { + out = append(out, views.ReadTool(delivery.IndexerModule, t.Tool, t.State, t.Words, state(t.State), marks, now)) + } + return out +} + +// keptSource is one of the indexer's states as its runtime reads it. +func keptSource(name string) views.Source { return views.KeptSource{State: stdio.State(name)} } diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go index a28b8db5..1234d824 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/integration_test.go @@ -5,11 +5,13 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "os" "path/filepath" "strings" + "sync/atomic" "testing" "time" @@ -20,6 +22,10 @@ import ( "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-delivery/sdk/views" + + "mesh-delivery-indexer/internal/index" ) // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the @@ -83,8 +89,115 @@ func TestIndexerWhole(t *testing.T) { // 1238/AC2, AC10 t.Errorf("%s#%d modules = %v, want [%s]", p.Repository, p.Number, p.Modules, want) } } + + // ADR 0615 §5 (DLV-O4): each change of the candidates is said on the indexer's views family, its value whole at its + // revision; a first view taken at the mark, followed by the events after it, holds the candidate as the bucket does; + // and a waiter on manifest-accepted never wakes on it. + views0 := waitViews(t, ctx, js, "candidates", "feat/x") + cur, err := candidates.Get(ctx, "feat/x") + if err != nil || views0.Revision > cur.Revision() || views0.Deleted || len(views0.Value) == 0 || !views0.Created.Equal(bucketCreated(t, ctx, candidates)) { + t.Errorf("views event = %+v, bucket at %d (%v)", views0, cur.Revision(), err) + } + marks, _ := js.KeyValue(ctx, index.MarksBucket) + f := &views.Follower{Owner: delivery.IndexerModule, Tool: "indexer_read_candidates", State: "candidates", + Ask: func(context.Context, string) (views.Answer, error) { + return views.FirstView(delivery.IndexerModule, "candidates", jsSource{ctx, candidates}, jsSource{ctx, marks}, "", time.Now()) + }} + if err := f.Start(ctx); err != nil { + t.Fatal(err) + } + for _, e := range viewsEvents(t, ctx, js, "candidates") { + if gap, why := f.Apply(e); gap { + t.Errorf("a gap after the first view: %s", why) + } + } + if v, ok := f.Get("feat/x"); !ok || string(v) != string(cur.Value()) { + t.Errorf("the follower holds %s, the bucket %s", v, cur.Value()) + } + if n := acceptedHeard.Load(); n != 0 { + t.Errorf("a waiter on manifest-accepted woke %d times on the views family", n) + } } +// jsSource is a bucket as a views.Source. +type jsSource struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (s jsSource) Keys() ([]string, error) { + k, err := s.kv.Keys(s.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return k, err +} + +func (s jsSource) 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 +} + +func bucketCreated(t *testing.T, ctx context.Context, kv jetstream.KeyValue) time.Time { + t.Helper() + st, err := kv.Status(ctx) + if err != nil { + t.Fatal(err) + } + return st.(*jetstream.KeyValueBucketStatus).StreamInfo().Created.UTC() +} + +// viewsEvents are every event the views family holds for state, in order, read message by message. +func viewsEvents(t *testing.T, ctx context.Context, js jetstream.JetStream, state string) []views.Event { + t.Helper() + st, err := js.Stream(ctx, views0Stream) + if err != nil { + t.Fatal(err) + } + info, err := st.Info(ctx) + if err != nil { + t.Fatal(err) + } + var out []views.Event + for seq := info.State.FirstSeq; seq > 0 && seq <= info.State.LastSeq; seq++ { + m, err := st.GetMsg(ctx, seq) + if err != nil || m.Subject != "mesh.stream.mesh-delivery-indexer.event."+views.EventName(state) { + continue + } + e, err := views.DecodeEvent(m.Data) + if err != nil { + t.Fatal(err) + } + out = append(out, e) + } + return out +} + +// waitViews waits for the views family's event of state's key. +func waitViews(t *testing.T, ctx context.Context, js jetstream.JetStream, state, key string) views.Event { + t.Helper() + for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) { + for _, e := range viewsEvents(t, ctx, js, state) { + if e.Key == key { + return e + } + } + } + t.Fatalf("no views event of %s.%s within 20 s", state, key) + return views.Event{} +} + +const views0Stream = "INDEXER_VIEWS" + +// acceptedHeard counts what a waiter on manifest-accepted heard. +var acceptedHeard atomic.Int64 + // startIndexer starts the bus, the stream FORGE with the indexer's durable consumer, the buckets, lets put fill // mesh-controller_builds, and runs the program. mesh-delivery-gate_source is never created. func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...func(context.Context, jetstream.JetStream)) (jetstream.JetStream, context.Context, jetstream.KeyValue, jetstream.KeyValue) { @@ -155,6 +268,20 @@ func startIndexer(t *testing.T, put func(context.Context, buildLog), before ...f Subjects: []string{buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog) + ".>"}, AllowDirect: true}); err != nil { t.Fatal(err) } + // The indexer's views family and its marks (ADR 0615 §5), as the grants part makes them from the old sections, and a + // waiter on manifest-accepted, which the views family never wakes. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: views0Stream, Subjects: []string{"mesh.stream.mesh-delivery-indexer.event.views.>"}}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: index.MarksBucket, History: 1}); err != nil { + t.Fatal(err) + } + acceptedHeard.Store(0) + waiter, err := ctl.Subscribe("mesh.mod.mesh-delivery-indexer.event.manifest-accepted", func(*nats.Msg) { acceptedHeard.Add(1) }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = waiter.Unsubscribe() }) put(ctx, buildLog{ctx: ctx, js: js}) for _, b := range before { b(ctx, js) diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 0d896889..1f0144bf 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -469,6 +469,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // What the indexer heard of FORGE (issue 1126 note 76) is bound when its state appears, as every state a delivery adds // (issue 1790): until then no message is written there, and the rest of the indexer runs. heard := awaitedStore{process.AwaitBucket(ctx, js, conn, index.HeardBucket, "no FORGE message the indexer handles is written for indexer_heard", logger.Printf)} + // Each change of the states its readers read is said on its views family (ADR 0615 §5), from the mark each state was + // said up to; a first view (the tools bundle) answers at that mark. + startViews(ctx, js, cred, process.AwaitBucket(ctx, js, conn, index.MarksBucket, "no change is said on the views family, so every reader waits on its first view", logger.Printf), + process.AwaitBucket(ctx, js, conn, findingsBucket, "no finding of the indexer is written", logger.Printf), + map[string]readyOf{"manifests": boundNow(manifests), "modules": boundNow(modules), "candidates": boundNow(candidates), + index.VerdictsState: awaited(verdicts), delivery.IndexerPredefinedState: awaited(predefined), + delivery.IndexerConfigCandidatesState: awaited(configCandidates), delivery.PackagesState: awaited(packages), + graphState: awaited(graph), index.HeardState: awaited(heard.b), "taken": awaited(taken)}, errs, logger.Printf) cc, err := process.HearEvents(cons, func(m jetstream.Msg) { handle(ctx, m, ix, heard, logger); ix.KickFill(ctx) }, logger.Printf) if err != nil { return err diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go new file mode 100644 index 00000000..7821d743 --- /dev/null +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/views.go @@ -0,0 +1,73 @@ +package main + +import ( + "context" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "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-delivery/sdk/graph" + "git.novox.be/novox/mesh-delivery/sdk/views" +) + +// findingsBucket is the indexer's own findings, which the condition keeper raises and clears (ADR 0564 §2): a value too +// large for its views event, and its first view of the assigner's placement unanswered past its bound. +var findingsBucket = health.FindingsBucket(delivery.IndexerModule) + +// readyOf binds a state's bucket: now for one bound at the start, once it appears for one awaited. +type readyOf func(func(jetstream.KeyValue)) + +func boundNow(kv jetstream.KeyValue) readyOf { return func(f func(jetstream.KeyValue)) { f(kv) } } + +func awaited(b *process.AwaitedBucket) readyOf { return b.Ready } + +// startViews says each change of the indexer's viewed states on its views family (ADR 0615 §5), once its marks state +// is bound: each state from where its mark says, as soon as its own bucket is bound. A watch that ends is sent on errs. +func startViews(ctx context.Context, js jetstream.JetStream, cred process.Account, marks, findings *process.AwaitedBucket, states map[string]readyOf, errs chan<- error, logf func(string, ...any)) { + tooLarge := func(state, key string, size int) { + kv := findings.KV() + if kv == nil { + return + } + k, v, err := views.TooLargeFinding(delivery.IndexerModule, state, key, size, time.Now()) + if err == nil { + _, err = kv.Put(ctx, k, v) + } + if err != nil { + logf("views: the finding of %s.%s too large is not written: %v", state, key, err) + } + } + marks.Ready(func(mk jetstream.KeyValue) { + em := views.Emitter{Owner: delivery.IndexerModule, Subject: cred.StreamSubject, Publish: views.JetStreamPublish(js), + Marks: mk, TooLarge: tooLarge, Logf: logf} + for _, state := range sortedStates(states) { + states[state](func(kv jetstream.KeyValue) { + if err := em.Start(ctx, state, kv, errs); err != nil { + logf("%v: its readers ask its first view at each change until the indexer starts again", err) + } + }) + } + }) +} + +func sortedStates(states map[string]readyOf) []string { + out := make([]string, 0, len(states)) + for s := range states { + out = append(out, s) + } + for i := range out { + for j := i + 1; j < len(out); j++ { + if out[j] < out[i] { + out[i], out[j] = out[j], out[i] + } + } + } + return out +} + +// graphState is the module graph's state, named apart from run's bucket of the same name. +const graphState = graph.State diff --git a/modules/mesh-delivery-indexer/internal/index/views.go b/modules/mesh-delivery-indexer/internal/index/views.go new file mode 100644 index 00000000..a66f9ef7 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/views.go @@ -0,0 +1,55 @@ +package index + +import ( + "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-delivery/sdk/graph" + "git.novox.be/novox/mesh-delivery/sdk/views" +) + +// ViewsStream is the indexer's views family on its old streams section (ADR 0615 §5): one event per changed key of +// each state in ViewedStates. +const ViewsStream = views.IndexerStream + +// MarksBucket is the indexer's state of what its views family said each state up to. +const MarksBucket = delivery.IndexerModule + "_" + views.MarksState + +// FirstViewTools are the indexer's first-view tools (ADR 0621 §1, ADR 0622 §1): the tool, the state it answers, and +// the words of what that state holds. Each answers only its callers list (CallersEnv), which the manifest writes. +var FirstViewTools = []ViewedTool{ + {Tool: "manifests", State: "manifests", Words: "every manifest the indexer keeps, by its digest."}, + {Tool: "modules", State: "modules", Words: "every module the indexer holds, its fingerprint, inputs and manifest digest."}, + {Tool: "verdicts", State: VerdictsState, Words: "the judge's verdict on each manifest: accepted, waiting or refused."}, +} + +// ReadTools are the indexer's ordinary read tools, one per state its readers read (design 25 § 12.2). +var ReadTools = []ViewedTool{ + {Tool: "indexer_read_manifests", State: "manifests", Words: "every manifest the indexer keeps, by its digest."}, + {Tool: "indexer_read_modules", State: "modules", Words: "every module the indexer holds, its fingerprint, inputs and manifest digest."}, + {Tool: "indexer_read_predefined", State: delivery.IndexerPredefinedState, Words: "the predefined config values of each module on the trunk."}, + {Tool: "indexer_read_config_candidates", State: delivery.IndexerConfigCandidatesState, Words: "the config declarations of each open pull request."}, + {Tool: "indexer_read_candidates", State: "candidates", Words: "the candidate groups: the pull requests sharing a branch, their heads and the modules each touches."}, + {Tool: "indexer_read_packages", State: delivery.PackagesState, Words: "the published packages per series."}, + {Tool: "indexer_read_graph", State: graph.State, Words: "the module graph: each module's build and provision edges."}, + {Tool: "indexer_read_heard", State: HeardState, Words: "the forge messages the indexer heard and what it did with each."}, + {Tool: "indexer_read_taken", State: "taken", Words: "each repository's merge the indexer took in last."}, +} + +// ViewedTool is one tool over one state of the indexer's. +type ViewedTool struct { + Tool, State, Words string +} + +// ViewedStates are the states the indexer's views family says, each by its bucket: the ten its readers read. +var ViewedStates = map[string]string{ + "manifests": delivery.ManifestsBucket, + "modules": delivery.ModulesBucket, + VerdictsState: VerdictsBucket, + delivery.IndexerPredefinedState: PredefinedBucket, + delivery.IndexerConfigCandidatesState: ConfigCandidatesBucket, + "candidates": delivery.CandidatesBucket, + delivery.PackagesState: delivery.PackagesBucket, + graph.State: graph.Bucket, + HeardState: HeardBucket, + "taken": delivery.TakenBucket, +} diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 4911041b..dba12132 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -16,6 +16,10 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-delivery/sdk/graph" + "git.novox.be/novox/mesh-delivery/sdk/views" + "git.novox.be/novox/mesh-sdk/go/health" + + "mesh-delivery-indexer/internal/index" ) // The controller's manifest reader is mesh-controller internal/catalogue (Manifest, manifest.go: Consumes :521, @@ -29,19 +33,29 @@ type state struct { } type artifact struct { - Name string `json:"name"` - Kind string `json:"kind"` - Language string `json:"language"` - System string `json:"system"` - From string `json:"from"` - Binary string `json:"binary"` - Loads []string `json:"loads"` + Name string `json:"name"` + Kind string `json:"kind"` + Language string `json:"language"` + System string `json:"system"` + From string `json:"from"` + Binary string `json:"binary"` + Loads []string `json:"loads"` + Env map[string]string `json:"env"` +} + +type stream struct { + Name string `json:"name"` + Events []string `json:"events"` + MaxAgeDays int `json:"max-age-days"` + MaxMiB int `json:"max-mib"` } type manifest struct { - Module string `json:"module"` - Version string `json:"version"` - Slug string `json:"slug"` + Module string `json:"module"` + Version string `json:"version"` + Slug string `json:"slug"` + RunsAs string `json:"runs-as"` + Streams []stream `json:"streams"` Claims []struct { Name string `json:"name"` Scope string `json:"scope"` @@ -99,7 +113,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build } // 1126 note 76: heard keeps the last 64 FORGE messages the indexer handled, for indexer_heard. if want := []state{{Name: "candidates", History: 10}, {Name: "heard", History: 64}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "predefined", History: 5}, {Name: "config-candidates", History: 5}, {Name: "taken", History: 1}, {Name: "takeins", History: 1}, {Name: "verdicts", History: 5}, {Name: "oldsections", History: 1}, {Name: delivery.PackagesState, History: delivery.PackagesHistory}, - {Name: graph.State, History: graph.History}}; !reflect.DeepEqual(m.State, want) { + {Name: graph.State, History: graph.History}, {Name: views.MarksState, History: 1}, {Name: health.FindingsState, History: 5}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-delivery-assigner.placement", "mesh-delivery-builder.build-log", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { @@ -125,17 +139,51 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build // 1945: indexer_take_in, served by its tools bundle through the takeins state; the process serves none. The same bundle // reads the verdicts state (indexer_verdicts) and asks the indexer to judge again through the takeins state // (indexer_judge; issue 774, clause 3). 1966/AC11: indexer_old_sections reads the count of the old sections (state oldsections). - if want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge"}; !reflect.DeepEqual(m.Tools, want) { + // ADR 0621 §1, ADR 0622 §1: the first-view tools manifests, modules and verdicts, and design 25 § 12.2's nine read + // tools, all of the tools bundle, which runs in a runtime of its own on the indexer's account (runs-as). + want := []string{"indexer_take_in", "indexer_old_sections", "indexer_verdicts", "indexer_judge"} + for _, v := range append(append([]index.ViewedTool{}, index.FirstViewTools...), index.ReadTools...) { + want = append(want, v.Tool) + if _, ok := index.ViewedStates[v.State]; !ok { + t.Errorf("%s answers the state %s, which the views family does not say", v.Tool, v.State) + } + } + if !reflect.DeepEqual(m.Tools, want) || len(want) != 16 { t.Errorf("tools = %v, want %v", m.Tools, want) } + if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs { + t.Errorf("runs-as = %q, secrets-owner %q: the tools bundle runs on the indexer's own account (ADR 0621 §1)", m.RunsAs, m.SecretsOwner) + } + // ADR 0621 §2, ADR 0622 §1: each first-view tool answers exactly its readers; the indexer is on none. + callers := map[string]string{"manifests": "mesh-grants,node-tools,mesh-delivery-assigner", "modules": "mesh-grants,node-tools,mesh-delivery-assigner", "verdicts": "mesh-grants"} + for _, a := range m.Build.Artifacts { + if a.Name != "tools" { + continue + } + if len(a.Env) != len(callers) { + t.Errorf("tools env = %v, want only the three callers lists", a.Env) + } + for tool, list := range callers { + if got := a.Env[views.CallersEnv(tool)]; got != list { + t.Errorf("%s callers = %q, want %q", tool, got, list) + } + if views.ParseCallers(a.Env[views.CallersEnv(tool)]).Allows("n1." + m.Module) { + t.Errorf("%s answers the indexer itself", tool) + } + } + } + // ADR 0615 §5, ADR 0619 §2: its views family on its old streams section, 7 days and 256 MiB, dropping its oldest. + if want := []stream{{Name: index.ViewsStream, Events: []string{views.EventPattern}, MaxAgeDays: views.StreamDays, MaxMiB: views.StreamMiB}}; !reflect.DeepEqual(m.Streams, want) { + t.Errorf("streams = %+v, want %+v", m.Streams, want) + } // ADR 0604 §2: it holds the seat mesh-catalog, serving its four verbs from its tools bundle (which stays without a // request subject). if len(m.Claims) != 1 || m.Claims[0].Name != catalogue.CatalogSeat || m.Claims[0].Scope != catalogue.ScopeMesh || !slices.Equal(m.Claims[0].Serves, []string{"modules", "module", "provides", "dependents"}) { t.Errorf("claims = %+v, want mesh-catalog with its four verbs", m.Claims) } - // No streams, no bus, no other seat (ADR 0437 §4); it says no event. - for _, k := range []string{"streams", "logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { + // No bus, no other seat (ADR 0437 §4). + for _, k := range []string{"logs", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the indexer has none", k) } diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index 3fa9e8a7..d6a3f127 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -2,6 +2,7 @@ "module": "mesh-delivery-indexer", "version": "1", "slug": "indexer", + "runs-as": "mesh-delivery-indexer", "claims": [ { "name": "mesh-catalog", @@ -18,7 +19,19 @@ "indexer_take_in", "indexer_old_sections", "indexer_verdicts", - "indexer_judge" + "indexer_judge", + "manifests", + "modules", + "verdicts", + "indexer_read_manifests", + "indexer_read_modules", + "indexer_read_predefined", + "indexer_read_config_candidates", + "indexer_read_candidates", + "indexer_read_packages", + "indexer_read_graph", + "indexer_read_heard", + "indexer_read_taken" ], "consumes": [ "mesh-build-go.toolchain.reported", @@ -38,7 +51,12 @@ { "name": "verdicts", "history": 5 }, { "name": "oldsections", "history": 1 }, { "name": "packages", "history": 5 }, - { "name": "graph", "history": 5 } + { "name": "graph", "history": 5 }, + { "name": "view-marks", "history": 1 }, + { "name": "findings", "history": 5 } + ], + "streams": [ + { "name": "indexer-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 } ], "emits": [ "manifest-refused", @@ -115,7 +133,12 @@ "binary": "mesh-delivery-indexer-tools", "loads": [ "mesh-delivery-indexer-tools" - ] + ], + "env": { + "MESH_VIEW_CALLERS_MANIFESTS": "mesh-grants,node-tools,mesh-delivery-assigner", + "MESH_VIEW_CALLERS_MODULES": "mesh-grants,node-tools,mesh-delivery-assigner", + "MESH_VIEW_CALLERS_VERDICTS": "mesh-grants" + } } ] } diff --git a/sdk/go.mod b/sdk/go.mod index 7e692d40..84b39a5d 100644 --- a/sdk/go.mod +++ b/sdk/go.mod @@ -2,4 +2,15 @@ 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 +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 + github.com/nats-io/nats.go v1.54.0 +) + +require ( + github.com/klauspost/compress v1.20.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 + golang.org/x/sys v0.48.0 // indirect +) diff --git a/sdk/views/emit.go b/sdk/views/emit.go new file mode 100644 index 00000000..3552dbef --- /dev/null +++ b/sdk/views/emit.go @@ -0,0 +1,196 @@ +package views + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +// Emitter is an owner's process saying each change of its own states on its views family (ADR 0615 §5): it follows +// each state it is given from the revision its mark says it said up to, publishes one Event per change, and writes the +// mark once it has said every change the bus had handed it. A part watches only its own state (ADR 0596 §4): an owner +// follows its own buckets, never another's. +type Emitter struct { + // Owner is the owner module, said in every event. + Owner string + // Subject is where an event of the given local name lands on the owner's own stream (process.Account.StreamSubject). + Subject func(event string) string + // Publish sends one message to the owner's stream, its id the bus's dedup id. + Publish func(ctx context.Context, msg *nats.Msg, id string) error + // Marks is the owner's state MarksState; nil says nothing is marked, and every start says from the bucket's end. + Marks jetstream.KeyValue + // TooLarge is told of a value too large for an event (said as a gap): the owner's finding naming the key. Nil + // only logs it. + TooLarge func(state, key string, size int) + Logf func(string, ...any) + Now func() time.Time +} + +// JetStreamPublish is Emitter.Publish through js, with the bus's dedup id. +func JetStreamPublish(js jetstream.JetStream) func(context.Context, *nats.Msg, string) error { + return func(ctx context.Context, msg *nats.Msg, id string) error { + pub, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second) + defer cancel() + _, err := js.PublishMsg(pub, msg, jetstream.WithMsgID(id)) + return err + } +} + +func (e Emitter) now() time.Time { + if e.Now != nil { + return e.Now().UTC() + } + return time.Now().UTC() +} + +func (e Emitter) logf(format string, args ...any) { + if e.Logf != nil { + e.Logf(format, args...) + } +} + +// created is when the bucket's stream was made, and its fill now. +func created(ctx context.Context, kv jetstream.KeyValue) (time.Time, uint64, Fill, error) { + st, err := kv.Status(ctx) + if err != nil { + return time.Time{}, 0, Fill{}, err + } + b, ok := st.(*jetstream.KeyValueBucketStatus) + if !ok || b.StreamInfo() == nil { + return time.Time{}, 0, Fill{}, fmt.Errorf("the bus answered no stream behind %s", st.Bucket()) + } + info := b.StreamInfo() + return info.Created.UTC(), info.State.LastSeq, Fill{Keys: uint64(info.State.NumSubjects), Bytes: info.State.Bytes, MaxBytes: info.Config.MaxBytes}, nil +} + +// Start follows state's bucket kv: from the revision after its mark when the mark is of the same bucket (the same +// Created), otherwise from the bucket's end, writing a mark there first, so a reader's first view taken now misses +// nothing. It returns once the watch is set; the changes are said on a goroutine until ctx ends, and a watch that ends +// is sent on errs. +func (e Emitter) Start(ctx context.Context, state string, kv jetstream.KeyValue, errs chan<- error) error { + made, last, fill, err := created(ctx, kv) + if err != nil { + return fmt.Errorf("the state %s cannot be followed for its views: %w", state, err) + } + from := last + 1 + if m, ok := e.mark(ctx, state); ok && m.Created.Equal(made) && m.Revision <= last { + from = m.Revision + 1 + } else { + e.writeMark(ctx, state, Mark{State: state, Revision: last, Created: made, Fill: fill}) + } + w, err := kv.WatchAll(ctx, jetstream.ResumeFromRevision(from)) + if err != nil { + return fmt.Errorf("the state %s cannot be watched for its views: %w", state, err) + } + go func() { + defer w.Stop() + said := uint64(0) + for { + select { + case <-ctx.Done(): + return + case u, ok := <-w.Updates(): + if !ok { + if errs != nil { + errs <- fmt.Errorf("the watch of %s for its views ended", state) + } + return + } + if u != nil { + if err := e.say(ctx, state, made, u); err != nil { + // Not marked: the next start says it again from the mark, and the bus keeps it once by its id. + e.logf("views: %s.%s at %d is not said: %v", state, u.Key(), u.Revision(), err) + continue + } + said = u.Revision() + } + // Marked once every change handed so far is said, so a busy state writes one mark per burst. + if said > 0 && len(w.Updates()) == 0 { + _, _, fill, err := created(ctx, kv) + if err != nil { + e.logf("views: the fill of %s is not read: %v", state, err) + } + e.writeMark(ctx, state, Mark{State: state, Revision: said, Created: made, Fill: fill}) + said = 0 + } + } + } + }() + return nil +} + +// Event is the event saying u, a change of state made at made. +func (e Emitter) Event(state string, made time.Time, u jetstream.KeyValueEntry) Event { + ev := Event{V: Version, Owner: e.Owner, State: state, Key: u.Key(), Revision: u.Revision(), Created: made} + if u.Operation() != jetstream.KeyValuePut { + ev.Deleted = true + return ev + } + if n := len(u.Value()); n > MaxValueBytes { + ev.Gap = true + e.logf("views: %s.%s at %d holds %d bytes, more than an event carries: said as a gap, so a reader asks its first view", state, u.Key(), u.Revision(), n) + if e.TooLarge != nil { + e.TooLarge(state, u.Key(), n) + } + return ev + } + ev.Value = json.RawMessage(u.Value()) + if !json.Valid(ev.Value) { + ev.Value, _ = json.Marshal(string(u.Value())) + } + return ev +} + +func (e Emitter) say(ctx context.Context, state string, made time.Time, u jetstream.KeyValueEntry) error { + ev := e.Event(state, made, u) + data, err := ev.Encode() + if err != nil { + return err + } + msg := nats.NewMsg(e.Subject(EventName(state))) + msg.Header.Set("x-event-id", ev.ID()) + msg.Header.Set("x-source", e.Owner) + msg.Header.Set("x-time", e.now().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + return e.Publish(ctx, msg, ev.ID()) +} + +func (e Emitter) mark(ctx context.Context, state string) (Mark, bool) { + if e.Marks == nil { + return Mark{}, false + } + got, err := e.Marks.Get(ctx, state) + if err != nil { + if !errors.Is(err, jetstream.ErrKeyNotFound) { + e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err) + } + return Mark{}, false + } + m, err := DecodeMark(got.Value()) + if err != nil { + e.logf("views: the mark of %s is not read (%v): said from the bucket's end", state, err) + return Mark{}, false + } + return m, true +} + +func (e Emitter) writeMark(ctx context.Context, state string, m Mark) { + if e.Marks == nil { + return + } + m.V, m.At = Version, e.now() + data, err := json.Marshal(m) + if err != nil { + e.logf("views: the mark of %s is not written: %v", state, err) + return + } + if _, err := e.Marks.Put(ctx, state, data); err != nil { + e.logf("views: the mark of %s at %d is not written (%v): a first view answers the one before", state, m.Revision, err) + } +} diff --git a/sdk/views/findings.go b/sdk/views/findings.go new file mode 100644 index 00000000..9795fe10 --- /dev/null +++ b/sdk/views/findings.go @@ -0,0 +1,47 @@ +package views + +import ( + "fmt" + "strings" + "time" + + "git.novox.be/novox/mesh-sdk/go/health" +) + +// The findings a views part writes in its own state `findings`, which the condition keeper raises and clears +// (novox/hq ADR 0564 §2): an owner's value too large for an event (TooLargeKind), and a reader's first view unanswered +// past its bound (UnansweredKind). +const ( + TooLargeKind = "view-too-large" + UnansweredKind = "first-view-unanswered" +) + +// TooLargeFinding is the owner's finding of state's key too large to say whole on its views family. +func TooLargeFinding(owner, state, key string, size int, at time.Time) (string, []byte, error) { + id := owner + "." + state + "." + key + summary := fmt.Sprintf("%s's %s.%s holds %d bytes, more than one views event carries (%d): every reader asks its first view again at each change of it", owner, state, key, size, MaxValueBytes) + return finding(owner, TooLargeKind, id, summary, + fmt.Sprintf("A value of %s is too large for its views event", owner), + "Each change of this key is said as a gap, so every reader of the state asks the owner's first view once more. "+ + "It works, but costs a first view per change; the value should be made smaller or split.", at) +} + +// UnansweredFinding is a reader's finding of its first view of s unanswered past its bound (ADR 0615 §6): it names the +// owner and the tool. A first view answered at revision 0 with nothing in it is said too, with the operator's step. +func UnansweredFinding(reader string, s Status, at time.Time) (string, []byte, error) { + id := reader + "." + s.Owner + "." + s.State + summary := fmt.Sprintf("%s has no first view of %s's %s: %s has not answered since %s (%s)", reader, s.Owner, s.State, s.Tool, + s.Asked.Format(time.RFC3339), strings.TrimSpace(s.Error)) + return finding(reader, UnansweredKind, id, summary, + fmt.Sprintf("%s waits for %s", reader, s.Owner), + fmt.Sprintf("%s reads %s's %s only through %s, and keeps asking it; it never reads the state itself. The answer clears this. "+ + "Look at %s's runtime of its own: is it running and connected?", reader, s.Owner, s.State, s.Tool, s.Owner), at) +} + +func finding(observer, kind, id, summary, headline, explanation string, at time.Time) (string, []byte, error) { + key := health.FindingKey(health.ScopeModule, id, kind) + value, err := health.EncodeFinding(health.Finding{Key: key, Kind: kind, Subject: health.Subject{Scope: health.ScopeModule, ID: id}, + Severity: health.Warning, Summary: summary, Headline: headline, Explanation: explanation, + Evidence: []health.Evidence{{At: at.UTC(), Said: summary}}, Source: observer + "." + kind, Observer: observer, At: at.UTC()}) + return key, value, err +} diff --git a/sdk/views/follow.go b/sdk/views/follow.go new file mode 100644 index 00000000..df2a54de --- /dev/null +++ b/sdk/views/follow.go @@ -0,0 +1,341 @@ +package views + +import ( + "cmp" + "context" + "encoding/json" + "fmt" + "maps" + "slices" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// Follower is a reader's picture of one state of an owner (ADR 0615 §4 to §6): it asks the owner's first-view tool at +// its start, then keeps the picture from the owner's events alone, asking the tool once more only on a gap in revisions +// (said in its log), never on a timer. Started while the owner is down, it keeps asking, never reading the owner's state +// instead, and past Bound raises its condition naming the owner and the tool; the answer clears it. +type Follower struct { + Owner string + // Tool is the owner's tool it asks, as the reader's invokes names it (.). + Tool string + State string + // Ask calls the tool for the page after after. + Ask func(ctx context.Context, after string) (Answer, error) + // Raise and Clear are the reader's condition of this first view (past Bound; cleared by the answer). Nil says + // nothing. + Raise func(Status) + Clear func(Status) + // Changed is told of each key the picture changes (deleted: the key is gone), after the change is held. + Changed func(key string, value json.RawMessage, deleted bool) + Logf func(string, ...any) + Now func() time.Time + // Bound is how long it asks before it raises; zero is views.Bound. Wait is the first wait between two asks that + // failed, doubling to a minute; zero is five seconds. + Bound time.Duration + Wait time.Duration + + mu sync.Mutex + values map[string]json.RawMessage + st Status + asking bool + // during are the events heard while an ask runs: taken again after its answer, so none is lost to a picture read + // before it was written. + during []Event + ready chan struct{} +} + +// Status is what a reader's read tool says of one first view (design 25 § 12.2): the owner, the tool and the revision +// its picture holds, when it asked and when it was answered, how many asks, the hand acts among them, and whether its +// condition is raised. +type Status struct { + Owner string `json:"owner"` + Tool string `json:"tool"` + State string `json:"state"` + Revision uint64 `json:"revision"` + Created time.Time `json:"created,omitzero"` + Keys int `json:"keys"` + Asked time.Time `json:"asked,omitzero"` + Answered time.Time `json:"answered,omitzero"` + Asks int `json:"asks"` + ByHand int `json:"by-hand"` + Raised bool `json:"raised"` + // Why is the last reason it asked (start, a gap and which, a hand act) or why the last ask failed. + Why string `json:"why,omitempty"` + Error string `json:"error,omitempty"` +} + +func (f *Follower) now() time.Time { + if f.Now != nil { + return f.Now().UTC() + } + return time.Now().UTC() +} + +func (f *Follower) logf(format string, args ...any) { + if f.Logf != nil { + f.Logf(format, args...) + } +} + +func (f *Follower) init() { + if f.values == nil { + f.values = map[string]json.RawMessage{} + f.ready = make(chan struct{}) + f.st = Status{Owner: f.Owner, Tool: f.Tool, State: f.State} + } +} + +// Status is the follower's status now. +func (f *Follower) Status() Status { + f.mu.Lock() + defer f.mu.Unlock() + f.init() + s := f.st + s.Keys = len(f.values) + return s +} + +// Ready is closed once a first view was answered. +func (f *Follower) Ready() <-chan struct{} { + f.mu.Lock() + defer f.mu.Unlock() + f.init() + return f.ready +} + +// Values is a copy of the picture. +func (f *Follower) Values() map[string]json.RawMessage { + f.mu.Lock() + defer f.mu.Unlock() + f.init() + return maps.Clone(f.values) +} + +// Get is one key of the picture. +func (f *Follower) Get(key string) (json.RawMessage, bool) { + f.mu.Lock() + defer f.mu.Unlock() + f.init() + v, ok := f.values[key] + return v, ok +} + +// Start asks the first view until it is answered or ctx ends (ADR 0615 §6): each failed ask waits, doubling, and past +// Bound raises the condition. It returns once answered. +func (f *Follower) Start(ctx context.Context) error { return f.askUntil(ctx, "start", false) } + +// Again asks the first view once more by hand (a re-read tool, recorded by the caller as a hand act), until answered or +// ctx ends. +func (f *Follower) Again(ctx context.Context, why string) error { + return f.askUntil(ctx, "by hand: "+why, true) +} + +func (f *Follower) askUntil(ctx context.Context, why string, byHand bool) error { + f.mu.Lock() + f.init() + if f.asking { + f.mu.Unlock() + return nil // one ask at a time; the one running brings the picture to its newest + } + f.asking = true + f.during = nil + f.st.Why = why + if byHand { + f.st.ByHand++ + } + f.mu.Unlock() + defer func() { f.mu.Lock(); f.asking = false; f.mu.Unlock() }() + bound, wait := f.Bound, f.Wait + if bound == 0 { + bound = Bound + } + if wait == 0 { + wait = 5 * time.Second + } + first := f.now() + for { + err := f.askOnce(ctx) + if err == nil { + return nil + } + f.mu.Lock() + f.st.Error = err.Error() + raise := !f.st.Raised && f.now().Sub(first) >= bound + if raise { + f.st.Raised = true + } + s := f.st + f.mu.Unlock() + f.logf("the first view %s of %s's %s is not answered (%v): asked again in %s", f.Tool, f.Owner, f.State, err, wait) + if raise && f.Raise != nil { + f.Raise(s) + } + pause, cancel := context.WithTimeout(ctx, wait) // a context's bound between two asks, no timer of its own + <-pause.Done() + cancel() + if ctx.Err() != nil { + return ctx.Err() + } + if wait *= 2; wait > time.Minute { + wait = time.Minute + } + } +} + +// askOnce asks every page and, once whole, takes it as the picture. +func (f *Follower) askOnce(ctx context.Context) error { + f.mu.Lock() + f.st.Asked, f.st.Asks = f.now(), f.st.Asks+1 + f.mu.Unlock() + var first Answer + values := map[string]json.RawMessage{} + after := "" + for page := 0; ; page++ { + a, err := f.Ask(ctx, after) + if err != nil { + return err + } + if page == 0 { + first = a + } + maps.Copy(values, a.Values) + if a.Next == "" || a.Next == after { + break + } + after = a.Next + } + f.mu.Lock() + old := f.values + f.values = values + wasRaised := f.st.Raised + f.st.Revision, f.st.Created, f.st.Answered, f.st.Raised, f.st.Error = first.Revision, first.Created, f.now(), false, "" + // The events heard while it asked, after the answer's revision and in order, are taken again. + during := f.during + f.during = nil + slices.SortFunc(during, func(a, b Event) int { return cmp.Compare(a.Revision, b.Revision) }) + for _, ev := range during { + if ev.Created.Equal(f.st.Created) && ev.Revision == f.st.Revision+1 && !ev.Gap { + if ev.Deleted { + delete(f.values, ev.Key) + } else { + f.values[ev.Key] = ev.Value + } + f.st.Revision = ev.Revision + } + } + values = maps.Clone(f.values) + s := f.st + select { + case <-f.ready: + default: + close(f.ready) + } + f.mu.Unlock() + if wasRaised && f.Clear != nil { + f.Clear(s) + } + if f.Changed != nil { + for k, v := range values { + if ov, ok := old[k]; !ok || string(ov) != string(v) { + f.Changed(k, v, false) + } + } + for k := range old { + if _, ok := values[k]; !ok { + f.Changed(k, nil, true) + } + } + } + return nil +} + +// Apply takes one event of the owner's views family into the picture and says whether it was a gap: another Created +// (the owner's state made again), a revision past the one after the picture's, or a value too large to carry. A +// revision at or below the picture's is already in it (a durable consumer handing what it held) and is passed over. +// Before the first view is answered an event is passed over too: the answer holds it. +func (f *Follower) Apply(ev Event) (gap bool, why string) { + f.mu.Lock() + f.init() + if f.asking { + f.during = append(f.during, ev) + } + select { + case <-f.ready: + default: + f.mu.Unlock() + return false, "" + } + switch { + case !ev.Created.Equal(f.st.Created): + why = fmt.Sprintf("%s's %s was made again (or not marked when the first view was answered): created %s, held %s", f.Owner, f.State, ev.Created.Format(time.RFC3339), f.st.Created.Format(time.RFC3339)) + case ev.Revision <= f.st.Revision: + f.mu.Unlock() + return false, "" + case ev.Revision > f.st.Revision+1: + why = fmt.Sprintf("%s's %s at revision %d, the picture holds %d: %d changes missed", f.Owner, f.State, ev.Revision, f.st.Revision, ev.Revision-f.st.Revision-1) + case ev.Gap: + why = fmt.Sprintf("%s's %s.%s at revision %d was too large to carry", f.Owner, f.State, ev.Key, ev.Revision) + } + if why != "" { + f.mu.Unlock() + return true, why + } + if ev.Deleted { + delete(f.values, ev.Key) + } else { + f.values[ev.Key] = ev.Value + } + f.st.Revision = ev.Revision + f.mu.Unlock() + if f.Changed != nil { + f.Changed(ev.Key, ev.Value, ev.Deleted) + } + return false, "" +} + +// Hear is a reader's consumer of one owner's views family, routing each event to the follower of its state; a gap asks +// that follower's first view once more on its own goroutine, said in the log. A message it cannot read is terminated. +func Hear(ctx context.Context, followers map[string]*Follower, logf func(string, ...any)) func(jetstream.Msg) { + return func(m jetstream.Msg) { + ev, err := DecodeEvent(m.Data()) + if err != nil { + logf("views: %s is not a views event, dropped: %v", m.Subject(), err) + _ = m.Term() + return + } + f := followers[ev.State] + if f == nil { + _ = m.Ack() // a state this reader does not follow + return + } + if gap, why := f.Apply(ev); gap { + logf("views: a gap in %s (%s): its first view %s is asked once more", ev.State, why, f.Tool) + go func() { + f.mu.Lock() + f.init() + f.st.Why = "a gap: " + why + f.mu.Unlock() + _ = f.askUntil(ctx, "a gap: "+why, false) + }() + } + _ = m.Ack() + } +} + +// AskVia is a Follower's Ask through a process's call of the owner's tool (process.Conn.Call), decoding its answer. +func AskVia(call func(key string, body any) (json.RawMessage, error), tool string) func(context.Context, string) (Answer, error) { + return func(_ context.Context, after string) (Answer, error) { + body := map[string]any{} + if after != "" { + body["after"] = after + } + raw, err := call(tool, body) + if err != nil { + return Answer{}, err + } + return DecodeAnswer(raw) + } +} diff --git a/sdk/views/tools.go b/sdk/views/tools.go new file mode 100644 index 00000000..e14f3e3e --- /dev/null +++ b/sdk/views/tools.go @@ -0,0 +1,159 @@ +package views + +import ( + "encoding/json" + "fmt" + "slices" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// KeptSource is a state as a tools bundle reads it through its runtime (stdio.KeptState), as a Source. +type KeptSource struct{ State stdio.KeptState } + +// Keys is every key with a value. +func (k KeptSource) Keys() ([]string, error) { return k.State.Keys() } + +// Get is one key's value and revision. +func (k KeptSource) Get(key string) (*Entry, bool, error) { + e, err := k.State.Get(key) + if err != nil || e == nil { + return nil, false, err + } + return &Entry{Value: e.Value, Revision: e.Revision}, true, nil +} + +// ParseCallers is a callers list as a manifest writes it into an env value: comma-separated, each a module (on any +// machine), . or person. (stdio.Callers). +func ParseCallers(list string) stdio.Callers { + var out stdio.Callers + for _, c := range strings.Split(list, ",") { + if c = strings.TrimSpace(c); c != "" { + out = append(out, c) + } + } + return out +} + +// CallersEnv is the env value of a first-view tool's callers list, in the owner's tools bundle: MESH_VIEW_CALLERS_, +// the tool's name in capitals, a hyphen an underscore. +func CallersEnv(tool string) string { + return "MESH_VIEW_CALLERS_" + strings.ToUpper(strings.ReplaceAll(tool, "-", "_")) +} + +// Refusal is why caller may not ask tool, or nil: the caller's proof, never an argument, against the tool's callers +// list holding exactly its readers (ADR 0621 §2, ADR 0622 §1); the owner itself is on no list, so it is refused too. +func Refusal(tool string, callers stdio.Callers, caller string) error { + if callers.Allows(caller) { + return nil + } + return fmt.Errorf("%s is refused to %q: the tool answers only %s (novox/hq ADR 0621 §2)", tool, caller, strings.Join(callers, ", ")) +} + +// FirstViewTool is an owner's first-view tool of one of its states (ADR 0621 §1): each call answers one page of what the +// state holds now, at the revision its mark says, to the callers its list names and nobody else. It reads only the +// owner's own state, so it answers as soon as the runtime is connected, before the owner has asked anyone anything +// (ADR 0621 §3). +func FirstViewTool(owner, name, state, words string, callers stdio.Callers, values, marks Source, now func() time.Time) stdio.Tool { + return stdio.Tool{ + Name: name, + Description: fmt.Sprintf("The first view of %s's state %s (novox/hq ADR 0615, ADR 0621): %s Each value whole by key, at the revision "+ + "the owner said its views family up to, so a reader then follows the events on that family and asks again only on a gap in "+ + "revisions. One page at a time: next is the key to ask after for the rest. Answered only to %s; any other caller is refused.", + owner, state, words, strings.Join(callers, ", ")), + Input: map[string]any{"type": "object", "properties": map[string]any{ + "after": map[string]any{"type": "string", "description": "the key the page before ended at (its next); none starts at the first"}}}, + Run: func(args map[string]any) (any, error) { + if err := Refusal(name, callers, stdio.CallerOf(args)); err != nil { + return nil, err + } + after, _ := args["after"].(string) + return FirstView(owner, state, values, marks, strings.TrimSpace(after), now()) + }, + } +} + +// ReadTool is an owner's ordinary read tool of one of its states (design 25 § 12.2): the keys, one key's value and +// revision, or a page of values in the first view's form, and the state's fill against its cap as its mark last said +// it. Granted, as any tool is, to every reader that invokes it, and callable by a session. +func ReadTool(owner, name, state, words string, values, marks Source, now func() time.Time) stdio.Tool { + return stdio.Tool{ + Name: name, + Description: fmt.Sprintf("Read %s's state %s: %s With no argument, its keys and how full it is against its cap (as the "+ + "owner last marked it); with key, that key's value and revision; with values, a page of values whole by key, at the "+ + "revision the owner said its views family up to, continued with after (the page's next).", owner, state, words), + Input: map[string]any{"type": "object", "properties": map[string]any{ + "key": map[string]any{"type": "string", "description": "one key: its value and revision"}, + "values": map[string]any{"type": "boolean", "description": "a page of values whole by key"}, + "after": map[string]any{"type": "string", "description": "with values: the key the page before ended at"}}}, + Run: func(args map[string]any) (any, error) { + key, _ := args["key"].(string) + after, _ := args["after"].(string) + all, _ := args["values"].(bool) + return Read(owner, state, values, marks, strings.TrimSpace(key), all, strings.TrimSpace(after), now()) + }, + } +} + +// ReadAnswer is a read tool's answer without values: the state's keys and fill. +type ReadAnswer struct { + Owner string `json:"owner"` + State string `json:"state"` + Keys []string `json:"keys"` + Fill *Fill `json:"fill,omitempty"` + Revision uint64 `json:"revision"` + Marked time.Time `json:"marked,omitzero"` + // Words say what the fill means where the owner has not marked the state yet. + Words string `json:"words,omitempty"` +} + +// KeyAnswer is one key of a read tool's answer. +type KeyAnswer struct { + Owner string `json:"owner"` + State string `json:"state"` + Key string `json:"key"` + Found bool `json:"found"` + Value json.RawMessage `json:"value,omitempty"` + Revision uint64 `json:"revision,omitempty"` +} + +// Read is a read tool's answer: one key, a page of values, or the keys and fill. +func Read(owner, state string, values, marks Source, key string, all bool, after string, now time.Time) (any, error) { + switch { + case key != "": + e, ok, err := values.Get(key) + if err != nil { + return nil, fmt.Errorf("%s.%s cannot be read: %w", state, key, err) + } + a := KeyAnswer{Owner: owner, State: state, Key: key, Found: ok} + if ok { + a.Value, a.Revision = e.Value, e.Revision + } + return a, nil + case all: + return FirstView(owner, state, values, marks, after, now) + } + keys, err := values.Keys() + if err != nil { + return nil, fmt.Errorf("the state %s cannot be listed: %w", state, err) + } + slices.Sort(keys) + a := ReadAnswer{Owner: owner, State: state, Keys: keys} + if marks != nil { + if e, ok, err := marks.Get(state); err == nil && ok { + if m, err := DecodeMark(e.Value); err == nil { + fill := m.Fill + a.Fill, a.Revision, a.Marked = &fill, m.Revision, m.At + } + } + } + if a.Fill == nil { + a.Words = "the owner has not marked this state yet: its fill is said once its views family says a change of it" + } + if a.Keys == nil { + a.Keys = []string{} + } + return a, nil +} diff --git a/sdk/views/views.go b/sdk/views/views.go new file mode 100644 index 00000000..d4bd069e --- /dev/null +++ b/sdk/views/views.go @@ -0,0 +1,242 @@ +// Package views is the owners' side and the readers' side of a first view (novox/hq ADR 0615, ADR 0621, ADR 0622; +// design 25 § 12.2): the indexer and the assigner answer a reader's first view of one of their states as a tool of +// their own module (Answer, FirstView), say each change of that state as one event on their views family (Event, +// Emitter), and a reader keeps its picture from the answer and the events, asking the tool again only on a gap in +// revisions (Follower). The tools' read side (Read) answers the ordinary read tools of § 12.2 too. +package views + +import ( + "encoding/json" + "errors" + "fmt" + "slices" + "strings" + "time" +) + +// Version is the version of every value this package writes and the only one it reads. +const Version = 1 + +// The owners' views families (design 25 § 12.2; ADR 0615 §5, ADR 0619 §2): one stream each on its owner's old `streams` +// section, kept by limits 7 days and at most 256 MiB, dropping its oldest when full, so a reader that missed a dropped +// event sees the gap in revisions and asks once more. A stream's name is the mesh's, so each carries its owner's name. +const ( + IndexerStream = "indexer-views" + AssignerStream = "assigner-views" + StreamDays = 7 + StreamMiB = 256 + // EventPrefix starts each event's local name: `views.`, one event per changed key of that state. + EventPrefix = "views." + // EventPattern is every event of a views family, as a stream's events and a reader's consumes name it. + EventPattern = EventPrefix + "**" +) + +// MarksState is an owner's own state of the revision each followed state was said up to (Mark), one key per state: +// what a first view answers its revision from, so a reader that takes the answer misses no event after it. +const MarksState = "view-marks" + +// MaxValueBytes is the largest value an event carries whole: under the bus's largest message (1 MiB, the server's +// max_payload) with the event's own fields. A larger one is said as its revision alone (Event.Gap), and the owner +// writes a finding naming the key (design 25 § 12.2). +const MaxValueBytes = 960 * 1024 + +// PageBytes bounds the values one answer carries, so the answer fits one bus message; Answer.Next continues it. +const PageBytes = 768 * 1024 + +// Bound is how long a reader waits for its first view before it raises the condition naming the owner and the verb +// (ADR 0615 §6): longer than an owner's start, so a normal start, readers and owners coming up together, tells nobody. +const Bound = 2 * time.Minute + +// EventName is the local name of the event saying a change of state: `views.`. +func EventName(state string) string { return EventPrefix + state } + +// StateOfEvent is the state an event's local name (or the last tokens of its subject) says, or "". +func StateOfEvent(name string) string { + if i := strings.LastIndex(name, ".event."+EventPrefix); i >= 0 { + name = name[i+len(".event."):] + } + s, ok := strings.CutPrefix(name, EventPrefix) + if !ok || s == "" || strings.ContainsAny(s, ".*> ") { + return "" + } + return s +} + +// Event is one change of one key of an owner's state, said on its views family: the key's value whole, or a delete +// mark, at the revision of the state the change was written at. Created is when the owner's state was made on the bus: +// a state made again (the bus module's recreate) starts its revisions again, and a reader takes another Created as a gap. +type Event struct { + V int `json:"v"` + Owner string `json:"owner"` + State string `json:"state"` + Key string `json:"key"` + Revision uint64 `json:"revision"` + Created time.Time `json:"created"` + Deleted bool `json:"deleted,omitempty"` + Value json.RawMessage `json:"value,omitempty"` + // Gap says the value was too large to carry (MaxValueBytes): the reader asks its first view once. + Gap bool `json:"gap,omitempty"` +} + +// ID is the event's id on the bus, so the same change said twice (an owner's restart) is kept once. +func (e Event) ID() string { + return fmt.Sprintf("%s-%s-%d-%d", e.Owner, e.State, e.Created.UnixNano(), e.Revision) +} + +// Encode writes an event. +func (e Event) Encode() ([]byte, error) { + e.V = Version + if e.Owner == "" || e.State == "" || e.Key == "" || e.Revision == 0 { + return nil, errors.New("views: an event names its owner, state, key and revision") + } + if e.Deleted || e.Gap { + e.Value = nil + } + return json.Marshal(e) +} + +// DecodeEvent reads an event of this version. +func DecodeEvent(data []byte) (Event, error) { + var e Event + if err := decode(data, &e); err != nil { + return Event{}, err + } + if e.State == "" || e.Key == "" || e.Revision == 0 { + return Event{}, errors.New("views: an event names no state, key or revision") + } + return e, nil +} + +// Mark is the revision an owner said its state up to, and when that state was made (MarksState). +type Mark struct { + V int `json:"v"` + State string `json:"state"` + Revision uint64 `json:"revision"` + Created time.Time `json:"created"` + At time.Time `json:"at"` + // Fill is the state's bucket as the bus held it when the mark was written: what a read tool says of its fill. + Fill Fill `json:"fill"` +} + +// Fill is a state's bucket's fill against its cap, as the bus says it. +type Fill struct { + Keys uint64 `json:"keys"` + Bytes uint64 `json:"bytes"` + MaxBytes int64 `json:"max-bytes"` +} + +// DecodeMark reads a mark of this version. +func DecodeMark(data []byte) (Mark, error) { + var m Mark + return m, decode(data, &m) +} + +// Answer is a first view, or one page of an ordinary read: the state's values by key, from After on, at the revision +// of the state the owner said up to (Mark) before it read them; Next is the key to ask after for the rest, empty when +// the answer is whole. A reader keeps the first page's revision: a value on a later page may be newer, and the events +// after that revision bring every key to its newest in order. +type Answer struct { + V int `json:"v"` + Owner string `json:"owner"` + State string `json:"state"` + Revision uint64 `json:"revision"` + Created time.Time `json:"created"` + Values map[string]json.RawMessage `json:"values"` + Next string `json:"next,omitempty"` + At time.Time `json:"at"` +} + +// DecodeAnswer reads an answer of this version. +func DecodeAnswer(data []byte) (Answer, error) { + var a Answer + if err := decode(data, &a); err != nil { + return Answer{}, err + } + if a.Values == nil { + a.Values = map[string]json.RawMessage{} + } + return a, nil +} + +// Entry is one key's value and revision, as a state read gives it. +type Entry struct { + Value json.RawMessage + Revision uint64 +} + +// Source is the read side of one state, as a tools bundle reaches its own state (stdio.KeptState's Keys and Get). +type Source interface { + Keys() ([]string, error) + Get(key string) (*Entry, bool, error) +} + +// FirstView answers one page of state from its owner's source: the mark first, so no change after the revision it +// answers at is missed, then the values in key order after after, until PageBytes. A state not said yet (no mark) +// answers revision 0: a first view of whatever it holds, which the reader takes as a gap at the first event. +func FirstView(owner, state string, values, marks Source, after string, now time.Time) (Answer, error) { + a := Answer{V: Version, Owner: owner, State: state, Values: map[string]json.RawMessage{}, At: now.UTC()} + if marks != nil { + e, ok, err := marks.Get(state) + if err != nil { + return Answer{}, fmt.Errorf("the mark of %s cannot be read: %w", state, err) + } + if ok { + m, err := DecodeMark(e.Value) + if err != nil { + return Answer{}, fmt.Errorf("the mark of %s: %w", state, err) + } + a.Revision, a.Created = m.Revision, m.Created + } + } + keys, err := values.Keys() + if err != nil { + return Answer{}, fmt.Errorf("the state %s cannot be listed: %w", state, err) + } + slices.Sort(keys) + size := 0 + for _, k := range keys { + if after != "" && k <= after { + continue + } + e, ok, err := values.Get(k) + if err != nil { + return Answer{}, fmt.Errorf("%s.%s cannot be read: %w", state, k, err) + } + if !ok { + continue + } + if size > 0 && size+len(k)+len(e.Value) > PageBytes { + a.Next = lastKey(a.Values) + break + } + a.Values[k] = e.Value + size += len(k) + len(e.Value) + } + return a, nil +} + +func lastKey(m map[string]json.RawMessage) string { + last := "" + for k := range m { + if k > last { + last = k + } + } + return last +} + +func decode(data []byte, into any) error { + var probe struct { + V *int `json:"v"` + } + if err := json.Unmarshal(data, &probe); err != nil { + return fmt.Errorf("views: value does not fit its shape: %w", err) + } + if probe.V == nil || *probe.V != Version { + return fmt.Errorf("views: unknown value version: this reader reads %d", Version) + } + if err := json.Unmarshal(data, into); err != nil { + return fmt.Errorf("views: value does not fit its shape: %w", err) + } + return nil +} diff --git a/sdk/views/views_test.go b/sdk/views/views_test.go new file mode 100644 index 00000000..47894933 --- /dev/null +++ b/sdk/views/views_test.go @@ -0,0 +1,202 @@ +package views + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// mapSource is a state held in memory, as a Source. +type mapSource map[string]Entry + +func (m mapSource) Keys() ([]string, error) { + var out []string + for k := range m { + out = append(out, k) + } + return out, nil +} + +func (m mapSource) Get(key string) (*Entry, bool, error) { + e, ok := m[key] + if !ok { + return nil, false, nil + } + return &e, true, nil +} + +func markOf(state string, rev uint64, made time.Time) Entry { + raw, _ := json.Marshal(Mark{V: Version, State: state, Revision: rev, Created: made, Fill: Fill{Keys: 2, Bytes: 100, MaxBytes: 1000}}) + return Entry{Value: raw, Revision: 1} +} + +// The owners' and the readers' side of a first view (ADR 0615 §4-§6, ADR 0621 §2-§3, ADR 0622): one table of cases, +// since mesh-delivery is over its test budget. +func TestViews(t *testing.T) { + made := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) + now := func() time.Time { return made.Add(time.Hour) } + values := mapSource{"a": {Value: json.RawMessage(`{"n":1}`), Revision: 7}, "b": {Value: json.RawMessage(`{"n":2}`), Revision: 9}} + marks := mapSource{"modules": markOf("modules", 9, made)} + + t.Run("a first view answers every value at the mark's revision, the mark read first", func(t *testing.T) { + a, err := FirstView("o", "modules", values, marks, "", now()) + if err != nil || a.Revision != 9 || !a.Created.Equal(made) || len(a.Values) != 2 || a.Next != "" { + t.Fatalf("answer = %+v, %v", a, err) + } + if a, _ := FirstView("o", "modules", values, mapSource{}, "", now()); a.Revision != 0 { + t.Errorf("a state not marked answers revision %d, want 0", a.Revision) + } + }) + t.Run("a first view larger than a page is paged by key", func(t *testing.T) { + big := mapSource{} + for i := range 4 { + big[fmt.Sprintf("k%d", i)] = Entry{Value: json.RawMessage(`"` + strings.Repeat("x", PageBytes/3) + `"`), Revision: uint64(i + 1)} + } + got := map[string]bool{} + after := "" + for pages := 0; ; pages++ { + a, err := FirstView("o", "s", big, nil, after, now()) + if err != nil || pages > 4 { + t.Fatalf("page %d: %+v %v", pages, a.Next, err) + } + for k := range a.Values { + got[k] = true + } + if a.Next == "" { + break + } + after = a.Next + } + if len(got) != 4 { + t.Errorf("pages hold %v, want the four keys", got) + } + }) + t.Run("each first-view tool answers its callers alone, never the owner, by the proof and not an argument", func(t *testing.T) { + callers := ParseCallers("mesh-grants, node-tools,mesh-delivery-assigner") + for caller, allowed := range map[string]bool{"novox.mesh-grants": true, "ace.node-tools": true, "g14.mesh-delivery-assigner": true, + "novox.mesh-delivery-indexer": false, "novox.mesh-config": false, "person.jochen": false, stdio.Unproven: false} { + if err := Refusal("manifests", callers, caller); (err == nil) != allowed { + t.Errorf("%q: err = %v, allowed %v", caller, err, allowed) + } + } + // A caller argument names nobody: a call with no proof is refused. + tool := FirstViewTool("mesh-delivery-indexer", "manifests", "modules", "w.", callers, values, marks, now) + if _, err := tool.Run(map[string]any{"caller": "novox.mesh-grants"}); err == nil { + t.Errorf("an unproven call naming a caller is answered") + } + }) + t.Run("a read tool says the keys and the fill, one key, or a page of values", func(t *testing.T) { + r, err := Read("o", "modules", values, marks, "", false, "", now()) + ra, ok := r.(ReadAnswer) + if err != nil || !ok || len(ra.Keys) != 2 || ra.Fill == nil || ra.Fill.MaxBytes != 1000 || ra.Revision != 9 { + t.Fatalf("read = %+v, %v", r, err) + } + r, _ = Read("o", "modules", values, marks, "b", false, "", now()) + if k := r.(KeyAnswer); !k.Found || k.Revision != 9 { + t.Errorf("one key = %+v", k) + } + r, _ = Read("o", "modules", values, nil, "", false, "", now()) + if ra := r.(ReadAnswer); ra.Fill != nil || ra.Words == "" { + t.Errorf("an unmarked state says its fill as %+v", ra) + } + }) + + // A follower whose owner answers from values and marks. + follower := func(ask func(context.Context, string) (Answer, error)) (*Follower, *[]string) { + var said []string + f := &Follower{Owner: "o", Tool: "o.modules", State: "modules", Ask: ask, Bound: time.Nanosecond, Wait: time.Millisecond, + Raise: func(s Status) { said = append(said, "raised "+s.Tool) }, Clear: func(s Status) { said = append(said, "cleared") }} + return f, &said + } + answer := func(context.Context, string) (Answer, error) { return FirstView("o", "modules", values, marks, "", now()) } + ev := func(rev uint64, key, value string) Event { + e := Event{V: Version, Owner: "o", State: "modules", Key: key, Revision: rev, Created: made} + if value == "" { + e.Deleted = true + } else { + e.Value = json.RawMessage(value) + } + return e + } + t.Run("a reader applies a put and a delete from the events alone, and passes over what it holds", func(t *testing.T) { + asks := 0 + f, _ := follower(func(ctx context.Context, after string) (Answer, error) { asks++; return answer(ctx, after) }) + if err := f.Start(context.Background()); err != nil { + t.Fatal(err) + } + for _, e := range []Event{ev(8, "a", `{"n":0}`), ev(10, "c", `{"n":3}`), ev(11, "a", "")} { + if gap, why := f.Apply(e); gap { + t.Fatalf("revision %d is a gap: %s", e.Revision, why) + } + } + if _, ok := f.Get("a"); ok || f.Status().Revision != 11 || asks != 1 { + t.Errorf("picture = %v at %d after %d asks", f.Values(), f.Status().Revision, asks) + } + if v, _ := f.Get("c"); string(v) != `{"n":3}` { + t.Errorf("c = %s", v) + } + }) + t.Run("a gap, a state made again and a value too large each ask once more", func(t *testing.T) { + for name, e := range map[string]Event{ + "a revision missed": ev(12, "a", `{}`), + "made again": {V: Version, Owner: "o", State: "modules", Key: "a", Revision: 1, Created: made.Add(time.Minute), Value: json.RawMessage(`{}`)}, + "too large": {V: Version, Owner: "o", State: "modules", Key: "a", Revision: 10, Created: made, Gap: true}, + } { + f, _ := follower(answer) + _ = f.Start(context.Background()) + if gap, _ := f.Apply(e); !gap { + t.Errorf("%s: not taken as a gap", name) + } + } + }) + t.Run("a reader whose owner is down keeps asking, raises past its bound, and the answer clears it", func(t *testing.T) { + down := 3 + f, said := follower(func(ctx context.Context, after string) (Answer, error) { + if down > 0 { + down-- + return Answer{}, errors.New("no responders") + } + return answer(ctx, after) + }) + if err := f.Start(context.Background()); err != nil { + t.Fatal(err) + } + if strings.Join(*said, ",") != "raised o.modules,cleared" || f.Status().Asks != 4 || f.Status().Raised { + t.Errorf("said %v after %d asks", *said, f.Status().Asks) + } + }) + t.Run("an event heard while an ask runs is taken again after its answer", func(t *testing.T) { + var f *Follower + f, _ = follower(func(ctx context.Context, after string) (Answer, error) { + a, err := answer(ctx, after) // the picture read at 9, before revision 10 was written + if f.Status().Asks == 2 { + f.Apply(ev(10, "d", `{"n":4}`)) + } + return a, err + }) + _ = f.Start(context.Background()) + _ = f.Again(context.Background(), "test") + if v, ok := f.Get("d"); !ok || string(v) != `{"n":4}` || f.Status().Revision != 10 || f.Status().ByHand != 1 { + t.Errorf("d = %s, %v at %d", v, ok, f.Status().Revision) + } + }) + t.Run("an event names its state on its subject, and decodes as it was encoded", func(t *testing.T) { + if s := StateOfEvent("mesh.stream.mesh-delivery-indexer.event.views.config-candidates"); s != "config-candidates" { + t.Errorf("state of subject = %q", s) + } + raw, err := ev(3, "k", `{"x":1}`).Encode() + got, err2 := DecodeEvent(raw) + if err != nil || err2 != nil || got.Revision != 3 || string(got.Value) != `{"x":1}` { + t.Errorf("round trip = %+v %v %v", got, err, err2) + } + if _, err := (Event{Owner: "o", State: "s"}).Encode(); err == nil { + t.Errorf("an event with no key or revision is encoded") + } + }) +}