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 51ec80bb..917f4ebb 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 @@ -102,6 +102,8 @@ 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 indexer as the owner of its manifests and modules (DLV-F2r), with the real owner's side. + serveIndexer(t, b, nc) // 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) @@ -1266,3 +1268,49 @@ func placementSubjects() []string { } return out } + +// serveIndexer is the indexer as the owner of its manifests and modules (ADR 0615, ADR 0621): its buckets, which a test +// fills as the indexer writes them; its views family said by the views emitter onto INDEXER_VIEWS, read by the +// assigner's durable consumer; and its first-view tools manifests and modules answered from the buckets at their marks. +func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) { + t.Helper() + marks, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.IndexerModule + "_" + views.MarksState, History: 1}) + if err != nil { + t.Fatal(err) + } + if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "INDEXER_VIEWS", Subjects: []string{"mesh.stream.mesh-delivery-indexer.event.views.>"}}); err != nil { + t.Fatal(err) + } + if _, err := b.js.CreateOrUpdateConsumer(b.ctx, "INDEXER_VIEWS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + em := views.Emitter{Owner: delivery.IndexerModule, Subject: func(e string) string { return process.ModuleStreamSubject(delivery.IndexerModule, e) }, + Publish: views.JetStreamPublish(b.js), Marks: marks} + for state, bucket := range map[string]string{"manifests": delivery.ManifestsBucket, "modules": delivery.ModulesBucket} { + kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 5}) + if err != nil { + t.Fatal(err) + } + b.buckets[bucket] = kv + if err := em.Start(b.ctx, state, kv, nil); err != nil { + t.Fatal(err) + } + sub, err := nc.Subscribe("mesh.mod.mesh-delivery-indexer.call."+state+".>", func(m *nats.Msg) { + var args struct { + After string `json:"after"` + } + _ = json.Unmarshal(m.Data, &args) + a, err := views.FirstView(delivery.IndexerModule, state, bedSource{b.ctx, kv}, bedSource{b.ctx, marks}, args.After, time.Now()) + out := map[string]any{"result": a, "node": "n0"} + if err != nil { + out = map[string]any{"error": err.Error()} + } + raw, _ := json.Marshal(out) + _ = m.Respond(raw) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = sub.Unsubscribe() }) + } +} 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 6cac6878..2c5d32b1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -58,8 +58,6 @@ var readBuckets = []string{ delivery.PartGrantsBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, - delivery.ManifestsBucket, - delivery.ModulesBucket, delivery.PresenceBucket, delivery.HaveBucket, provisions.ProvisionsBucket, @@ -101,6 +99,7 @@ var keptStates = []announce.State{ {Name: delivery.AssignerPoliciesState, Readable: true}, {Name: delivery.SeatHoldersState, Readable: true}, {Name: views.MarksState, Readable: true}, // what its views family said each state up to (views.go) + {Name: views.StatusState, Readable: true}, // its first views of the indexer's states (views.go) } // buckets is every bucket the assigner opened, by name. @@ -223,7 +222,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro conn.Close() } }() - return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, logger) + iv := &indexerViews{js: js, account: account, call: said.Call, findings: findings, logf: logger.Printf, + status: process.AwaitBucket(ctx, js, conn, firstViewsBucket, "assigner_first_views reads nothing and assigner_first_view_again is not heard", logger.Printf)} + return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, iv, logger) } // findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, which the @@ -258,7 +259,6 @@ var watched = []string{ delivery.PartGrantsBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, - delivery.ModulesBucket, delivery.PlacementBucket, delivery.PlacementProposalsBucket, delivery.PlacementCandidatesBucket, @@ -278,7 +278,7 @@ var watched = []string{ // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after // each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends. -func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, logger *log.Logger) error { +func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, iv *indexerViews, logger *log.Logger) error { w := newWorld() dirty := make(chan struct{}, 1) wake := func() { @@ -377,8 +377,10 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, return err } } - if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil { - return err + // The indexer's manifests and modules, through its first views and views family, never its buckets (DLV-F2r): the + // first pass waits for both answers, as it waits for every watch to hand what stands. + if iv != nil { + started += iv.start(ctx, w, wake, settled) } // The seat's states arrive with the same delivery as this assigner (ADR 0567 §1): one not on the bus at this start is // bound once the bus says it was created, and watched from then on, with no restart (issue 1790's kind). Until then no @@ -410,7 +412,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, logger.Printf("the bucket %s is open and watched", a.name) return watch(a.name) } - for range started + 1 { + for range started { select { case <-ctx.Done(): return nil 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 89aee0ce..b411f949 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", append(slices.Clone(ownBuckets), marksBucket), states}} { // the marks are bound by the views family (views.go) + }{{"reads", readBuckets, reads}, {"states", append(slices.Clone(ownBuckets), marksBucket, firstViewsBucket), 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) @@ -67,7 +67,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) { for _, s := range keptStates { kept = append(kept, m.Module+"_"+s.Name) } - if got, want := slices.Sorted(slices.Values(kept)), slices.Sorted(slices.Values(append(slices.Clone(ownBuckets), marksBucket))); !reflect.DeepEqual(got, want) { + if got, want := slices.Sorted(slices.Values(kept)), slices.Sorted(slices.Values(append(slices.Clone(ownBuckets), marksBucket, firstViewsBucket))); !reflect.DeepEqual(got, want) { t.Errorf("the program announces %v, keeps %v", got, want) } // ADR 0546: whatever keeps the puts' bucket from opening (not there, not granted, no answer in time), the assigner opens @@ -113,7 +113,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) { } kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote - go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, log.New(io.Discard, "", 0)) + go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, nil, log.New(io.Discard, "", 0)) select { case key := <-wrote: if key != "want.ace.node" { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index cc520b40..f8480563 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -107,7 +107,7 @@ func (p *passer) nodeWants(ctx context.Context) { // the hold of a fingerprint whose health check failed. The assigner compares its wants with nothing the controller // sends: the differences bucket is kept read-only with its history (ADR 0539 §8). func (p *passer) wants(ctx context.Context) { - get := kvGetter(p.kvs[delivery.ManifestsBucket]) + get := p.w.getter(delivery.ManifestsBucket) // the indexer's manifests, as its first view and views family say them assigned := p.w.assigned() trunk := p.w.trunk() manifests := map[string]catalogue.Manifest{} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go index fbf20327..e493328c 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/puts.go @@ -99,7 +99,7 @@ func (p *passer) puts(ctx context.Context) { } assigned := p.w.assigned() trunk := p.w.trunk() - get := kvGetter(p.kvs[delivery.ManifestsBucket]) + get := p.w.getter(delivery.ManifestsBucket) // the indexer's manifests, as its first view and views family say them wanted := p.w.wantsOf() modules := make([]string, 0, len(all)) for m := range all { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 98c1cd1c..96a81df1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go @@ -2,11 +2,13 @@ package main import ( "context" + "encoding/json" "time" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/health" "git.novox.be/novox/mesh-sdk/go/process" @@ -64,7 +66,8 @@ func viewTools(getenv func(string) string, state func(string) views.Source, 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 + // Its own first views, of the indexer's manifests and modules (DLV-F3r): read, and asked again by hand. + return append(out, views.StatusTools(delivery.AssignerModule, "assigner_first_views", "assigner_first_view_again", stdio.State(views.StatusState), now)...) } func keptSource(name string) views.Source { return views.KeptSource{State: stdio.State(name)} } @@ -101,3 +104,77 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib } }) } + +// firstViewsBucket is the assigner's own state of its first views (views.StatusState). +var firstViewsBucket = delivery.AssignerModule + "_" + views.StatusState + +// indexerViews are the assigner's first views of the indexer's manifests and modules (ADR 0615 §1, ADR 0621 §2; +// DLV-F2r): asked at the start through the indexer's tools, then followed on indexer-views, asking once more on a gap; +// past the bound its finding names the indexer and the tool, cleared by the answer. +type indexerViews struct { + js jetstream.JetStream + account process.Account + call func(string, any) (json.RawMessage, error) + findings assign.Findings + status *process.AwaitedBucket + logf func(string, ...any) +} + +// start asks both first views, each answer taken into w under the indexer's bucket names and settled once (a pass waits +// for it), and hears the views family; it answers how many settles to wait for. +func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settled chan<- struct{}) int { + finding := func(s views.Status, raised bool) { + if iv.findings.Put == nil { + return + } + key, value, err := views.UnansweredFinding(delivery.AssignerModule, s, time.Now()) + if err == nil && raised { + err = iv.findings.Put(key, value) + } else if err == nil { + err = iv.findings.Delete(key) + } + if err != nil { + iv.logf("views: the finding of the first view %s is not written: %v", s.Tool, err) + } + } + followers := map[string]*views.Follower{} + var all []*views.Follower + for state, bucket := range map[string]string{"manifests": delivery.ManifestsBucket, "modules": delivery.ModulesBucket} { + bucket := bucket + tool := delivery.IndexerModule + "." + state + f := &views.Follower{Owner: delivery.IndexerModule, Tool: tool, State: state, Ask: views.AskVia(iv.call, tool), Logf: iv.logf, + Raise: func(s views.Status) { finding(s, true) }, Clear: func(s views.Status) { finding(s, false) }, + Changed: func(key string, value json.RawMessage, deleted bool, at time.Time) { + if deleted { + value = nil + } + w.putEntry(bucket, key, value, 0, at) + wake() + }} + followers[state] = f + all = append(all, f) + go func() { + if f.Start(ctx) == nil { + settled <- struct{}{} + } + }() + } + if iv.status != nil { + iv.status.Ready(func(kv jetstream.KeyValue) { + for _, f := range all { + f.Said = views.KeepStatus(ctx, kv, iv.logf) + } + if err := views.HearAgain(ctx, kv, all, iv.logf); err != nil { + iv.logf("views: assigner_first_view_again is not heard (%v) until the assigner starts again", err) + } + }) + } + stream := catalogue.StreamBusName(views.IndexerStream) + if cons, err := iv.account.EventConsumer(ctx, iv.js, stream); err != nil { + iv.logf("the consumer %s on %s does not exist yet (%v): the indexer's manifests and modules are the first views' alone until the assigner starts again", + iv.account.EventConsumerName(), stream, err) + } else if _, err := process.HearEvents(cons, views.Hear(ctx, followers, iv.logf), iv.logf); err != nil { + iv.logf("the indexer's views family is not heard (%v) until the assigner starts again", err) + } + return len(all) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index f37724e7..ac28931e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -197,6 +197,19 @@ func kvGetter(kv jetstream.KeyValue) getter { } } +// getter reads a key of a bucket the world holds: errNotThere where it holds none. +func (w *world) getter(bucket string) getter { + return func(_ context.Context, key string) ([]byte, error) { + w.mu.Lock() + defer w.mu.Unlock() + v, ok := w.values[bucket][key] + if !ok { + return nil, errNotThere + } + return v, nil + } +} + // manifest reads the manifest of a digest from the indexer's bucket once and keeps it: a digest's manifest never changes. func (w *world) manifest(ctx context.Context, get getter, digest string) parsed { w.mu.Lock() diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index 9fa690d0..0f1c8dc7 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -38,7 +38,8 @@ func TestManifest(t *testing.T) { // ADR 0593 §8: it composes nothing, so it reads no candidate, held or resolved value, compose step or build log; it reads // every node's published compositions to compare them, and the gate's source and merges, for the release that carries a build (ADR 0601 §1); // and the indexer's module graph, whose verb edges a want waits on (ADR 0614 §2, §3). - want := []string{"mesh-delivery-builder.builds", "mesh-delivery-gate.merges", "mesh-delivery-gate.source", "mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + // DLV-F2r (ADR 0615 §1): the indexer's manifests and modules are read through its first views and views family. + want := []string{"mesh-delivery-builder.builds", "mesh-delivery-gate.merges", "mesh-delivery-gate.source", "mesh-delivery-indexer.graph", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed", "mesh-provisioner.provisions", "node-engine.have"} if !reflect.DeepEqual(reads, want) { @@ -49,7 +50,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, views.MarksState: 1}; !reflect.DeepEqual(states, want) { + "placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1}; !reflect.DeepEqual(states, want) { t.Errorf("state = %v, want %v", states, want) } if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 { @@ -83,7 +84,8 @@ func TestManifest(t *testing.T) { // 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"}) { + "assigner_read_placement", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", + "assigner_first_views", "assigner_first_view_again"}) { t.Errorf("the assigner serves tools %v, want its three put tools, three first views and six read tools", m.Tools) } if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs { @@ -106,7 +108,10 @@ func TestManifest(t *testing.T) { !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) } - if want := []string{ + if !slices.Equal(m.Consumes, []string{"mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules"}) { + t.Errorf("consumes = %v, want the indexer's views of manifests and modules", m.Consumes) + } + if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone "seat:node-backup.backed-up", "seat:node-backup.now"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { t.Errorf("invokes = %v, want %v", m.Invokes, want) diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 5678029f..496111dc 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -29,8 +29,6 @@ "mesh-delivery-gate.merges", "mesh-delivery-gate.source", "mesh-delivery-indexer.graph", - "mesh-delivery-indexer.manifests", - "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", @@ -41,7 +39,13 @@ "uses": [ "operator-channel" ], + "consumes": [ + "mesh-delivery-indexer.views.manifests", + "mesh-delivery-indexer.views.modules" + ], "invokes": [ + "mesh-delivery-indexer.manifests", + "mesh-delivery-indexer.modules", "seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", @@ -63,7 +67,9 @@ "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", - "assigner_read_differences" + "assigner_read_differences", + "assigner_first_views", + "assigner_first_view_again" ], "state": [ { "name": "placement", "history": 64 }, @@ -77,7 +83,8 @@ { "name": "findings", "history": 5 }, { "name": "policies", "history": 20 }, { "name": "seat-holders", "history": 5 }, - { "name": "view-marks", "history": 1 } + { "name": "view-marks", "history": 1 }, + { "name": "first-views", "history": 1 } ], "streams": [ {