diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index 24216b05..fbc8b195 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -59,6 +59,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + // The commits named a recovery, whose steps are asked on the stream delivery (§81, §86, A-207). + recovered, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, "recoveries"), "know a recovery's commit") + if err != nil { + return err + } // The bound of each ask a build seat's holder must take: a key with its own time to live, its expiry heard (§535, A-77). bounds, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, "bounds"), "bound an ask no holder takes") if err != nil { @@ -117,7 +122,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // for its step. starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say, Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module, - Arm: armBound(ctx, bounds), Fills: fills, Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)} + Arm: armBound(ctx, bounds), Fills: fills, + Recovered: kvRecoveries{ctx, recovered}.Has, Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)} resume := func(branch string) { half, ok := rv.Catalog.View(branch) if !ok { @@ -149,6 +155,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hearPublished(ctx, heard, starter, calls, rv.Catalog, logger.Printf); err != nil { return err } + if err := hearRecovery(heard, kvRecoveries{ctx, recovered}, logger.Printf); err != nil { + return err + } if err := hearTakenIn(heard, starter, rv.Catalog, logger.Printf); err != nil { return err } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries.go new file mode 100644 index 00000000..468b9019 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries.go @@ -0,0 +1,55 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + + "github.com/nats-io/nats.go/jetstream" + + meshdelivery "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" +) + +// recoveries is the runner's state `recoveries`: one key per commit named a recovery, holding the event that named it (design 47 +// G §81, §86, A-207). +type recoveries interface { + Put(commit string, raw []byte) error + Has(commit string) (bool, error) +} + +type kvRecoveries struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (r kvRecoveries) Put(commit string, raw []byte) error { + _, err := r.kv.Put(r.ctx, commit, raw) + return err +} + +func (r kvRecoveries) Has(commit string) (bool, error) { + _, err := r.kv.Get(r.ctx, commit) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + return err == nil, err +} + +// hearRecovery keeps each commit the seat mesh-delivery says a recovery merged in the state recoveries, which a pipeline's start +// and resume read (§81, §86): a commit heard after its pipeline started asks nothing again. +func hearRecovery(hub handlers, r recoveries, logf func(string, ...any)) error { + return hub.Handle(meshdelivery.SubjectRecoveryMerged, func(m jetstream.Msg) { + var ev meshdelivery.RecoveryMergedEvent + if err := json.Unmarshal(m.Data(), &ev); err != nil || ev.Commit == "" { + logf("%s is no recovery-merged: %v", m.Subject(), err) + _ = m.Term() + return + } + if err := r.Put(ev.Commit, m.Data()); err != nil { + logf("the recovery %s of %s is not kept, heard again: %v", ev.Commit, ev.Repository, err) + _ = m.Nak() + return + } + _ = m.Ack() + }) +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries_test.go new file mode 100644 index 00000000..476ee663 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries_test.go @@ -0,0 +1,35 @@ +package main + +import ( + "testing" +) + +type memRecoveries map[string][]byte + +func (m memRecoveries) Put(c string, raw []byte) error { m[c] = raw; return nil } +func (m memRecoveries) Has(c string) (bool, error) { _, ok := m[c]; return ok, nil } + +type dataMsg struct { + ackedMsg + data []byte + acks *[]string +} + +func (m dataMsg) Data() []byte { return m.data } +func (m dataMsg) Term() error { *m.acks = append(*m.acks, "term"); return nil } + +// A recovery-merged of the seat mesh-delivery is kept by its commit in the state recoveries; one that is no event is dropped +// (design 47 G §81, §86, A-207). +func TestARecoveryMergedIsKeptByItsCommit(t *testing.T) { + hub := &oneHub{} + kept := memRecoveries{} + if err := hearRecovery(hub, kept, func(string, ...any) {}); err != nil { + t.Fatal(err) + } + var acks []string + hub.handle(dataMsg{data: []byte(`{"v":1,"repository":"o/app","commit":"c9","why":"the build is down","by":"s@n"}`), acks: &acks}) + hub.handle(dataMsg{data: []byte(`not json`), acks: &acks}) + if ok, _ := kept.Has("c9"); !ok || len(kept) != 1 || len(acks) != 1 || acks[0] != "term" { + t.Errorf("kept %v, acks %v: want c9 kept and the other dropped", kept, acks) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go index 1053f872..a5823e8c 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go @@ -57,7 +57,7 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes defer cancel() const module = "mesh-delivery-pipeline" - for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views", "consumers", "aged"} { + for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views", "consumers", "aged", "recoveries"} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_" + state, History: 1}); err != nil { t.Fatal(err) } diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod index 39856e0a..2f4e727f 100644 --- a/modules/mesh-delivery-pipeline/go.mod +++ b/modules/mesh-delivery-pipeline/go.mod @@ -31,4 +31,4 @@ require ( require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000 -require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 // indirect +require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/place.go b/modules/mesh-delivery-pipeline/internal/pipeline/place.go index 05eee910..8e3bb91a 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/place.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/place.go @@ -32,6 +32,14 @@ var acceptStep = map[string]string{ nodebuildimage.Seat: nodebuildimage.AcceptSubjectCheckStep, } +// acceptRecovery is where a recovery's step is sent on the build seats' stream `delivery`, ahead of the first round (§86, §415); +// the seats mesh-builds and mesh-wants have no such stream, so their steps stay on their own accept subject. +var acceptRecovery = map[string]string{ + nodebuildgo.Seat: nodebuildgo.AcceptSubjectDeliveryStep, + nodebuildnode.Seat: nodebuildnode.AcceptSubjectDeliveryStep, + nodebuildimage.Seat: nodebuildimage.AcceptSubjectDeliveryStep, +} + // buildSteps are the steps an artifact's build seat runs, in the order they are asked (design 47 G §22). var buildSteps = []string{seat.StepResultStepLint, seat.StepResultStepUnit, seat.StepResultStepIntegration} @@ -143,10 +151,13 @@ func stepAsk(p seat.Pipeline, e catalog.CatalogEntry, a catalog.Artifact, step, } // askAll puts each ask on its seat's work queue; the first refusal is named, the rest are still put. -func askAll(asker StepAsker, arm Arm, key string, asks []Asked) error { +func askAll(asker StepAsker, arm Arm, key string, asks []Asked, recovery bool) error { var first error for _, a := range asks { accept, ok := acceptStep[a.Seat] + if to, named := acceptRecovery[a.Seat]; recovery && named { + accept = to + } err := fmt.Errorf("the seat %s is no build seat the runner asks", a.Seat) if ok { // A build seat's holder says ask-taken: the ask's bound is armed before it is put (§535, §647). diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index ad70f5b9..da3b6845 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -72,6 +72,8 @@ type Starter struct { Arm Arm // Fills refuses a start while a store of the runner is full (§366); nil refuses none. Fills *Fills + // Recovered says whether a commit is named a recovery (§81, §86): its steps are asked on the build seats' stream delivery. + Recovered func(commit string) (bool, error) } // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second @@ -120,7 +122,7 @@ func (s *Starter) Start(b Begin) (bool, error) { } if s.Steps != nil && len(asks) > 0 { s.carry(key, asks) - if err := askAll(s.Steps, s.Arm, key, asks); err != nil { + if err := askAll(s.Steps, s.Arm, key, asks, s.recovered(key, b.Commit)); err != nil { return true, err } } @@ -178,7 +180,7 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) { continue } s.carry(key, due) - if err := askAll(s.Steps, s.Arm, key, due); err != nil && first == nil { + if err := askAll(s.Steps, s.Arm, key, due, s.recovered(key, p.Commit)); err != nil && first == nil { first = err } asked += len(due) @@ -186,6 +188,19 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) { return asked, first } +// recovered says whether the commit of pipeline key is named a recovery; a read that fails asks on the stream check, said, since +// a step is asked either way (a recovery's only goes ahead of the first round). +func (s *Starter) recovered(key, commit string) bool { + if s.Recovered == nil || commit == "" { + return false + } + named, err := s.Recovered(commit) + if err != nil { + s.logf("whether %s of %s is a recovery is not read, so its steps are asked on check: %v", commit, key, err) + } + return named +} + // carry puts the kept results of each ask's package and the toolchain of its language on its step ask (§25, §416, A-104). A // read that fails asks the steps with none, so every package runs: slower, never wrong; it is said. func (s *Starter) carry(key string, asks []Asked) { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go index 7f8e832e..23506089 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go @@ -335,12 +335,12 @@ func TestTheBoundIsArmedBeforeAnAskIsPut(t *testing.T) { asker := askerFunc(func(accept, id string, _ any) error { order = append(order, "put "+id); return nil }) arm := Arm(func(ask string) error { order = append(order, "arm "+ask); return nil }) asks := []Asked{{Seat: nodebuildgo.Seat, ID: "step-1", Ask: 1}, {Seat: wants.Seat, ID: "step-2", Ask: 2}} - if err := askAll(asker, arm, "k", asks); err != nil || strings.Join(order, ",") != "arm step-1,put step-1,put step-2" { + if err := askAll(asker, arm, "k", asks, false); err != nil || strings.Join(order, ",") != "arm step-1,put step-1,put step-2" { t.Errorf("order %v, %v; want the build seat's bound armed first and mesh-wants' ask with none", order, err) } order = nil failing := Arm(func(string) error { return errors.New("bucket gone") }) - if err := askAll(asker, failing, "k", asks[:1]); err == nil || len(order) != 0 { + if err := askAll(asker, failing, "k", asks[:1], false); err == nil || len(order) != 0 { t.Errorf("an ask whose bound cannot be armed was put or said nothing: %v %v", order, err) } } @@ -348,3 +348,30 @@ func TestTheBoundIsArmedBeforeAnAskIsPut(t *testing.T) { type askerFunc func(accept, id string, body any) error func (f askerFunc) AskStep(accept, id string, body any) error { return f(accept, id, body) } + +// A recovery's steps are asked on the build seats' stream delivery, ahead of the first round; the seats with no such stream keep +// their own accept subject; every other commit's steps go on check (design 47 G §86, §415). +func TestARecoverysStepsAreAskedOnTheStreamDelivery(t *testing.T) { + asks := []Asked{{Seat: nodebuildgo.Seat, ID: "s1", Ask: 1}, {Seat: wants.Seat, ID: "s2", Ask: 2}} + for _, c := range []struct { + recovery bool + want string + }{{true, nodebuildgo.AcceptSubjectDeliveryStep + "," + wants.AcceptSubjectStep}, {false, nodebuildgo.AcceptSubjectCheckStep + "," + wants.AcceptSubjectStep}} { + var to []string + asker := askerFunc(func(accept, _ string, _ any) error { to = append(to, accept); return nil }) + if err := askAll(asker, nil, "k", asks, c.recovery); err != nil || strings.Join(to, ",") != c.want { + t.Errorf("recovery %v: asked on %v (%v), want %s", c.recovery, to, err, c.want) + } + } + // Starter.recovered reads the commit; a failed read asks on check and says so. + var said []string + s := &Starter{Logf: func(f string, a ...any) { said = append(said, f) }, Recovered: func(c string) (bool, error) { + if c == "bad" { + return false, errors.New("unread") + } + return c == "rec", nil + }} + if !s.recovered("k", "rec") || s.recovered("k", "other") || s.recovered("k", "bad") || len(said) != 1 { + t.Errorf("recovered: said %v", said) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go index e3caef80..9f4c587c 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go @@ -281,7 +281,7 @@ func (v Verbs) Rerun(caller string, in seat.RerunInput) (seat.RerunAnswer, error } if ask != nil { v.Starter.carry(in.Pipeline, []Asked{*ask}) - if err := askAll(v.Starter.Steps, v.Starter.Arm, in.Pipeline, []Asked{*ask}); err != nil { + if err := askAll(v.Starter.Steps, v.Starter.Arm, in.Pipeline, []Asked{*ask}, v.Starter.recovered(in.Pipeline, p.Commit)); err != nil { return seat.RerunAnswer{}, fmt.Errorf("the step is written running but not asked, it is asked again at the next start: %w", err) } } diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index b5fae04f..191c9b9d 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -54,7 +54,7 @@ func TestManifest(t *testing.T) { {"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat && equalSets(m.Claims[0].Serves, seat.Verbs())}, // §27, §85, §330, §355, §77: the old state sections. - {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views", "consumers"})}, + {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "recoveries", "bounds", "toolchains", "findings", "first-views", "consumers"})}, // §535, §647, A-77, A-104: an ask no holder takes is bound by a key with its own time to live. {"bounds is key-ttl", keyTTL["bounds"]}, // RN25 (§469, §85): an old state section declares its cap, pipelines 256 MiB and kept 128 MiB. @@ -78,7 +78,7 @@ func TestManifest(t *testing.T) { "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", "operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended", "node-build-image.step-ended", "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", "mesh-wants.step-ended", "mesh-catalog.registration-changed", - "mesh-catalog.catalog-revision", "mesh-conditions.findings-asked"})}, + "mesh-catalog.catalog-revision", "mesh-conditions.findings-asked", "mesh-delivery.recovery-merged"})}, // RN2 (§21, §195, §200, §77): the registrations and the catalog half asked by their verbs, the first views read and // asked again by hand (a hand act). {"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act", @@ -90,7 +90,9 @@ func TestManifest(t *testing.T) { "seat:mesh-module-assigner.assignments", // A-102 (§22, §24): the accepted kinds the runner puts its steps on. "seat:mesh-builds.step", "seat:mesh-wants.step", "seat:node-build-go.check.step", "seat:node-build-node.check.step", - "seat:node-build-image.check.step"})}, + "seat:node-build-image.check.step", + // RN22 (§86, §415, A-207): a recovery's steps on the build seats' stream delivery. + "seat:node-build-go.delivery.step", "seat:node-build-node.delivery.step", "seat:node-build-image.delivery.step"})}, {"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})}, // RN25 (§539): the runner's tools carry the prefix runner_. {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health"})}, diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index ac444a60..6d2ca9cc 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -42,7 +42,8 @@ "mesh-wants.step-ended", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", - "mesh-conditions.findings-asked" + "mesh-conditions.findings-asked", + "mesh-delivery.recovery-merged" ], "invokes": [ "seat:mesh-catalog.registrations", @@ -55,7 +56,10 @@ "seat:mesh-wants.step", "seat:node-build-go.check.step", "seat:node-build-node.check.step", - "seat:node-build-image.check.step" + "seat:node-build-image.check.step", + "seat:node-build-go.delivery.step", + "seat:node-build-node.delivery.step", + "seat:node-build-image.delivery.step" ], "uses": [ "operator-channel" @@ -75,6 +79,10 @@ "name": "aged", "history": 1 }, + { + "name": "recoveries", + "history": 1 + }, { "name": "toolchains", "history": 1