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 1066135e..bf067462 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -53,6 +53,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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 { + return err + } // The registrations and each watched branch's catalog half, first views then events (§21, §195, §200): until a // branch's view is held the runner acts on nothing new of it, and says so. calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true)) @@ -87,7 +92,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Its asking half (§22, §24): a pipeline's steps are placed from the module's registration and each build seat is asked // 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} + Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module, + Arm: armBound(ctx, bounds)} resume := func(branch string) { half, ok := rv.Catalog.View(branch) if !ok { @@ -104,6 +110,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hearStepEnded(hub, hearer, logger.Printf); err != nil { return err } + if err := hearAskTaken(ctx, hub, bounds, logger.Printf); err != nil { + return err + } + if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil { + return err + } if err := hearDecided(hub, hearer, cred.Module, logger.Printf); err != nil { return err } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go index 6b05c102..0e026400 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -10,6 +10,7 @@ import ( "github.com/nats-io/nats.go/jetstream" seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/process" @@ -132,3 +133,69 @@ func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string, } return nil } + +// hearAskTaken deletes the bound of each ask a build seat's holder took (§602, §647): the ask is answered, so its five minutes +// end with no finding. +func hearAskTaken(ctx context.Context, hub *process.EventHub, bounds jetstream.KeyValue, logf func(string, ...any)) error { + for _, ts := range pipeline.TakenSeats { + ts := ts + err := hub.Handle(ts.Subject, func(m jetstream.Msg) { + var ev nodebuildgo.AskTaken + if err := json.Unmarshal(m.Data(), &ev); err != nil || ev.Ask == "" { + logf("%s is no ask-taken: %v", m.Subject(), err) + _ = m.Term() + return + } + if err := bounds.Delete(ctx, ev.Ask); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + logf("the bound of ask %s, taken on %s, is not deleted, heard again: %v", ev.Ask, ev.Node, err) + _ = m.Nak() + return + } + _ = m.Ack() + }) + if err != nil { + return err + } + } + return nil +} + +// armBound is the bound of an ask: a key of the state bounds with its own time to live, named for the ask. +func armBound(ctx context.Context, bounds jetstream.KeyValue) pipeline.Arm { + return func(ask string) error { + _, err := bounds.Create(ctx, ask, []byte(`{"ask":"`+ask+`"}`), jetstream.KeyTTL(pipeline.AskBound)) + if errors.Is(err, jetstream.ErrKeyExists) { + return nil + } + return err + } +} + +// watchBounds hears the end of each ask's bound, the bus's marker of the key's expiry, and ends the step no holder took (§32, +// §535, A-77): no timer in the process. +func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hearer, logf func(string, ...any)) error { + w, err := bounds.WatchAll(ctx, jetstream.UpdatesOnly()) + if err != nil { + return err + } + go func() { + for { + select { + case <-ctx.Done(): + _ = w.Stop() + return + case e, ok := <-w.Updates(): + if !ok { + return + } + if e == nil || e.Operation() != jetstream.KeyValuePurge || !pipeline.IsAskKey(e.Key()) { + continue + } + if err := h.NotTaken(e.Key(), pipeline.AskBound); err != nil { + logf("the step of ask %s, which no holder took, is not ended: %v", e.Key(), err) + } + } + } + }() + return nil +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go b/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go new file mode 100644 index 00000000..94d3757f --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go @@ -0,0 +1,95 @@ +package pipeline + +import ( + "errors" + "fmt" + "strings" + "time" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" + nodebuildimage "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage" + nodebuildnode "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode" +) + +// AskBound is how long an ask on a build seat's work queue may wait for its holder to take it (design 47 G §535, §647): five +// minutes, the bound of a key of the state `bounds` with its own time to live (A-77, A-104), never a timer. +const AskBound = 5 * time.Minute + +// TakenSeat is a build seat whose holder says `ask-taken` when it takes an ask (design 47 G §602): where the runner hears it. +type TakenSeat struct { + Seat string + // Subject hears every node's ask-taken of the seat. + Subject string +} + +// TakenSeats are the three build seats, from their contracts: the seat mesh-builds and the seat mesh-wants say no ask-taken, +// so no bound is armed for their asks. +var TakenSeats = []TakenSeat{ + {Seat: nodebuildgo.Seat, Subject: nodebuildgo.SubjectAskTakenAll}, + {Seat: nodebuildnode.Seat, Subject: nodebuildnode.SubjectAskTakenAll}, + {Seat: nodebuildimage.Seat, Subject: nodebuildimage.SubjectAskTakenAll}, +} + +// BoundSeat says whether the holder of a seat says ask-taken, so an ask put on its work queue has a bound. +func BoundSeat(name string) bool { + for _, s := range TakenSeats { + if s.Seat == name { + return true + } + } + return false +} + +// Arm puts a bound of one ask's wait for its holder: a key of the state `bounds` named for the ask, written with its own +// time to live. It is armed before the ask is put, so an ask-taken heard at once finds it. +type Arm func(ask string) error + +// NotTaken ends the step asked under ask that no holder took within its bound, the bound's expiry heard (design 47 G §32, +// §535, §647, A-77): the step ends `could not run` naming the seat, so its pipeline ends failed once the others end, never +// a silent wait. A step that ended or was never asked of a seat takes nothing. +func (h *Hearer) NotTaken(ask string, bound time.Duration) error { + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + return fmt.Errorf("the pipelines are not listed: %w", err) + } + for _, key := range pkeys { + p, _, err := h.Store.Get(key) + if err != nil || p.Outcome != seat.PipelineOutcomeRunning || stepOfAsk(p, ask) < 0 { + continue + } + return h.endNotTaken(key, ask, bound) + } + return nil +} + +func (h *Hearer) endNotTaken(key, ask string, bound time.Duration) error { + for { + p, rev, err := h.Store.Get(key) + if err != nil { + return fmt.Errorf("the pipeline %s is not read: %w", key, err) + } + i := stepOfAsk(p, ask) + if i < 0 || p.Steps[i].Outcome != seat.StepResultOutcomeRunning { + return nil + } + now := h.now() + s := &p.Steps[i] + s.Outcome, s.Ended = seat.StepResultOutcomeCouldNotRun, &now + s.Why = fmt.Sprintf("the seat %s has no holder that took ask %s within %s", s.Seat, ask, bound) + ended := endPipeline(&p, now) + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return fmt.Errorf("pipeline %s cannot be written: %w", key, err) + } + h.logEnd(key, *s, now) + if err := h.say(seat.EventStepEnded, p, s); err != nil { + return fmt.Errorf("pipeline %s's %s is written but not said: %w", key, seat.EventStepEnded, err) + } + return h.said(p, false, ended) + } +} + +// IsAskKey says whether a key of the state `bounds` is an ask's bound. +func IsAskKey(key string) bool { return strings.HasPrefix(key, "step-") && !strings.Contains(key, ".") } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/place.go b/modules/mesh-delivery-pipeline/internal/pipeline/place.go index 3a89ddd8..885a44d4 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/place.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/place.go @@ -136,11 +136,20 @@ 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, key string, asks []Asked) error { +func askAll(asker StepAsker, arm Arm, key string, asks []Asked) error { var first error for _, a := range asks { accept, ok := acceptStep[a.Seat] 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). + if arm != nil && BoundSeat(a.Seat) { + if aerr := arm(a.ID); aerr != nil { + err = fmt.Errorf("its bound is not armed: %w", aerr) + ok = false + } + } + } if ok { err = asker.AskStep(accept, a.ID, a.Ask) } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 3bfc6e56..e5366bb9 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -60,6 +60,8 @@ type Starter struct { // asker. Steps StepAsker Name string + // Arm arms the bound of an ask put on a build seat's work queue (§535, §647); nil arms none. + Arm Arm } // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second @@ -104,7 +106,7 @@ func (s *Starter) Start(b Begin) (bool, error) { s.logf("pipeline-started of %s is not said: %v", key, err) } if s.Steps != nil && len(asks) > 0 { - if err := askAll(s.Steps, key, asks); err != nil { + if err := askAll(s.Steps, s.Arm, key, asks); err != nil { return true, err } } @@ -161,7 +163,7 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) { if len(due) == 0 { continue } - if err := askAll(s.Steps, key, due); err != nil && first == nil { + if err := askAll(s.Steps, s.Arm, key, due); err != nil && first == nil { first = err } asked += len(due) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go index 331cb6b8..a1e73cc7 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go @@ -9,6 +9,7 @@ import ( "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + wants "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" "git.novox.be/novox/mesh-sdk/go/keys" ) @@ -298,3 +299,50 @@ func TestAnOptionalStepStillWaitingHoldsNoEnd(t *testing.T) { } } } + +// An ask no holder took within its bound ends its step `could not run` naming the seat, and with the rest ended the pipeline +// failed (design 47 G §32, §535, §647, A-104); a step that ended, or ask nobody placed, ends nothing. +func TestAnAskNoHolderTookEndsItsStepNamingTheSeat(t *testing.T) { + h, store, said, _, _, _ := fixture() + e := store[pkey] + e.p.Steps[1].Outcome = seat.StepResultOutcomePassed // the unit step ended; only lint is waiting for a holder + e.p.Steps[2].Outcome = seat.StepResultOutcomeNotAsked + store[pkey] = e + if err := h.NotTaken("nobody-asked", AskBound); err != nil || store[pkey].p.Steps[0].Outcome != seat.StepResultOutcomeRunning { + t.Fatalf("an ask no step names ended something: %v", err) + } + if err := h.NotTaken("a1", AskBound); err != nil { + t.Fatal(err) + } + p := store[pkey].p + if p.Steps[0].Outcome != seat.StepResultOutcomeCouldNotRun || !strings.Contains(p.Steps[0].Why, nodebuildgo.Seat) || + p.Outcome != seat.PipelineOutcomeFailed || !strings.Contains(strings.Join(*said, " "), seat.EventPipelineEnded) { + t.Errorf("step %+v, pipeline %s (%s), said %v: want could not run naming %s and a failed pipeline", p.Steps[0], p.Outcome, p.Why, *said, nodebuildgo.Seat) + } + // Its end taken once: the bound's expiry heard again changes nothing. + n := len(*said) + if err := h.NotTaken("a1", AskBound); err != nil || len(*said) != n { + t.Errorf("a second expiry said %d more events: %v", len(*said)-n, err) + } +} + +// The bound is armed before the ask is put, only for the build seats whose holder says ask-taken, and an ask whose bound is +// not armed is not put (§535, §647). +func TestTheBoundIsArmedBeforeAnAskIsPut(t *testing.T) { + var order []string + 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" { + 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 { + t.Errorf("an ask whose bound cannot be armed was put or said nothing: %v %v", order, err) + } +} + +type askerFunc func(accept, id string, body any) error + +func (f askerFunc) AskStep(accept, id string, body any) error { return f(accept, id, body) } diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index ed457e40..813ecbd6 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -21,8 +21,10 @@ func TestManifest(t *testing.T) { t.Fatalf("the manifest does not read: %v", err) } var states, streams, objects []string + keyTTL := map[string]bool{} for _, s := range m.State { states = append(states, s.Name) + keyTTL[s.Name] = s.KeyTTL } for _, s := range m.Streams { streams = append(streams, s.Name) @@ -48,7 +50,9 @@ 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", "findings", "first-views"})}, + {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "findings", "first-views"})}, + // §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"]}, // §27, §85: the step log on an old stream, its outputs in an old objects section. // RN24 (§383, §387): beside it, the finding stream its findings are said on. {"old streams section", equalSets(streams, []string{seat.StateStepLog, "mesh-delivery-pipeline-findings"})}, @@ -63,7 +67,7 @@ func TestManifest(t *testing.T) { {"consumes", equalSets(m.Consumes, []string{ "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", "mesh-wants.step-ended", "mesh-catalog.registration-changed", + "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"})}, // 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). diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index 5ead274a..a9e6e8bd 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -35,6 +35,9 @@ "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", @@ -68,6 +71,11 @@ "name": "aged", "history": 1 }, + { + "name": "bounds", + "history": 1, + "key-ttl": true + }, { "name": "findings", "history": 1