diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go index 56516c70..650113cb 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go @@ -17,6 +17,7 @@ import ( "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/views" "mesh-delivery-builder/internal/build" ) @@ -264,7 +265,11 @@ func main() { a, err := build.DecodeReAsk(v) return err == nil && !a.Answered.IsZero() }) - if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now)); err != nil { + // builder_first_views and builder_first_view_again: where the builder's first views of the indexer's and the assigner's + // states stand, and asking one again by hand (ADR 0615, DLV-M1). + tools := append(Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now), + views.StatusTools(delivery.BuilderModule, "builder_first_views", "builder_first_view_again", stdio.State(views.StatusState), time.Now)...) + if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err) os.Exit(1) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go index 96c9d54d..2b9e6c38 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/integration_test.go @@ -27,6 +27,7 @@ import ( runner "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep" "git.novox.be/novox/mesh-sdk/go/buildstep" "git.novox.be/novox/mesh-sdk/go/buscompose" + "git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" @@ -270,6 +271,13 @@ func startBedWith(t *testing.T, closed bool) *bed { b.candidates = mk(delivery.CandidatesBucket, delivery.CandidatesHistory) b.source = mk(delivery.SourceBucket, delivery.SourceHistory) mk(delivery.ManifestsBucket, delivery.ManifestsHistory) + // The indexer and the assigner as the owners of the states the builder follows: their read tools and views families + // (ADR 0615, DLV-M1); n1 and n2 each have their durable consumer. + readers := []string{"n1_" + delivery.BuilderModule, "n2_" + delivery.BuilderModule} + for _, st := range []string{modulesState, manifestsState, candidatesState} { + firstviewtest.Serve(t, ctx, js, ctl, delivery.IndexerModule, st, readers...) + } + firstviewtest.Serve(t, ctx, js, ctl, delivery.AssignerModule, delivery.PlacementState, readers...) b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory) b.instances = mk(instancesBucket, 1) // The builder's module log, as the runtime raises it (ADR 0297): its stream on the log's subjects. @@ -791,7 +799,12 @@ func TestTheBuilderAsksTheBuildSeatsAndCallsNoNode(t *testing.T) { } defer sub.Close() var calls atomic.Int64 - s, err := sub.Subscribe("mesh.mod.*.call.>", func(*nats.Msg) { calls.Add(1) }) + // A build module's tool, never the owners' read tools the builder's first views ask (DLV-M1). + s, err := sub.Subscribe("mesh.mod.*.call.>", func(m *nats.Msg) { + if strings.HasPrefix(m.Subject, "mesh.mod.mesh-build-") { + calls.Add(1) + } + }) if err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index bdfdb4a3..f0d2308f 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -25,6 +25,9 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" + + "git.novox.be/novox/mesh-delivery/sdk/firstview" "mesh-delivery-builder/internal/build" ) @@ -61,16 +64,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - modules, err := process.EssentialBucket(ctx, js, "builder", delivery.ModulesBucket, "know any fingerprint to build") + // The indexer's modules, manifests and candidates and the assigner's placement are their owners' states, read by the + // first view of their read tools and then their views families, never their buckets (ADR 0615 §1, DLV-M1). An owner that + // is down never holds the start: each is read on a goroutine once its first view is answered, and until then the + // unanswered first view is the builder's finding. + calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) if err != nil { return err } - manifests, err := process.EssentialBucket(ctx, js, "builder", delivery.ManifestsBucket, "read what a module builds") - if err != nil { - return err - } - // The candidates' fingerprints are built too (issue 1684): a builder that cannot read them builds the trunk's. - candidates := process.OptionalBucket(ctx, js, delivery.CandidatesBucket, "no candidate's fingerprint is built until the builder starts again; the trunk's are", logger.Printf) + defer calls.Close() + reader := &firstview.Reader{Module: delivery.BuilderModule, Call: calls.Call, Logf: logger.Printf} + findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf) + reader.Findings = findings.KV + modules := reader.Follow(delivery.IndexerModule, modulesState) + manifests := reader.Follow(delivery.IndexerModule, manifestsState) + // The candidates' fingerprints are built too (issue 1684). + candidates := reader.Follow(delivery.IndexerModule, candidatesState) + placement := reader.Follow(delivery.AssignerModule, delivery.PlacementState) bound := defaultStepBound if v := strings.TrimSpace(getenv("MESH_BUILDER_STEP_BOUND")); v != "" { if bound, err = time.ParseDuration(v); err != nil { @@ -103,12 +113,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro heard := buildrun.HearEnds(ctx, asks, asks, buildModules, logger.Printf) conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again(); heard.Nudge("") }) // The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs - // (the seat verb held, novox/hq issue 1937) before it leaves the key done. - calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) - if err != nil { - return err - } - defer calls.Close() + // (the seat verb held, novox/hq issue 1937) before it leaves the key done, on its calls connection. b := &build.Builder{ Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module}, Asks: buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound}, @@ -165,10 +170,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // placement is settled first, so the toolchains settle through the gate already knowing it. gate := buildrun.NewToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf) watch.put = gate.Toolchain - placementErrs, err := buildrun.FollowPlacement(ctx, js, gate, b.NodesChanged, logger.Printf) - if err != nil { - return err - } + // Until the placement's first view is answered the gate passes every report on, as before the placement is settled. + placementErrs := make(chan error, 1) + go func() { + select { + case <-ctx.Done(): + return + case <-placement.Follower().Ready(): + } + pe, err := buildrun.FollowPlacement(ctx, pictures{placement}, gate, b.NodesChanged, logger.Printf) + if err != nil { + placementErrs <- err + return + } + placementErrs <- <-pe + }() if err := watch.start(); err != nil { return err } @@ -202,6 +218,78 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer cc.Stop() } + reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "builder_first_views reads nothing and builder_first_view_again is not heard", logger.Printf)) + // The modules, once their first views and the manifests' are answered: what a module builds is read from the manifests. + go func() { errs <- buildFromModules(ctx, b, modules, manifests, candidates) }() + // A person asks a key's build again with builder_ask_again, which writes the reasks state (issue 1945): watched, as + // every other input. The builder asks again by itself without it. + // A state not on the bus yet at this start is watched once it appears (issue 1790). + process.AwaitBucket(ctx, js, conn, build.ReAsksBucket, "builder_ask_again is not heard; an ask that can no longer be answered is still asked again by itself", logger.Printf).Ready(func(kv jetstream.KeyValue) { + eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) } + if rw, err := settle(ctx, kv, eachReAsk); err != nil { + process.NotWatched(logger.Printf, build.ReAsksBucket, err, "builder_ask_again is not heard until the builder starts again") + } else { + go func() { errs <- follow(ctx, rw, eachReAsk) }() + } + }) + // A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the + // builder alone may read the stream, so it answers there. + process.AwaitBucket(ctx, js, conn, build.QueuesBucket, "builder_seat_queue is not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) { + eachRead := func(e jetstream.KeyValueEntry) { queueRead(ctx, heldOnTheBus{js}, kv, e, logger.Printf) } + if qw, err := settle(ctx, kv, eachRead); err != nil { + process.NotWatched(logger.Printf, build.QueuesBucket, err, "builder_seat_queue is not answered until the builder starts again") + } else { + go func() { errs <- follow(ctx, qw, eachRead) }() + } + }) + logger.Printf("started on %s", cred.Node) + select { + case <-ctx.Done(): + b.Wait() + return nil + case err := <-errs: + return err + case err := <-placementErrs: + return err + case err := <-watch.errs: + return err + } +} + +// The builder's own states of its first views (views.StatusState) and of its findings, where the condition keeper raises a +// first view unanswered past its bound (ADR 0615 §6). +var ( + firstViewsBucket = delivery.BuilderModule + "_" + views.StatusState + findingsBucket = delivery.BuilderModule + "_findings" +) + +// The indexer's states the builder follows (delivery.ModulesBucket, ManifestsBucket, CandidatesBucket). +const ( + modulesState = "modules" + manifestsState = "manifests" + candidatesState = "candidates" +) + +// pictures opens the placement's picture where the gate's follower opens its bucket. +type pictures struct{ placement *firstview.Picture } + +func (p pictures) KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) { + if bucket != p.placement.Bucket() { + return nil, fmt.Errorf("the builder reads %s by its owner's first view alone", bucket) + } + return p.placement, nil +} + +// buildFromModules builds from the indexer's modules and candidates once the first views of the modules, the manifests and +// the candidates are answered (never at the start: an indexer that is down holds no other part of it), and follows them. +func buildFromModules(ctx context.Context, b *build.Builder, modules, manifests, candidates *firstview.Picture) error { + for _, p := range []*firstview.Picture{modules, manifests, candidates} { + select { + case <-ctx.Done(): + return nil + case <-p.Follower().Ready(): + } + } each := func(e jetstream.KeyValueEntry) { if opOf(e) { b.Module(ctx, e.Key(), e.Value()) @@ -225,48 +313,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro b.ForgetCandidate(e.Key()) } } - if candidates != nil { - cw, err := settle(ctx, candidates, eachCandidate) - if err != nil { - process.NotWatched(logger.Printf, delivery.CandidatesBucket, err, "no candidate's fingerprint is built until the builder starts again") - } else { - go func() { errs <- follow(ctx, cw, eachCandidate) }() - } - } - // A person asks a key's build again with builder_ask_again, which writes the reasks state (issue 1945): watched, as - // every other input. The builder asks again by itself without it. - // A state not on the bus yet at this start is watched once it appears (issue 1790). - process.AwaitBucket(ctx, js, conn, build.ReAsksBucket, "builder_ask_again is not heard; an ask that can no longer be answered is still asked again by itself", logger.Printf).Ready(func(kv jetstream.KeyValue) { - eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) } - if rw, err := settle(ctx, kv, eachReAsk); err != nil { - process.NotWatched(logger.Printf, build.ReAsksBucket, err, "builder_ask_again is not heard until the builder starts again") - } else { - go func() { errs <- follow(ctx, rw, eachReAsk) }() - } - }) - // A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the - // builder alone may read the stream, so it answers there. - process.AwaitBucket(ctx, js, conn, build.QueuesBucket, "builder_seat_queue is not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) { - eachRead := func(e jetstream.KeyValueEntry) { queueRead(ctx, heldOnTheBus{js}, kv, e, logger.Printf) } - if qw, err := settle(ctx, kv, eachRead); err != nil { - process.NotWatched(logger.Printf, build.QueuesBucket, err, "builder_seat_queue is not answered until the builder starts again") - } else { - go func() { errs <- follow(ctx, qw, eachRead) }() - } - }) - logger.Printf("started on %s", cred.Node) - go func() { errs <- follow(ctx, w, each) }() - select { - case <-ctx.Done(): - b.Wait() - return nil - case err := <-errs: - return err - case err := <-placementErrs: - return err - case err := <-watch.errs: + cw, err := settle(ctx, candidates, eachCandidate) + if err != nil { return err } + go func() { _ = follow(ctx, cw, eachCandidate) }() + return follow(ctx, w, each) } // sourceHeads are the trunk heads of each revision of the gate's source.main the watch delivered, the newest diff --git a/modules/mesh-delivery-builder/go.mod b/modules/mesh-delivery-builder/go.mod index 3131c6f3..ec9702c5 100644 --- a/modules/mesh-delivery-builder/go.mod +++ b/modules/mesh-delivery-builder/go.mod @@ -25,3 +25,7 @@ require ( ) replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun => ./buildrun + +require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000 + +replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index f94b1720..dda6d84c 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -93,13 +93,13 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 if m.Module != "mesh-delivery-builder" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "builds", History: 5}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory + if want := []state{{Name: "builds", History: 5}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory t.Errorf("state = %+v, want %+v", m.State, want) } // 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and // holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5). if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-delivery-assigner.placement", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !reflect.DeepEqual(sorted(m.Reads), want) { + "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) { @@ -108,13 +108,22 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 // 1937/AC3: the builder asks the controller whether the artifact store still holds a done key's blobs (held), and no // other tool; it holds no capacity. // 1955/L7 (ADR 0571 §2): the seat mesh-artifact-sweep's held, the controller's `held` is no longer asked (issue 1955). - if want := []string{"seat:mesh-artifact-sweep.held"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 { + // DLV-M1 (ADR 0615 §1): the indexer's modules, manifests and candidates and the assigner's placement are asked by their + // read tools and followed by their views families, never read. + for _, r := range m.Reads { + if strings.HasPrefix(r, "mesh-delivery-indexer.") || strings.HasPrefix(r, "mesh-delivery-assigner.") { + t.Errorf("reads %s: the indexer's and the assigner's state is read by their tools and views families alone", r) + } + } + if want := []string{"mesh-delivery-assigner.assigner_read_placement", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests", + "mesh-delivery-indexer.indexer_read_modules", "seat:mesh-artifact-sweep.held"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 { t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want) } // The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it // starts; and by the seat mesh-artifact-sweep's artifacts-gone, a blob the sweep found gone (1937/AC4, 1955/L7), the // controller's no longer (1126/B9, ADR 0590). - if want := []string{"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported"}; !reflect.DeepEqual(sorted(m.Consumes), want) { + if want := []string{"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", + "mesh-delivery-assigner.views.placement", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules"}; !reflect.DeepEqual(sorted(m.Consumes), want) { t.Errorf("consumes = %v, want %v", m.Consumes, want) } if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" { @@ -124,7 +133,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1 t.Errorf("secrets-owner = %q", m.SecretsOwner) } // 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again. - if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_seat_queue", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) { + if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_first_view_again", "builder_first_views", "builder_seat_queue", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) { t.Errorf("tools = %v, want %v", m.Tools, want) } if want := []string{"build.asked-again"}; !reflect.DeepEqual(m.Emits, want) { diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index 4440f866..a99e4e97 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -8,7 +8,9 @@ "builder_build_again", "builder_builds_since", "builder_seat_queue", - "builder_stuck" + "builder_stuck", + "builder_first_views", + "builder_first_view_again" ], "emits": [ "build.asked-again" @@ -16,17 +18,27 @@ "consumes": [ "mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", - "mesh-build-node.toolchain.reported" + "mesh-build-node.toolchain.reported", + "mesh-delivery-assigner.views.placement", + "mesh-delivery-indexer.views.candidates", + "mesh-delivery-indexer.views.manifests", + "mesh-delivery-indexer.views.modules" ], "invokes": [ - "seat:mesh-artifact-sweep.held" + "seat:mesh-artifact-sweep.held", + "mesh-delivery-assigner.assigner_read_placement", + "mesh-delivery-indexer.indexer_read_candidates", + "mesh-delivery-indexer.indexer_read_manifests", + "mesh-delivery-indexer.indexer_read_modules" ], "state": [ { "name": "builds", "history": 5 }, { "name": "hearing", "history": 1 }, { "name": "instances", "history": 1 }, { "name": "queues", "history": 1 }, - { "name": "reasks", "history": 1 } + { "name": "reasks", "history": 1 }, + { "name": "first-views", "history": 1 }, + { "name": "findings", "history": 1 } ], "logs": [ { "name": "build-log", "max-age-days": 30 } @@ -37,11 +49,7 @@ "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-delivery-assigner.placement", - "mesh-delivery-gate.source", - "mesh-delivery-indexer.candidates", - "mesh-delivery-indexer.manifests", - "mesh-delivery-indexer.modules" + "mesh-delivery-gate.source" ], "uses": [ "node-build-go",