diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 32069148..9c5e1c07 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -5,6 +5,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "maps" @@ -30,6 +31,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/provisions" "mesh-delivery-assigner/internal/judge" + "git.novox.be/novox/mesh-delivery/sdk/views" + "mesh-delivery-assigner/internal/assign" ) @@ -99,6 +102,15 @@ func startBed(t *testing.T, leaveOut string) *bed { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CONDITIONS", Subjects: []string{"mesh.seat.operator-channel.accept.condition.>"}}); err != nil { t.Fatal(err) } + // The assigner's views family and its marks (ADR 0615 §5, ADR 0619 §2), on its old sections. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNER_VIEWS", Subjects: []string{"mesh.stream.mesh-delivery-assigner.event.views.>"}}); err != nil { + t.Fatal(err) + } + if kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: marksBucket, History: 1}); err != nil { + t.Fatal(err) + } else { + b.buckets[marksBucket] = kv + } // The Go build seat's work queue, as the broker derives it from the seat's accepted kinds (delivery.AskStream). seat, _ := delivery.AskStream(delivery.AskGo) if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: seat, Subjects: []string{"mesh.seat.node-build-go.accept.>"}, @@ -376,6 +388,9 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { if w := b.wantNow(t, "shanks", "web", "shanks' first want at what it runs", func(w delivery.Want) bool { return w.FP != "" }); w.FP != fOld || w.Composed != "" || !w.Self || w.Newest != fNew { t.Errorf("shanks' want = %+v, want the fingerprint it runs, the node composing it, held back from F", w) } + // DLV-O5 (ADR 0615 §5, ADR 0619 §2): each want is said on the assigner's views family, its value whole at its revision, + // and the first view at the mark followed by those events holds it as the bucket does; a waiter on put never wakes. + b.viewsHold(t, "wants", wkey) var neither asks.Condition for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && neither.Key == ""; time.Sleep(100 * time.Millisecond) { for k, c := range b.conditionsSaid(t) { @@ -1167,3 +1182,78 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) { t.Error("the removed node's key stays") } } + +// viewsHold waits until a follower of state, from a first view at its mark and the views family's events, holds key as +// the bucket does, and checks that no views event landed where a waiter on the assigner's put waits. +func (b *bed) viewsHold(t *testing.T, state, key string) { + t.Helper() + kv, marks := b.buckets[viewedStates[state]], b.buckets[marksBucket] + src := func(kv jetstream.KeyValue) views.Source { return bedSource{b.ctx, kv} } + st, err := b.js.Stream(b.ctx, "ASSIGNER_VIEWS") + if err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(20 * time.Second); ; time.Sleep(50 * time.Millisecond) { + f := &views.Follower{Owner: delivery.AssignerModule, Tool: state, State: state, Ask: func(context.Context, string) (views.Answer, error) { + return views.FirstView(delivery.AssignerModule, state, src(kv), src(marks), "", time.Now()) + }} + if err := f.Start(b.ctx); err != nil { + t.Fatal(err) + } + info, _ := st.Info(b.ctx) + gap := false + for seq := info.State.FirstSeq; seq > 0 && seq <= info.State.LastSeq; seq++ { + m, err := st.GetMsg(b.ctx, seq) + if err != nil || views.StateOfEvent(m.Subject) != state { + continue + } + e, err := views.DecodeEvent(m.Data) + if err != nil { + t.Fatal(err) + } + if g, _ := f.Apply(e); g { + gap = true + } + } + e, err := kv.Get(b.ctx, key) + v, ok := f.Get(key) + if err == nil && ok && !gap && string(v) == string(e.Value()) && f.Status().Revision >= e.Revision() { + break + } + if time.Now().After(deadline) { + t.Fatalf("the views family holds %s.%s as %s at %d (gap %v), the bucket %v", state, key, v, f.Status().Revision, gap, err) + } + } + ev, err := b.js.Stream(b.ctx, "EVENTS") + if err != nil { + t.Fatal(err) + } + if _, err := ev.GetLastMsgForSubject(b.ctx, "mesh.mod.mesh-delivery-assigner.event.put"); err == nil { + t.Errorf("a waiter on put woke: the bus holds a put") + } +} + +// bedSource is a bucket as a views.Source. +type bedSource struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (s bedSource) Keys() ([]string, error) { + k, err := s.kv.Keys(s.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + return k, err +} + +func (s bedSource) Get(key string) (*views.Entry, bool, error) { + e, err := s.kv.Get(s.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, false, nil + } + if err != nil { + return nil, false, err + } + return &views.Entry{Value: e.Value(), Revision: e.Revision()}, true, nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index ec894b3b..76c6a818 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -209,6 +209,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro thisNode = account.Node logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs)) findings := findingsOf(ctx, kvs, logger.Printf) + // Each change of the states its readers read is said on its views family (ADR 0615 §5, ADR 0619 §2); a watch of it + // that ends stops the program, as a watch of what a pass reads does. + viewErrs := make(chan error, len(viewedStates)) + startViews(ctx, js, conn, account, kvs, findings, viewErrs, logger.Printf) + go func() { + select { + case <-ctx.Done(): + case err := <-viewErrs: + logger.Printf("%v: the assigner stops, and is started again", err) + conn.Close() + } + }() return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, logger) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index 6adca56d..9c94e403 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -57,7 +57,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) { for _, pair := range []struct { what string code, want []string - }{{"reads", readBuckets, reads}, {"states", ownBuckets, states}} { + }{{"reads", readBuckets, reads}, {"states", append(slices.Clone(ownBuckets), marksBucket), states}} { // the marks are bound by the views family (views.go) got, want := slices.Sorted(slices.Values(pair.code)), slices.Sorted(slices.Values(pair.want)) if !reflect.DeepEqual(got, want) { t.Errorf("the program's %s are %v, the manifest's %v", pair.what, got, want) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index fd93f700..0506e404 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -127,7 +127,8 @@ func TestTheSeatsVerbs(t *testing.T) { } } for _, stream := range raw0.Streams { - if !reads["stream "+stream.Name] { + // The views family is no seat's stream: the readers of the assigner's states follow it (ADR 0615 §5). + if stream.Name != viewsStream && !reads["stream "+stream.Name] { t.Errorf("no verb reads the seat's stream %s", stream.Name) } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go index 2ed4d101..30f55902 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/tools.go @@ -358,6 +358,7 @@ func serveTools() error { }) wants.watched = stdio.NewWatchedStates() // the large states read whole from one watched copy (ADR 0580 clause 3) tools = append(tools, wants.tools()...) + tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...) return stdio.Serve("", tools) } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go new file mode 100644 index 00000000..98c1cd1c --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -0,0 +1,103 @@ +package main + +import ( + "context" + "time" + + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" + "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/views" + + "mesh-delivery-assigner/internal/assign" +) + +// viewsStream is the assigner's views family on its old streams section, beside placement (ADR 0619 §2, ADR 0615 §5). +const viewsStream = views.AssignerStream + +// marksBucket is the assigner's state of what its views family said each state up to. +var marksBucket = delivery.AssignerModule + "_" + views.MarksState + +// viewedTool is one tool over one state of the assigner's. +type viewedTool struct{ tool, state, words string } + +// firstViewTools are the assigner's first-view tools (ADR 0621 §1): each answers only its callers list +// (views.CallersEnv), which the manifest writes. +var firstViewTools = []viewedTool{ + {"placement", delivery.PlacementState, "where each module is placed: the seat mesh-placement's placement, by module."}, + {"seat-holders", delivery.SeatHoldersState, "who holds which seat on which node."}, + {"wants", "wants", "what each node should run: want.., its fingerprint and the rule that gave it."}, +} + +// readTools are the assigner's ordinary read tools, one per state its readers read (design 25 § 12.2). +var readTools = []viewedTool{ + {"assigner_read_placement", delivery.PlacementState, "where each module is placed, by module."}, + {"assigner_read_seat_holders", delivery.SeatHoldersState, "who holds which seat on which node."}, + {"assigner_read_wants", "wants", "what each node should run: want..."}, + {"assigner_read_findings", health.FindingsState, "the assigner's findings, which the condition keeper raises and clears."}, + {"assigner_read_asks", "asks", "the asks the assigner put to a person, and their answers."}, + {"assigner_read_differences", "differences", "every difference between a want and what a node runs."}, +} + +// viewedStates are the states the views family says, each by its bucket: the six its readers read. +var viewedStates = map[string]string{ + delivery.PlacementState: delivery.PlacementBucket, + delivery.SeatHoldersState: delivery.SeatHoldersBucket, + "wants": delivery.AssignerWantsBucket, + health.FindingsState: assign.FindingsBucket, + "asks": delivery.AssignerAsksBucket, + "differences": delivery.AssignerDifferencesBucket, +} + +// viewTools are the assigner's first-view and read tools, reading only its own states, so they answer as soon as its +// 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 firstViewTools { + out = append(out, views.FirstViewTool(delivery.AssignerModule, t.tool, t.state, t.words, views.ParseCallers(getenv(views.CallersEnv(t.tool))), state(t.state), marks, now)) + } + for _, t := range readTools { + out = append(out, views.ReadTool(delivery.AssignerModule, t.tool, t.state, t.words, state(t.state), marks, now)) + } + return out +} + +func keptSource(name string) views.Source { return views.KeptSource{State: stdio.State(name)} } + +// startViews says each change of the assigner's viewed states on its views family, once its marks state is bound, +// each state as soon as its bucket is (kvs, or bound once it appears). A watch that ends is sent on errs. +func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, account process.Account, kvs buckets, findings assign.Findings, errs chan<- error, logf func(string, ...any)) { + tooLarge := func(state, key string, size int) { + if findings.Put == nil { + return + } + k, v, err := views.TooLargeFinding(delivery.AssignerModule, state, key, size, time.Now()) + if err == nil { + err = findings.Put(k, v) + } + if err != nil { + logf("views: the finding of %s.%s too large is not written: %v", state, key, err) + } + } + process.AwaitBucket(ctx, js, nc, marksBucket, "no change is said on the views family, so every reader waits on its first view", logf).Ready(func(mk jetstream.KeyValue) { + em := views.Emitter{Owner: delivery.AssignerModule, Subject: account.StreamSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf} + for _, state := range []string{"asks", "differences", health.FindingsState, delivery.PlacementState, delivery.SeatHoldersState, "wants"} { + state := state + start := 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 assigner starts again", err) + } + } + if kv := kvs[viewedStates[state]]; kv != nil { + start(kv) + continue + } + process.AwaitBucket(ctx, js, nc, viewedStates[state], "its changes are not said on the views family", logf).Ready(start) + } + }) +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 087b34cd..9fa690d0 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -8,6 +8,8 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" + + "git.novox.be/novox/mesh-delivery/sdk/views" ) func load(t *testing.T) catalogue.Manifest { @@ -47,7 +49,7 @@ func TestManifest(t *testing.T) { states[s.Name] = s.History } if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20, - "placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5}; !reflect.DeepEqual(states, want) { + "placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -78,10 +80,29 @@ func TestManifest(t *testing.T) { if seat, _ := catalogue.SeatNamed(catalogue.PlacementSeat); catalogue.CanHold(m, seat) != nil { t.Errorf("the assigner cannot hold the seat: %v", catalogue.CanHold(m, seat)) } - if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts"}) { - t.Errorf("the assigner serves tools %v, want its three put tools", m.Tools) + // ADR 0621 §1: the first-view tools placement, seat-holders and wants and design 25 § 12.2's six read tools, of the + // program's bundle, which runs in a runtime of its own on the assigner's account. + if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "placement", "seat-holders", "wants", + "assigner_read_placement", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences"}) { + t.Errorf("the assigner serves tools %v, want its three put tools, three first views and six read tools", m.Tools) } - if len(m.Streams) != 1 || m.Streams[0].Name != delivery.PlacementStream || m.Streams[0].MaxAgeDays != delivery.PlacementStreamDays || + if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs { + t.Errorf("runs-as = %q: the bundle runs on the assigner's own account (ADR 0621 §1)", m.RunsAs) + } + // ADR 0621 §2: each first-view tool answers exactly its readers, never the assigner. + callers := map[string]string{"placement": "mesh-grants,node-tools,mesh-delivery-indexer", "seat-holders": "node-tools", "wants": "mesh-grants"} + env := m.Build.Artifacts[0].Env + for tool, list := range callers { + if env[views.CallersEnv(tool)] != list || views.ParseCallers(env[views.CallersEnv(tool)]).Allows("n1."+m.Module) { + t.Errorf("%s callers = %q, want %q", tool, env[views.CallersEnv(tool)], list) + } + } + // ADR 0619 §2, ADR 0615 §5: its views family beside placement on its old streams section. + if len(m.Streams) != 2 || m.Streams[1].Name != views.AssignerStream || !slices.Equal(m.Streams[1].Events, []string{views.EventPattern}) || + m.Streams[1].MaxAgeDays != views.StreamDays || m.Streams[1].MaxMiB != views.StreamMiB || m.Streams[1].Retention != "" { + t.Errorf("streams = %+v, want placement and %s", m.Streams, views.AssignerStream) + } + if m.Streams[0].Name != delivery.PlacementStream || m.Streams[0].MaxAgeDays != delivery.PlacementStreamDays || !slices.Equal(slices.Sorted(slices.Values(m.Streams[0].Events)), slices.Sorted(slices.Values(delivery.PlacementEvents))) { t.Errorf("streams = %+v, want placement carrying the seat's four events for %d days", m.Streams, delivery.PlacementStreamDays) } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 7f94deaa..5678029f 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -2,6 +2,7 @@ "module": "mesh-delivery-assigner", "version": "1", "slug": "assigner", + "runs-as": "mesh-delivery-assigner", "claims": [ { "name": "mesh-placement", @@ -53,7 +54,16 @@ "tools": [ "assigner_put", "assigner_release", - "assigner_puts" + "assigner_puts", + "placement", + "seat-holders", + "wants", + "assigner_read_placement", + "assigner_read_seat_holders", + "assigner_read_wants", + "assigner_read_findings", + "assigner_read_asks", + "assigner_read_differences" ], "state": [ { "name": "placement", "history": 64 }, @@ -66,7 +76,8 @@ { "name": "wants", "history": 20 }, { "name": "findings", "history": 5 }, { "name": "policies", "history": 20 }, - { "name": "seat-holders", "history": 5 } + { "name": "seat-holders", "history": 5 }, + { "name": "view-marks", "history": 1 } ], "streams": [ { @@ -78,7 +89,8 @@ "placement-withdrawn" ], "max-age-days": 30 - } + }, + { "name": "assigner-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 } ], "own-secrets": { "broker": "${dir:state}/broker" @@ -128,7 +140,12 @@ "binary": "mesh-delivery-assigner", "loads": [ "mesh-delivery-assigner" - ] + ], + "env": { + "MESH_VIEW_CALLERS_PLACEMENT": "mesh-grants,node-tools,mesh-delivery-indexer", + "MESH_VIEW_CALLERS_SEAT_HOLDERS": "node-tools", + "MESH_VIEW_CALLERS_WANTS": "mesh-grants" + } } ] }