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 9b8e5c35..ab75f8c3 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -48,6 +48,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + kept, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, seat.StateKept), "keep a package's passing result") + 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)) @@ -62,14 +66,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } // A pipeline starts by compare-and-set on its key's absence and says pipeline-started (§54, §103); a publish starts // one per dependent behind (§53, §334). - starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, - Say: func(subject string, body any) error { - data, err := json.Marshal(body) - if err == nil { - _, err = js.Publish(ctx, subject, data) - } - return err - }} + say := func(subject string, body any) error { + data, err := json.Marshal(body) + if err == nil { + _, err = js.Publish(ctx, subject, data) + } + return err + } + starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say} + if err := hearStepEnded(hub, newHearer(ctx, js, cred, pipelines, kept, say, logger.Printf), logger.Printf); err != nil { + return err + } if err := hearPublished(ctx, hub, starter, calls, rv.Catalog, 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 new file mode 100644 index 00000000..b8924c17 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -0,0 +1,127 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "strings" + + "github.com/nats-io/nats.go/jetstream" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// pipelineStore is the state pipelines as the hearer reads and writes it: a write by compare-and-set on the revision read (§27). +type pipelineStore struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (s pipelineStore) Get(key string) (seat.Pipeline, uint64, error) { + e, err := s.kv.Get(s.ctx, key) + if err != nil { + return seat.Pipeline{}, 0, err + } + var p seat.Pipeline + if err := json.Unmarshal(e.Value(), &p); err != nil { + return seat.Pipeline{}, 0, fmt.Errorf("the pipeline %s is no pipeline: %w", key, err) + } + return p, e.Revision(), nil +} + +func (s pipelineStore) Update(key string, p seat.Pipeline, revision uint64) (uint64, error) { + raw, err := json.Marshal(p) + if err != nil { + return 0, err + } + rev, err := s.kv.Update(s.ctx, key, raw, revision) + if err != nil && (errors.Is(err, jetstream.ErrKeyExists) || strings.Contains(err.Error(), "wrong last sequence")) { + return 0, pipeline.ErrChanged + } + return rev, err +} + +func (s pipelineStore) Keys(filter string) ([]string, error) { + all, err := s.kv.Keys(s.ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return nil, nil + } + var out []string + want := strings.Split(filter, ".") + for _, k := range all { + got := strings.Split(k, ".") + if len(got) != len(want) { + continue + } + ok := true + for i := range want { + ok = ok && (want[i] == "*" || want[i] == got[i]) + } + if ok { + out = append(out, k) + } + } + return out, err +} + +// newHearer is the hearer over the runner's own states: pipelines, kept, the step log on its stream and step-outputs on its +// objects (§27). Each store is opened where it is used, so one the grants part has not made yet fails that write loudly. +func newHearer(ctx context.Context, js jetstream.JetStream, cred process.Account, pipelines, kept jetstream.KeyValue, + say func(string, any) error, logf func(string, ...any)) *pipeline.Hearer { + return &pipeline.Hearer{Store: pipelineStore{ctx: ctx, kv: pipelines}, Say: say, Logf: logf, Name: seat.Seat, + Log: func(e seat.LogEntry) error { + return say(cred.EventSubject("step-log."+keys.EscapeName(e.Pipeline)), e) + }, + Keep: func(key string, k seat.KeptResult) error { + raw, err := json.Marshal(k) + if err == nil { + _, err = kept.Put(ctx, key, raw) + } + return err + }, + Output: func(key string, data []byte) error { + o, err := js.ObjectStore(ctx, catalogue.ObjectsBucket(cred.Module, seat.StateStepOutputs)) + if err != nil { + return fmt.Errorf("the objects %s are not open: %w", seat.StateStepOutputs, err) + } + _, err = o.PutBytes(ctx, key, data) + return err + }} +} + +// hearStepEnded takes each build seat's step-ended into its pipeline (§24; ADR 0666 §21): refused and said, never taken, +// when it is not as the runner asked it, and heard again when a write failed. +func hearStepEnded(hub *process.EventHub, h *pipeline.Hearer, logf func(string, ...any)) error { + for _, bs := range pipeline.StepSeats { + bs := bs + err := hub.Handle(bs.Prefix+"*", func(m jetstream.Msg) { + var ev pipeline.StepEnded + if err := json.Unmarshal(m.Data(), &ev); err != nil { + logf("%s is no step-ended: %v", m.Subject(), err) + _ = m.Term() + return + } + _, err := h.OnStepEnded(bs, m.Subject(), ev) + switch { + case errors.Is(err, pipeline.ErrRefused): + logf("%v", err) + _ = m.Term() + case err != nil: + logf("the step-ended of ask %s is not taken yet, heard again: %v", ev.ID, err) + _ = m.Nak() + default: + _ = m.Ack() + } + }) + if err != nil { + return err + } + } + return nil +} diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod index 1f8869f5..50178e66 100644 --- a/modules/mesh-delivery-pipeline/go.mod +++ b/modules/mesh-delivery-pipeline/go.mod @@ -13,6 +13,11 @@ require ( git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/clients/meshpipelinerunnerclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildimage v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildnode v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 github.com/klauspost/compress v1.20.1 // indirect diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/approval.go b/modules/mesh-delivery-pipeline/internal/pipeline/approval.go new file mode 100644 index 00000000..e0367e8b --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/approval.go @@ -0,0 +1,218 @@ +package pipeline + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "strings" + "time" + + "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" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// The approval step is an authorising ask of tier approve, the seat operator-channel's accepted kind ask under the +// runner's own name, its about the module and the watched branch only, so a newer commit's ask replaces the older one's +// (design 47 G §26, §37; ADR 0666 §10). It is asked last, once every other required step passed. +const ( + OptionApprove = "approve" + OptionDecline = "decline" +) + +// About is what an approval ask is about: the module and the watched branch, each a key's name, joined by a dot. +func About(module, branch string) string { + return keys.EscapeName(module) + "." + keys.EscapeName(branch) +} + +// approvalDue is the index of the approval step to ask now (§26): waiting, not bypassed, every other step ended and every +// other required step passed or carried; -1 when none is due. +func approvalDue(p seat.Pipeline) int { + at := -1 + for i, s := range p.Steps { + if s.Step == seat.StepResultStepApproval { + if s.Status != seat.StepResultStatusBypass && s.Outcome == seat.StepResultOutcomeWaiting { + at = i + } + continue + } + if s.Outcome == seat.StepResultOutcomeWaiting || s.Outcome == seat.StepResultOutcomeRunning { + return -1 + } + if s.Status == seat.StepResultStatusRequired && s.Outcome != seat.StepResultOutcomePassed && s.Outcome != seat.StepResultOutcomeCarried { + return -1 + } + } + return at +} + +// ApprovalAsk is the ask of a pipeline's approval step. Its id and its expiry follow from the pipeline and from when the +// step was asked, so the ask built again for the same pipeline is the same ask, digest included: a warrant is held to it. +func ApprovalAsk(p seat.Pipeline, from time.Time) (oc.Ask, error) { + sum := sha256.Sum256([]byte(p.Pipeline + "\n" + p.Commit)) + act := func(do string) (string, error) { + return operatorchannelclient.ActDigest(operatorchannelclient.Act{"pipeline": p.Pipeline, "commit": p.Commit, + "fingerprint": p.Fingerprint, "do": do}) + } + approve, err := act(OptionApprove) + if err != nil { + return oc.Ask{}, err + } + decline, err := act(OptionDecline) + if err != nil { + return oc.Ask{}, err + } + a := oc.Ask{ID: "approval-" + hex.EncodeToString(sum[:12]), About: About(p.Module, p.Branch), Who: operatorchannelclient.Operator, + Headline: trimRunes("Deliver "+p.Module+" from "+p.Branch+"?", operatorchannelclient.HeadlineLength), + Explanation: fmt.Sprintf("Every required step of %s's pipeline at commit %s on %s passed; its policy asks your approval before it is delivered.", p.Module, p.Commit, p.Branch), + Details: "pipeline " + p.Pipeline + ", fingerprint " + p.Fingerprint, + Expires: operatorchannelclient.FormatTime(from.Add(operatorchannelclient.ApproveLasts)), + OnExpiry: "the step ends declined and the commit is not delivered; ask-again asks once more", + Options: []oc.Option{ + {ID: OptionApprove, Label: "Approve", Does: "delivers this commit of " + p.Module, Level: operatorchannelclient.Approve, Binds: approve}, + {ID: OptionDecline, Label: "Decline", Does: "ends the step declined; nothing is delivered", Level: operatorchannelclient.Approve, Binds: decline}, + }} + return a, operatorchannelclient.Check(a, from) +} + +func trimRunes(s string, n int) string { + r := []rune(s) + if len(r) <= n { + return s + } + return string(r[:n-1]) + "…" +} + +// Asker is the runner's side of the operator-channel: Ask publishes an ask under the runner's name, its duplicate id the +// ask's own, so an ask published twice is kept once. +type Asker interface { + Ask(oc.Ask) error +} + +// askApproval asks p's approval step when it is due, before the pipeline is written: an ask published and then not +// written is published again, the same ask, when its step's end is heard again. +func askApproval(key string, p *seat.Pipeline, from time.Time, asker Asker) (bool, error) { + i := approvalDue(*p) + if i < 0 || asker == nil { + return false, nil + } + a, err := ApprovalAsk(*p, from) + if err != nil { + return false, fmt.Errorf("the approval of %s cannot be asked: %w", key, err) + } + if err := asker.Ask(a); err != nil { + return false, fmt.Errorf("the approval of %s is not asked: %w", key, err) + } + s := &p.Steps[i] + s.Outcome, s.Ask, s.Seat, s.Started = seat.StepResultOutcomeRunning, a.ID, seat.Seat, &from + s.Why = "waiting on the operator" + return true, nil +} + +// ErrNoAsk is a warrant naming no approval step the runner keeps for its about. +var ErrNoAsk = errors.New("no approval step asked under the warrant's ask") + +// decide ends an approval step on its warrant: a chosen approve, held to the ask rebuilt from the step, ends it passed; a +// chosen decline, an expiry, a cancel or a refusal ends it declined; a replacement by a newer commit's ask ends it +// superseded (design 47 G §37). It says whether the step changed. +func decide(p *seat.Pipeline, w oc.Warrant, asker string, now time.Time) (bool, error) { + for i := range p.Steps { + s := &p.Steps[i] + if s.Step != seat.StepResultStepApproval || s.Ask != w.Ask || s.Started == nil { + continue + } + if s.Outcome != seat.StepResultOutcomeRunning { + return false, nil + } + switch w.Outcome { + case oc.WarrantOutcomeChosen: + asked, err := ApprovalAsk(*p, *s.Started) + if err != nil { + return false, err + } + o, err := operatorchannelclient.For(w, asker, asked) + if err != nil { + return false, fmt.Errorf("the warrant for %s is not taken: %w", w.Ask, err) + } + s.Outcome = seat.StepResultOutcomeDeclined + if o.ID == OptionApprove { + s.Outcome = seat.StepResultOutcomePassed + } + if w.By != nil { + s.By = w.By.Identity + } + s.Why = "the operator chose " + o.ID + case oc.WarrantOutcomeReplaced: + s.Outcome, s.Why = seat.StepResultOutcomeSuperseded, "a newer commit's ask replaced it" + default: + s.Outcome, s.Why = seat.StepResultOutcomeDeclined, "the ask ended "+w.Outcome+" "+w.Words + } + s.Ended = &now + return true, nil + } + return false, ErrNoAsk +} + +// ErrNoRecord is a decided event whose ask the router's record holds no warrant for yet: the event is heard again. +var ErrNoRecord = errors.New("the router's record of the ask holds no warrant yet") + +// OnDecided takes the operator-channel's decided for the runner's own asks: it only wakes the runner (design 47 G A-89), which +// reads the warrant from the router's record of the ask through Where, never from the event's body (§37, §700). The approval +// step asked under that ask, among the pipelines of the module and watched branch the ask's about names, ends as decide says. +func (h *Hearer) OnDecided(event oc.Warrant) (seat.Pipeline, error) { + if event.Asker != h.Name { + return seat.Pipeline{}, fmt.Errorf("%w: the decided event is %s's, not %s's", ErrRefused, event.Asker, h.Name) + } + if h.Where == nil { + return seat.Pipeline{}, fmt.Errorf("the runner reads no router's record, so the ask %s is not taken", event.Ask) + } + _, rec, err := h.Where(event.Ask) + if err != nil { + return seat.Pipeline{}, fmt.Errorf("the router's record of the ask %s cannot be read: %w", event.Ask, err) + } + if rec == nil { + return seat.Pipeline{}, fmt.Errorf("%w: %s", ErrNoRecord, event.Ask) + } + w := *rec + module, branch, ok := strings.Cut(w.About, ".") + if !ok { + return seat.Pipeline{}, fmt.Errorf("%w: the warrant for %s is about %q, not a module and a branch", ErrNoAsk, w.Ask, w.About) + } + found, err := h.Store.Keys(branch + ".*." + module) + if err != nil { + return seat.Pipeline{}, err + } + for _, key := range found { + for { + p, rev, err := h.Store.Get(key) + if err != nil { + return seat.Pipeline{}, err + } + now := h.now() + changed, err := decide(&p, w, h.Name, now) + if errors.Is(err, ErrNoAsk) { + break + } + if err != nil || !changed { + return p, err + } + ended := endPipeline(&p, now) + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return p, err + } + i := stepOfAsk(p, w.Ask) + h.logEnd(key, p.Steps[i], now) + for _, ev := range []string{seat.EventWaitEnded, seat.EventStepEnded} { + if err := h.say(ev, p, &p.Steps[i]); err != nil { + return p, fmt.Errorf("pipeline %s's %s is written but not said: %w", key, ev, err) + } + } + return p, h.said(p, false, ended) + } + } + return seat.Pipeline{}, fmt.Errorf("%w: %s about %s", ErrNoAsk, w.Ask, w.About) +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go new file mode 100644 index 00000000..4920a321 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go @@ -0,0 +1,269 @@ +package pipeline + +import ( + "errors" + "fmt" + "strings" + "time" + + oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + 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" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// StepSeat is a seat whose holder runs the lint and test steps (design 47 G §22, §24): where the runner hears its +// step-ended, on the subject of the node that ran it. +type StepSeat struct { + Seat string + // Prefix is the subject before . + Prefix string +} + +// StepSeats are the three build seats, from their contracts. +var StepSeats = []StepSeat{ + {Seat: nodebuildgo.Seat, Prefix: strings.TrimSuffix(nodebuildgo.SubjectStepEndedAll, "*")}, + {Seat: nodebuildnode.Seat, Prefix: strings.TrimSuffix(nodebuildnode.SubjectStepEndedAll, "*")}, + {Seat: nodebuildimage.Seat, Prefix: strings.TrimSuffix(nodebuildimage.SubjectStepEndedAll, "*")}, +} + +// StepEnded is a build seat's step-ended. The three seats' bodies are one form, the part build-step's, held equal by +// their definitions, so the runner reads each in this one type. +type StepEnded = nodebuildgo.StepEnded + +// ErrRefused is a step-ended the runner does not take: not the node's holder's subject, or naming an ask, a commit or a +// fingerprint the runner did not ask (ADR 0666 §21; design 47 G §24). It is said, never taken. +var ErrRefused = errors.New("a step's end the runner does not take") + +// ErrChanged is an Updater's answer to an Update whose revision is no longer the key's: another write came between. +var ErrChanged = errors.New("the pipeline was written since it was read") + +// Updater is the state pipelines read and written by compare-and-set on the revision read (§27). +type Updater interface { + Get(key string) (seat.Pipeline, uint64, error) + Update(key string, p seat.Pipeline, revision uint64) (uint64, error) + // Keys are the keys matching a filter, `*` one key part. + Keys(filter string) ([]string, error) +} + +// Hearer takes the steps' ends into their pipelines and ends a pipeline whose every step ended (design 47 G §24, §27). +type Hearer struct { + Store Updater + // Say says one of the runner's events of a pipeline, on its subject; Log appends to the step log. + Say func(subject string, body any) error + Log func(seat.LogEntry) error + // Keep writes one package's passing result into the state kept, under its key (§27): only from a step's end the + // runner took, so a kept result is its seat holder's alone. + Keep func(key string, k seat.KeptResult) error + // Output keeps a step's output under key in the objects step-outputs. + Output func(key string, data []byte) error + // Asker publishes an approval step's ask (§37); Name is the runner's own name, the asker a warrant names. + Asker Asker + Name string + // Where reads the router's record of one of the runner's asks, as operatorchannelclient.Client.Where does: its state and + // its warrant (design 47 G A-89, §700). + Where func(ask string) (string, *oc.Warrant, error) + Logf func(string, ...any) + Now func() time.Time +} + +func (h *Hearer) now() time.Time { + if h.Now != nil { + return h.Now().UTC() + } + return time.Now().UTC() +} + +// OnStepEnded takes the step-ended heard on subject, from seat's holders. It is refused, and said, when the subject's +// node is not the one the body names, when no pipeline the runner keeps names the ask, or when the ask's step names +// another seat, commit, fingerprint, module or step (ADR 0666 §21; the bus grants step-ended. to that node's +// holder alone, so the subject's node is the account's). A step already ended takes nothing (the bus gives an event again). +func (h *Hearer) OnStepEnded(bs StepSeat, subject string, ev StepEnded) (seat.Pipeline, error) { + node := strings.TrimPrefix(subject, bs.Prefix) + if node == subject || node == "" || strings.Contains(node, ".") { + return seat.Pipeline{}, fmt.Errorf("%w: %s is not a subject of %s's step-ended", ErrRefused, subject, bs.Seat) + } + if ev.Node != node { + return seat.Pipeline{}, fmt.Errorf("%w: ask %s ended on %s's subject names node %q", ErrRefused, ev.ID, node, ev.Node) + } + for { + p, rev, err := h.Store.Get(ev.Pipeline) + if err != nil { + return seat.Pipeline{}, fmt.Errorf("%w: ask %s names pipeline %q, which the runner cannot read: %v", ErrRefused, ev.ID, ev.Pipeline, err) + } + i := stepOfAsk(p, ev.ID) + switch { + case i < 0: + return p, fmt.Errorf("%w: pipeline %s asked no step under ask %s", ErrRefused, ev.Pipeline, ev.ID) + case p.Steps[i].Seat != bs.Seat: + return p, fmt.Errorf("%w: ask %s was asked of %s, not %s", ErrRefused, ev.ID, p.Steps[i].Seat, bs.Seat) + case ev.Commit != p.Commit || ev.Fingerprint != p.Fingerprint || ev.Module != p.Module || ev.Step != p.Steps[i].Step: + return p, fmt.Errorf("%w: ask %s ended %s of %s at %s (%s), asked as %s at %s (%s)", ErrRefused, ev.ID, ev.Step, + ev.Module, ev.Commit, ev.Fingerprint, p.Steps[i].Step, p.Commit, p.Fingerprint) + case p.Steps[i].Outcome != seat.StepResultOutcomeRunning: + return p, nil + case !endOutcome(ev.Outcome): + return p, fmt.Errorf("%w: ask %s ended %q, not an outcome of step-ended", ErrRefused, ev.ID, ev.Outcome) + } + now := h.now() + s := &p.Steps[i] + s.Outcome, s.Node, s.Why, s.Rerun, s.By, s.Ended = ev.Outcome, node, ev.Said, ev.Rerun, ev.By, &now + if ev.Tests != nil { + s.Tests = &seat.TestCounts{Unit: ev.Tests.Unit, Integration: ev.Tests.Integration} + } + if ev.Output != "" && h.Output != nil { + name := seat.StepOutputsByPipelineStepArtifactKey(keys.EscapeName(ev.Pipeline), s.Step, keys.EscapeName(s.Artifact)) + if err := h.Output(name, []byte(ev.Output)); err != nil { + return p, fmt.Errorf("the output of ask %s cannot be kept: %w", ev.ID, err) + } + s.Output = name + } + from := now + if ev.Ended != nil { + from = ev.Ended.UTC() + } + asked, err := askApproval(ev.Pipeline, &p, from, h.Asker) + if err != nil { + return p, err + } + ended := endPipeline(&p, now) + if _, err := h.Store.Update(ev.Pipeline, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return p, fmt.Errorf("pipeline %s cannot be written: %w", ev.Pipeline, err) + } + if h.Keep != nil { + for _, k := range ev.Kept { + kept := seat.KeptResult{Digest: k.Digest, Level: k.Level, Package: k.Package, Pipeline: k.Pipeline} + if k.Toolchain != nil { + kept.Toolchain = &seat.ToolchainUsed{Language: k.Toolchain.Language, Recipe: k.Toolchain.Recipe, Version: k.Toolchain.Version} + } + if kept.Pipeline == "" { + kept.Pipeline = ev.Pipeline + } + if err := h.Keep(seat.KeptByLevelDigestKey(keys.EscapeName(k.Level), keys.EscapeName(k.Digest)), kept); err != nil { + h.logf("the passing result of %s (%s) at %s is not kept, so it runs again next time: %v", k.Package, k.Level, k.Digest, err) + } + } + } + h.logEnd(ev.Pipeline, *s, now) + if err := h.say(seat.EventStepEnded, p, s); err != nil { + return p, fmt.Errorf("pipeline %s's %s is written but not said: %w", ev.Pipeline, seat.EventStepEnded, err) + } + return p, h.said(p, asked, ended) + } +} + +// said says what a written pipeline became: an approval asked is a wait held on the operator (ADR 0283), never a stall. +func (h *Hearer) said(p seat.Pipeline, asked, ended bool) error { + if asked { + if err := h.say(seat.EventWaitHeld, p, nil); err != nil { + return fmt.Errorf("pipeline %s waits on the operator but %s is not said: %w", p.Pipeline, seat.EventWaitHeld, err) + } + } + if ended { + if err := h.say(seat.EventPipelineEnded, p, nil); err != nil { + return fmt.Errorf("pipeline %s ended but %s is not said: %w", p.Pipeline, seat.EventPipelineEnded, err) + } + } + return nil +} + +func (h *Hearer) logEnd(key string, s seat.StepResult, at time.Time) { + if h.Log == nil { + return + } + if err := h.Log(seat.LogEntry{Pipeline: key, Step: s.Step, Artifact: s.Artifact, What: seat.LogEntryWhatEnded, Outcome: s.Outcome, + Output: s.Output, At: at}); err != nil { + h.logf("the step log misses the end of %s of %s: %v", s.Step, key, err) + } +} + +func (h *Hearer) logf(format string, args ...any) { + if h.Logf != nil { + h.Logf(format, args...) + } +} + +// say says one event of p on the runner's subject for it; step, when not nil, is the result a step-ended carries. +func (h *Hearer) say(event string, p seat.Pipeline, step *seat.StepResult) error { + last := p.Commit + if p.Group != "" { + last = p.Group + } + subject := "mesh.seat." + seat.Seat + ".event." + event + "." + keys.EscapeName(p.Module) + "." + keys.EscapeName(p.Branch) + "." + keys.EscapeName(last) + at := h.now() + switch event { + case seat.EventStepEnded: + return h.Say(subject, seat.StepEvent{V: 1, At: &at, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit, + Fingerprint: p.Fingerprint, Group: p.Group, Revision: p.Revision, Source: p.Source, Result: *step}) + case seat.EventPipelineEnded: + return h.Say(subject, pipelineEnded(p)) + default: + return h.Say(subject, seat.Wait{V: 1, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit, Fingerprint: p.Fingerprint, + Group: p.Group, Revision: p.Revision, Source: p.Source, Step: seat.StepResultStepApproval, On: "operator", Since: at}) + } +} + +func pipelineEnded(p seat.Pipeline) seat.PipelineEnded { + return seat.PipelineEnded{V: 1, Pipeline: p.Pipeline, Branch: p.Branch, Module: p.Module, Commit: p.Commit, Fingerprint: p.Fingerprint, + Group: p.Group, Revision: p.Revision, Source: p.Source, Run: p.Run, Outcome: p.Outcome, Why: p.Why, Started: p.Started, + Ended: p.Ended, EndCount: p.EndCount, Steps: p.Steps} +} + +func endOutcome(o string) bool { + switch o { + case nodebuildgo.StepEndedOutcomePassed, nodebuildgo.StepEndedOutcomeFailed, nodebuildgo.StepEndedOutcomeNothingToRun, + nodebuildgo.StepEndedOutcomeCouldNotRun: + return true + } + return false +} + +func stepOfAsk(p seat.Pipeline, id string) int { + for i, s := range p.Steps { + if id != "" && s.Ask == id { + return i + } + } + return -1 +} + +// endPipeline ends a pipeline whose every step ended: passed when every required step passed or was carried, else +// failed, naming each required step that held it (ADR 0666 §7, §9; design 47 G §24, §26). A failed optional step holds +// nothing. An approval step still waiting when a required step did not pass is `not asked` (§26). It says whether the +// pipeline ended now, its end count one higher. +func endPipeline(p *seat.Pipeline, now time.Time) bool { + if p.Outcome != seat.PipelineOutcomeRunning { + return false + } + var held []string + notAsked := -1 + for i, s := range p.Steps { + switch { + case s.Step == seat.StepResultStepApproval && s.Outcome == seat.StepResultOutcomeWaiting: + notAsked = i + continue + case s.Outcome == seat.StepResultOutcomeWaiting, s.Outcome == seat.StepResultOutcomeRunning: + return false + } + if s.Status == seat.StepResultStatusRequired && s.Outcome != seat.StepResultOutcomePassed && s.Outcome != seat.StepResultOutcomeCarried { + held = append(held, s.Artifact+" "+s.Step+" "+s.Outcome) + } + } + if notAsked >= 0 { + if len(held) == 0 { + return false + } + p.Steps[notAsked].Outcome = seat.StepResultOutcomeNotAsked + } + p.Ended = &now + p.EndCount++ + p.Outcome, p.Why = seat.PipelineOutcomePassed, "" + if len(held) > 0 { + p.Outcome, p.Why = seat.PipelineOutcomeFailed, "required steps not passed: "+strings.Join(held, ", ") + } + return true +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go new file mode 100644 index 00000000..3f6c18bc --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go @@ -0,0 +1,278 @@ +package pipeline + +import ( + "errors" + "strings" + "testing" + "time" + + "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" + nodebuildgo "git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +type memStore map[string]struct { + p seat.Pipeline + rev uint64 +} + +func (m memStore) Get(key string) (seat.Pipeline, uint64, error) { + e, ok := m[key] + if !ok { + return seat.Pipeline{}, 0, errors.New("no such key") + } + return e.p, e.rev, nil +} + +func (m memStore) Update(key string, p seat.Pipeline, rev uint64) (uint64, error) { + if m[key].rev != rev { + return 0, ErrChanged + } + m[key] = struct { + p seat.Pipeline + rev uint64 + }{p, rev + 1} + return rev + 1, nil +} + +func (m memStore) Keys(filter string) ([]string, error) { + var out []string + for k := range m { + if ok, _ := matchKey(filter, k); ok { + out = append(out, k) + } + } + return out, nil +} + +func matchKey(filter, key string) (bool, error) { + f, k := strings.Split(filter, "."), strings.Split(key, ".") + if len(f) != len(k) { + return false, nil + } + for i := range f { + if f[i] != "*" && f[i] != k[i] { + return false, nil + } + } + return true, nil +} + +type askRecorder struct{ asked []oc.Ask } + +func (a *askRecorder) Ask(x oc.Ask) error { a.asked = append(a.asked, x); return nil } + +const pkey = "main.5.app" + +// fixture is a pipeline of app with a required lint asked of node-build-go, an optional unit asked of the same seat and an +// approval step waiting, and a hearer over it. +func fixture() (*Hearer, memStore, *[]string, *askRecorder, map[string]seat.KeptResult, map[string]string) { + store := memStore{pkey: {p: seat.Pipeline{V: 1, Pipeline: pkey, Branch: "main", Source: 5, Module: "app", Commit: "c1", Fingerprint: "src2:aa", + Outcome: seat.PipelineOutcomeRunning, Steps: []seat.StepResult{ + {Artifact: "program", Step: seat.StepResultStepLint, Status: seat.StepResultStatusRequired, Outcome: seat.StepResultOutcomeRunning, Ask: "a1", Seat: nodebuildgo.Seat}, + {Artifact: "program", Step: seat.StepResultStepUnit, Status: seat.StepResultStatusOptional, Outcome: seat.StepResultOutcomeRunning, Ask: "a2", Seat: nodebuildgo.Seat}, + {Artifact: "program", Step: seat.StepResultStepApproval, Status: seat.StepResultStatusRequired, Outcome: seat.StepResultOutcomeWaiting}, + }}, rev: 1}} + said := &[]string{} + asker := &askRecorder{} + kept := map[string]seat.KeptResult{} + outs := map[string]string{} + h := &Hearer{Store: store, Name: "mesh-pipeline-runner", Asker: asker, + Say: func(subject string, _ any) error { + *said = append(*said, subject[strings.Index(subject, ".event.")+7:]) + return nil + }, + Log: func(seat.LogEntry) error { return nil }, + Keep: func(k string, v seat.KeptResult) error { kept[k] = v; return nil }, + Output: func(k string, d []byte) error { outs[k] = string(d); return nil }, + Now: func() time.Time { return time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC) }} + return h, store, said, asker, kept, outs +} + +func ended(ask, step, outcome string) StepEnded { + return StepEnded{ID: ask, Ask: ask, Node: "n1", Pipeline: pkey, Module: "app", Commit: "c1", Fingerprint: "src2:aa", Step: step, Outcome: outcome} +} + +var goSeat = StepSeat{Seat: nodebuildgo.Seat, Prefix: "mesh.seat.node-build-go.event.step-ended."} + +// A step's end is taken only as the runner asked it (design 47 G §24; ADR 0666 §21). +func TestStepEndedIsTakenOnlyAsAsked(t *testing.T) { + h, store, _, _, _, _ := fixture() + other := ended("a1", "lint", "passed") + other.Fingerprint = "src2:bb" + wrongNode := ended("a1", "lint", "passed") + wrongNode.Node = "n2" + unknownPipeline := ended("a1", "lint", "passed") + unknownPipeline.Pipeline = "main.9.app" + for name, c := range map[string]struct { + subject string + ev StepEnded + }{ + "a subject of another seat": {"mesh.seat.node-build-node.event.step-ended.n1", ended("a1", "lint", "passed")}, + "a body naming another node": {goSeat.Prefix + "n1", wrongNode}, + "an unknown pipeline": {goSeat.Prefix + "n1", unknownPipeline}, + "an ask nobody made": {goSeat.Prefix + "n1", ended("zz", "lint", "passed")}, + "another fingerprint": {goSeat.Prefix + "n1", other}, + "another step": {goSeat.Prefix + "n1", ended("a1", "unit", "passed")}, + "an outcome step-ended lacks": {goSeat.Prefix + "n1", ended("a1", "lint", "waiting")}, + } { + if _, err := h.OnStepEnded(goSeat, c.subject, c.ev); !errors.Is(err, ErrRefused) { + t.Errorf("%s: err %v, want a refusal", name, err) + } + } + if store[pkey].p.Steps[0].Outcome != seat.StepResultOutcomeRunning { + t.Error("a refused end changed the pipeline") + } + // The seat the step was asked of is the only one that ends it. + if _, err := h.OnStepEnded(StepSeat{Seat: "node-build-node", Prefix: "mesh.seat.node-build-node.event.step-ended."}, "mesh.seat.node-build-node.event.step-ended.n1", ended("a1", "lint", "passed")); !errors.Is(err, ErrRefused) { + t.Errorf("another seat's holder ended the step: %v", err) + } +} + +// A taken end is written, kept and said; the pipeline ends once every step ended (§24, §27). +func TestStepEndedEndsThePipeline(t *testing.T) { + h, store, said, _, kept, outs := fixture() + h.Asker = nil + lint := ended("a1", "lint", "passed") + lint.Output = "all clean" + unit := ended("a2", "unit", "failed") + unit.Kept = []nodebuildgo.KeptResult{{Digest: "sha256:d1", Level: "unit", Package: "app/x"}} + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", lint); err != nil { + t.Fatal(err) + } + if p := store[pkey].p; p.Outcome != seat.PipelineOutcomeRunning || p.Steps[0].Node != "n1" || p.Steps[0].Output == "" { + t.Fatalf("after the lint: %+v", p) + } + if len(outs) != 1 { + t.Errorf("outputs kept %v", outs) + } + // Approval waits with no asker, so the pipeline does not end on the last test step alone. + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", unit); err != nil { + t.Fatal(err) + } + if p := store[pkey].p; p.Outcome != seat.PipelineOutcomeRunning { + t.Fatalf("a pipeline with its approval waiting ended: %+v", p) + } + if k := kept[seat.KeptByLevelDigestKey("unit", keys.EscapeName("sha256:d1"))]; k.Package != "app/x" || k.Pipeline != pkey { + t.Errorf("kept %+v", kept) + } + // A step heard again takes nothing. + n := len(*said) + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", unit); err != nil || len(*said) != n { + t.Errorf("a second end said %v (%v)", (*said)[n:], err) + } +} + +// The approval step is asked last, only when every other required step passed, and ended by its warrant (§26, §37). +func TestApprovalIsAskedLastAndEndedByItsWarrant(t *testing.T) { + h, store, said, asker, _, _ := fixture() + var record *oc.Warrant + h.Where = func(string) (string, *oc.Warrant, error) { return "chosen", record, nil } + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a1", "lint", "passed")); err != nil || len(asker.asked) != 0 { + t.Fatalf("asked %d before the unit step ended (%v)", len(asker.asked), err) + } + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a2", "unit", "failed")); err != nil { + t.Fatal(err) + } + // An optional step that failed holds nothing: the approval is asked. + if len(asker.asked) != 1 || asker.asked[0].About != "app.main" { + t.Fatalf("asked %+v", asker.asked) + } + a := asker.asked[0] + if got := (*said)[len(*said)-1]; !strings.HasPrefix(got, "wait-held.app.main.c1") { + t.Errorf("the wait was said as %q", got) + } + warrant := func(option, outcome string) oc.Warrant { + return oc.Warrant{Asker: "mesh-pipeline-runner", Ask: a.ID, About: a.About, Outcome: outcome, Option: option, Level: operatorchannelclient.Approve, + AskDigest: operatorchannelclient.Digest(a), At: operatorchannelclient.FormatTime(time.Date(2026, 11, 11, 13, 0, 0, 0, time.UTC)), + By: &oc.Person{Who: operatorchannelclient.Operator, Identity: "op", Kind: "telegram", Verified: "linked"}} + } + // The event only wakes the runner (A-89): an event that says approve while the router's record holds none takes nothing. + event := warrant(OptionApprove, oc.WarrantOutcomeChosen) + if _, err := h.OnDecided(event); !errors.Is(err, ErrNoRecord) || store[pkey].p.Steps[2].Outcome != seat.StepResultOutcomeRunning { + t.Errorf("an event with no record was taken: %v", err) + } + // A record whose warrant is for another ask's digest is not taken; an event of another asker neither. + bad := warrant(OptionApprove, oc.WarrantOutcomeChosen) + bad.AskDigest = "sha256:other" + record = &bad + if _, err := h.OnDecided(event); err == nil || store[pkey].p.Steps[2].Outcome != seat.StepResultOutcomeRunning { + t.Errorf("a warrant for another ask was taken: %v", err) + } + notMine := event + notMine.Asker = "someone" + if _, err := h.OnDecided(notMine); !errors.Is(err, ErrRefused) { + t.Errorf("another asker's event: %v", err) + } + // The record says decline although the event body says approve: the record decides. + decline := warrant(OptionDecline, oc.WarrantOutcomeChosen) + record = &decline + if p, err := h.OnDecided(event); err != nil || p.Steps[2].Outcome != seat.StepResultOutcomeDeclined { + t.Fatalf("the record said decline: %+v (%v)", p.Steps[2], err) + } + // (the declined pipeline above is a fresh fixture's: approve it on a new one) + h, store, said, asker, _, _ = fixture() + h.Where = func(string) (string, *oc.Warrant, error) { return "chosen", record, nil } + for _, s := range []StepEnded{ended("a1", "lint", "passed"), ended("a2", "unit", "failed")} { + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", s); err != nil { + t.Fatal(err) + } + } + a = asker.asked[0] + approve := warrant(OptionApprove, oc.WarrantOutcomeChosen) + record = &approve + p, err := h.OnDecided(event) + if err != nil || p.Steps[2].Outcome != seat.StepResultOutcomePassed || p.Outcome != seat.PipelineOutcomePassed || p.EndCount != 1 { + t.Fatalf("approved: %+v (%v)", p, err) + } + if !strings.HasPrefix((*said)[len(*said)-1], "pipeline-ended.app.main.c1") { + t.Errorf("said %v", *said) + } + // A pipeline that ended takes no second decision. + record = &decline + if p, err := h.OnDecided(event); err != nil || p.Steps[2].Outcome != seat.StepResultOutcomePassed { + t.Errorf("a second warrant changed it: %+v (%v)", p.Steps[2], err) + } +} + +// A decline, an expiry or a replacement ends the approval step declined or superseded, and a failed required step leaves it +// not asked with the pipeline failed (§26, §37). +func TestApprovalEnds(t *testing.T) { + for name, c := range map[string]struct { + outcome, option, want string + }{ + "declined": {oc.WarrantOutcomeChosen, OptionDecline, seat.StepResultOutcomeDeclined}, + "expired": {oc.WarrantOutcomeExpired, "", seat.StepResultOutcomeDeclined}, + "refused": {oc.WarrantOutcomeRefused, "", seat.StepResultOutcomeDeclined}, + "superseded": {oc.WarrantOutcomeReplaced, "", seat.StepResultOutcomeSuperseded}, + } { + h, _, _, asker, _, _ := fixture() + for _, s := range []StepEnded{ended("a1", "lint", "passed"), ended("a2", "unit", "passed")} { + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", s); err != nil { + t.Fatal(err) + } + } + a := asker.asked[0] + rec := oc.Warrant{Asker: h.Name, Ask: a.ID, About: a.About, Outcome: c.outcome, Option: c.option, Level: operatorchannelclient.Approve, + AskDigest: operatorchannelclient.Digest(a), At: operatorchannelclient.FormatTime(time.Date(2026, 11, 11, 13, 0, 0, 0, time.UTC)), + By: &oc.Person{Who: operatorchannelclient.Operator, Identity: "op", Kind: "telegram", Verified: "linked"}} + h.Where = func(string) (string, *oc.Warrant, error) { return c.outcome, &rec, nil } + p, err := h.OnDecided(oc.Warrant{Asker: h.Name, Ask: a.ID, About: a.About}) + if err != nil || p.Steps[2].Outcome != c.want || p.Outcome != seat.PipelineOutcomeFailed { + t.Errorf("%s: step %q, pipeline %q (%v)", name, p.Steps[2].Outcome, p.Outcome, err) + } + } + h, store, _, asker, _, _ := fixture() + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a1", "lint", "failed")); err != nil { + t.Fatal(err) + } + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a2", "unit", "passed")); err != nil { + t.Fatal(err) + } + p := store[pkey].p + if len(asker.asked) != 0 || p.Steps[2].Outcome != seat.StepResultOutcomeNotAsked || p.Outcome != seat.PipelineOutcomeFailed || !strings.Contains(p.Why, "lint") { + t.Errorf("asked %d; %+v", len(asker.asked), p) + } +}