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 59c6b4c9..23a357aa 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -136,6 +136,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hub.StartUntilBound(ctx, js, cred, nil, nil); err != nil { logger.Printf("the consumer %s is not bound yet (%v): bound once the bus makes it", cred.EventConsumerName(), err) } + // The seat's write verbs are answered here, by the process that holds the pipelines (A-107). + stopVerbs, err := answerVerbs(calls, writeVerbs(ctx, hearer, starter, func(branch string) (pipeline.Half, bool) { return rv.Catalog.View(branch) }, calls, bounds)) + if err != nil { + return err + } + defer stopVerbs() go func() { <-ctx.Done(); hub.Stop() }() logger.Printf("holding %s", seat.Seat) <-ctx.Done() diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go new file mode 100644 index 00000000..b082fef0 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go @@ -0,0 +1,60 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/process" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// verbOf makes the handler of one verb from its function of the proved caller and the decoded input: the caller is the bus +// user the call's subject names, and a call naming none is refused (design 47 G §394, A-107). +func verbOf[In any](do func(caller string, in In) (any, error)) process.SeatHandler { + return func(caller string, body json.RawMessage) (any, error) { + if caller == "" { + return nil, errors.New("the call names no caller the bus proved: it is refused") + } + var in In + if len(body) > 0 { + if err := json.Unmarshal(body, &in); err != nil { + return nil, fmt.Errorf("the arguments do not fit the verb's input: %w", err) + } + } + return do(caller, in) + } +} + +// answerVerbs answers the seat's write verbs `review`, `rerun`, `accept` and `cancel` over the bus, each writing the state +// pipelines by compare-and-set (design 47 G §22, §70, §482, A-107). The returned stop ends the answers. +func answerVerbs(conn *process.Conn, v pipeline.Verbs) (func(), error) { + return conn.AnswerSeat(seat.Seat, map[string]process.SeatHandler{ + seat.VerbReview: verbOf(func(c string, in seat.ReviewInput) (any, error) { return v.Review(c, in) }), + seat.VerbRerun: verbOf(func(c string, in seat.RerunInput) (any, error) { return v.Rerun(c, in) }), + seat.VerbAccept: verbOf(func(c string, in seat.AcceptInput) (any, error) { return v.Accept(c, in) }), + seat.VerbCancel: verbOf(func(c string, in seat.CancelInput) (any, error) { return v.Cancel(c, in) }), + }) +} + +// dropBound deletes the bound of an ask a cancelled pipeline no longer waits on. +func dropBound(ctx context.Context, bounds jetstream.KeyValue) func(string) error { + return func(ask string) error { + if err := bounds.Delete(ctx, ask); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) { + return err + } + return nil + } +} + +// writeVerbs are the verbs over the hearer and the starter, their hand acts recorded through the seat mesh-hand-acts. +func writeVerbs(ctx context.Context, h *pipeline.Hearer, s *pipeline.Starter, view func(string) (pipeline.Half, bool), + calls *process.Conn, bounds jetstream.KeyValue) pipeline.Verbs { + return pipeline.Verbs{Hearer: h, Starter: s, View: view, Acts: meshhandactsclient.Recorder{Ask: calls.Call}, Drop: dropBound(ctx, bounds)} +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go new file mode 100644 index 00000000..5f85c5f2 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go @@ -0,0 +1,24 @@ +package main + +import ( + "encoding/json" + "testing" + + seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner" +) + +// A write verb is answered only for a caller the bus proved, and its input is the verb's own (design 47 G §394, A-107). +func TestAVerbIsAnsweredOnlyForAProvedCallerAndItsOwnInput(t *testing.T) { + var got seat.CancelInput + h := verbOf(func(c string, in seat.CancelInput) (any, error) { got = in; return c, nil }) + if _, err := h("", json.RawMessage(`{"pipeline":"p","why":"w"}`)); err == nil { + t.Error("a call naming no proved caller was answered") + } + if _, err := h("operator", json.RawMessage(`{"pipeline":`)); err == nil { + t.Error("a malformed input was answered") + } + out, err := h("operator", json.RawMessage(`{"pipeline":"p","why":"w"}`)) + if err != nil || out != "operator" || got.Pipeline != "p" || got.Why != "w" { + t.Errorf("answer = %v, %v, input %+v", out, err, got) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go index e239c575..21063641 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go @@ -236,7 +236,7 @@ func (h *Hearer) say(event string, p seat.Pipeline, step *seat.StepResult) error 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: + case seat.EventStepEnded, seat.EventStepStarted: 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: diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go index a1e73cc7..fec31c8c 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go @@ -24,6 +24,8 @@ func (m memStore) Get(key string) (seat.Pipeline, uint64, error) { if !ok { return seat.Pipeline{}, 0, errors.New("no such key") } + // A read is a copy, as one from the bus is: a change not written is not seen by the next read. + e.p.Steps = append([]seat.StepResult(nil), e.p.Steps...) return e.p, e.rev, nil } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go new file mode 100644 index 00000000..e664c149 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go @@ -0,0 +1,336 @@ +package pipeline + +import ( + "errors" + "fmt" + "strings" + "time" + + builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds" + 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/views" +) + +// Verbs are the runner's write verbs `review`, `rerun`, `accept` and `cancel`, answered by its own process (design 47 G §22, +// §70, §482, A-107): each writes the state pipelines by compare-and-set on the revision read and says its event as the +// hearer does for a step's end. The caller each takes is the one the bus proved. +type Verbs struct { + Hearer *Hearer + // Starter asks a step again (rerun), as it asks the steps of a start; View reads a branch's catalog half for the ask. + Starter *Starter + View func(branch string) (Half, bool) + // Acts records a hand act before the verb does it (rerun, accept, cancel); a nil Acts refuses the verb, nothing recorded. + Acts views.HandActs + // Drop drops the bound of an ask a cancelled pipeline no longer waits on. + Drop func(ask string) error +} + +func (v Verbs) now() time.Time { return v.Hearer.now() } + +func (v Verbs) record(verb, what, why string) error { + if v.Acts == nil { + return errors.New("no way to record a hand act is given: nothing was done") + } + if strings.TrimSpace(why) == "" { + return fmt.Errorf("%s says why: nothing was done", verb) + } + if err := v.Acts.RecordHandAct(views.HandAct{What: what, Why: why, Verb: seat.Seat + "." + verb}); err != nil { + return fmt.Errorf("the hand act is not recorded, so nothing was done: %w", err) + } + return nil +} + +// change writes one change of a pipeline by compare-and-set, read again when another write came between, and says what it +// became: the step-ended of the step mutate changed (its index, -1 for none), the wait of an approval asked and the +// pipeline-ended of a pipeline it ended. It returns the written pipeline and the changed step. +func (v Verbs) change(key, event string, record func() error, mutate func(p *seat.Pipeline, now time.Time) (int, error)) (seat.Pipeline, int, error) { + h := v.Hearer + recorded := false + for { + p, rev, err := h.Store.Get(key) + if err != nil { + return p, -1, fmt.Errorf("the pipeline %s is not read: %w", key, err) + } + now := h.now() + i, err := mutate(&p, now) + if err != nil { + return p, -1, err + } + // The hand act is recorded once the verb is valid and before it writes (ADR 0565 clause 11); a refused verb records none. + if record != nil && !recorded { + if err := record(); err != nil { + return p, -1, err + } + recorded = true + } + asked, err := askApproval(key, &p, now, h.Asker) + if err != nil { + return p, i, err + } + ended := endPipeline(&p, now) + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return p, i, fmt.Errorf("pipeline %s cannot be written: %w", key, err) + } + if i >= 0 { + if event == seat.EventStepEnded { + h.logEnd(key, p.Steps[i], now) + } + if err := h.say(event, p, &p.Steps[i]); err != nil { + return p, i, fmt.Errorf("pipeline %s's %s is written but not said: %w", key, event, err) + } + } + return p, i, h.said(p, asked, ended) + } +} + +// Review takes a code review's result for a commit and fingerprint from caller, the bus's proof, refusing none for +// authorship (§482): it ends the placed review step of each running pipeline of the module at that commit and fingerprint, +// recording caller in the step's result. It answers the first such pipeline; a review no pipeline waits on is refused. +func (v Verbs) Review(caller string, in seat.ReviewInput) (seat.ReviewAnswer, error) { + if in.Outcome != seat.ReviewInputOutcomePassed && in.Outcome != seat.ReviewInputOutcomeFailed { + return seat.ReviewAnswer{}, fmt.Errorf("a review ends %q or %q, not %q", seat.ReviewInputOutcomePassed, seat.ReviewInputOutcomeFailed, in.Outcome) + } + keys, err := v.Hearer.Store.Keys("*.*.*") + if err != nil { + return seat.ReviewAnswer{}, fmt.Errorf("the pipelines are not listed: %w", err) + } + var answer *seat.ReviewAnswer + for _, key := range keys { + p, _, err := v.Hearer.Store.Get(key) + if err != nil || p.Module != in.Module || p.Commit != in.Commit || p.Fingerprint != in.Fingerprint || reviewWaiting(p) < 0 { + continue + } + written, i, err := v.change(key, seat.EventStepEnded, nil, func(p *seat.Pipeline, now time.Time) (int, error) { + i := reviewWaiting(*p) + if i < 0 { + return -1, fmt.Errorf("%s has no review step waiting any more", key) + } + s := &p.Steps[i] + s.Outcome, s.By, s.Why, s.Ended = in.Outcome, caller, in.Said, &now + return i, nil + }) + if err != nil { + return seat.ReviewAnswer{}, err + } + if answer == nil { + answer = &seat.ReviewAnswer{Pipeline: key, Result: written.Steps[i]} + } + } + if answer == nil { + return seat.ReviewAnswer{}, fmt.Errorf("no running pipeline of %s at %s (%s) waits on a review", in.Module, in.Commit, in.Fingerprint) + } + return *answer, nil +} + +func reviewWaiting(p seat.Pipeline) int { + if p.Outcome != seat.PipelineOutcomeRunning { + return -1 + } + for i, s := range p.Steps { + if s.Step == seat.StepResultStepReview && s.Outcome == seat.StepResultOutcomeWaiting { + return i + } + } + return -1 +} + +// stepOf is the index of the step named, and of its artifact when one is named. +func stepOf(p seat.Pipeline, step, artifact string) (int, error) { + found := -1 + for i, s := range p.Steps { + if s.Step != step || (artifact != "" && s.Artifact != artifact) { + continue + } + if found >= 0 { + return -1, fmt.Errorf("%s names more than one %s step: name its artifact", p.Pipeline, step) + } + found = i + } + if found < 0 { + return -1, fmt.Errorf("%s has no %s step", p.Pipeline, step) + } + return found, nil +} + +// Accept accepts a failed `optional` step, recorded in the hand-act log first with who and why (§30, §394). +func (v Verbs) Accept(caller string, in seat.AcceptInput) (seat.AcceptAnswer, error) { + p, i, err := v.change(in.Pipeline, seat.EventStepEnded, func() error { + return v.record(seat.VerbAccept, fmt.Sprintf("accepted the failed %s step of %s", in.Step, in.Pipeline), in.Why) + }, func(p *seat.Pipeline, now time.Time) (int, error) { + i, err := stepOf(*p, in.Step, in.Artifact) + if err != nil { + return -1, err + } + s := &p.Steps[i] + if s.Status != seat.StepResultStatusOptional || s.Outcome != seat.StepResultOutcomeFailed { + return -1, fmt.Errorf("only a failed optional step is accepted: the %s step of %s is %s and %s", in.Step, in.Pipeline, s.Status, s.Outcome) + } + s.Outcome, s.Accepted, s.Ended = seat.StepResultOutcomeAccepted, &seat.Accepted{At: &now, By: caller, Why: in.Why}, &now + return i, nil + }) + if err != nil { + return seat.AcceptAnswer{}, err + } + return seat.AcceptAnswer{Result: p.Steps[i]}, nil +} + +// Cancel ends a running pipeline `cancelled`, each step not ended cancelled and the bound of each ask dropped, recorded in +// the hand-act log first (§394); it says pipeline-ended. +func (v Verbs) Cancel(caller string, in seat.CancelInput) (seat.CancelAnswer, error) { + h := v.Hearer + recorded := false + for { + p, rev, err := h.Store.Get(in.Pipeline) + if err != nil { + return seat.CancelAnswer{}, fmt.Errorf("the pipeline %s is not read: %w", in.Pipeline, err) + } + if p.Outcome != seat.PipelineOutcomeRunning { + return seat.CancelAnswer{}, fmt.Errorf("the pipeline %s is %s, not running", in.Pipeline, p.Outcome) + } + if !recorded { + if err := v.record(seat.VerbCancel, "cancelled the pipeline "+in.Pipeline, in.Why); err != nil { + return seat.CancelAnswer{}, err + } + recorded = true + } + now := h.now() + var drop []string + for i := range p.Steps { + s := &p.Steps[i] + if s.Outcome != seat.StepResultOutcomeRunning && s.Outcome != seat.StepResultOutcomeWaiting { + continue + } + if s.Ask != "" { + drop = append(drop, s.Ask) + } + s.Outcome, s.By, s.Why, s.Ended = seat.StepResultOutcomeCancelled, caller, in.Why, &now + } + p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeCancelled, in.Why, &now + p.EndCount++ + if _, err := h.Store.Update(in.Pipeline, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + return seat.CancelAnswer{}, fmt.Errorf("pipeline %s cannot be written: %w", in.Pipeline, err) + } + if v.Drop != nil { + for _, ask := range drop { + if err := v.Drop(ask); err != nil { + h.logf("the bound of ask %s of the cancelled %s is not dropped: %v", ask, in.Pipeline, err) + } + } + } + return seat.CancelAnswer{Pipeline: p}, h.said(p, false, true) + } +} + +// Rerun asks one step again, recorded in the hand-act log first (§22, §39): the step goes back to running under a new ask, +// the ask saying it is a rerun and the caller the bus proved; a failed pipeline runs again and ends once more with its end +// count one higher (§325), an approval never asked waiting again. A review waits again; an approval is asked again by +// `ask-again`. +func (v Verbs) Rerun(caller string, in seat.RerunInput) (seat.RerunAnswer, error) { + if in.Step == seat.StepResultStepApproval { + return seat.RerunAnswer{}, errors.New("an approval is asked again by ask-again, not rerun") + } + var ask *Asked + p, _, err := v.change(in.Pipeline, seat.EventStepStarted, func() error { + return v.record(seat.VerbRerun, fmt.Sprintf("asked the %s step of %s again", in.Step, in.Pipeline), in.Why) + }, func(p *seat.Pipeline, now time.Time) (int, error) { + ask = nil + switch p.Outcome { + case seat.PipelineOutcomeCancelled, seat.PipelineOutcomeSuperseded: + return -1, fmt.Errorf("the pipeline %s is %s: a step of it is not asked again", p.Pipeline, p.Outcome) + } + i, err := stepOf(*p, in.Step, in.Artifact) + if err != nil { + return -1, err + } + s := &p.Steps[i] + if s.Outcome == seat.StepResultOutcomeRunning { + return -1, fmt.Errorf("the %s step of %s is running: it is not asked again", in.Step, p.Pipeline) + } + if s.Seat != "" { + if ask, err = v.askOf(*p, *s, caller, now); err != nil { + return -1, err + } + } + *s = seat.StepResult{Step: s.Step, Artifact: s.Artifact, Status: s.Status, Fingerprint: s.Fingerprint, Seat: s.Seat, + Outcome: seat.StepResultOutcomeWaiting, By: caller, Rerun: true} + if ask != nil { + s.Outcome, s.Started, s.Ask = seat.StepResultOutcomeRunning, &now, ask.ID + } + // The pipeline runs again; an approval never asked waits again, to be asked when the step passes. + if p.Outcome != seat.PipelineOutcomeRunning { + p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeRunning, "", nil + } + for j := range p.Steps { + if p.Steps[j].Step == seat.StepResultStepApproval && p.Steps[j].Outcome == seat.StepResultOutcomeNotAsked { + p.Steps[j].Outcome, p.Steps[j].Why = seat.StepResultOutcomeWaiting, "" + } + } + if ask == nil { + return -1, nil + } + return i, nil + }) + if err != nil { + return seat.RerunAnswer{}, err + } + 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 { + return seat.RerunAnswer{}, fmt.Errorf("the step is written running but not asked, it is asked again at the next start: %w", err) + } + } + res, err := stepOf(p, in.Step, in.Artifact) + if err != nil { + return seat.RerunAnswer{}, err + } + r := p.Steps[res] + return seat.RerunAnswer{Pipeline: in.Pipeline, Result: &r}, nil +} + +// askOf is the ask to put on the serving seat for step s again: the one Place makes for it, under a new id (an ask put twice +// under one id is kept once), saying it is a rerun and by whom. +func (v Verbs) askOf(p seat.Pipeline, s seat.StepResult, caller string, now time.Time) (*Asked, error) { + if v.Starter == nil || v.Starter.Registration == nil || v.Starter.Steps == nil || v.View == nil { + return nil, errors.New("the runner cannot ask a step: it holds no registration or no ask") + } + reg, ok := v.Starter.Registration(p.Module) + if !ok { + return nil, fmt.Errorf("%s has no registration", p.Module) + } + half, ok := v.View(p.Branch) + if !ok { + return nil, fmt.Errorf("the catalog half of %s is not held yet: the step is asked when it is", p.Branch) + } + _, placed := Place(p, reg, half.Entries, now, v.Starter.Name) + base := AskID(p.Pipeline, s.Artifact, s.Step) + for _, a := range placed { + if a.ID != base { + continue + } + a.ID = AskID(p.Pipeline, s.Artifact, s.Step+"\nrerun\n"+now.Format(time.RFC3339Nano)) + switch b := a.Ask.(type) { + case nodebuildgo.StepAsk: + b.ID = a.ID + if b.Step != nil { + step := *b.Step + step.Rerun, step.By = true, caller + b.Step = &step + } + a.Ask = b + case builds.StepRun: + b.Rerun, b.By = true, caller + a.Ask = b + case wants.StepRun: + b.Rerun, b.By = true, caller + a.Ask = b + } + return &a, nil + } + return nil, fmt.Errorf("the %s step of %s is placed no ask now", s.Step, p.Pipeline) +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go new file mode 100644 index 00000000..77f7ea31 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go @@ -0,0 +1,172 @@ +package pipeline + +import ( + "errors" + "strings" + "testing" + "time" + + catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + 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/views" +) + +type actLog []string + +func (a *actLog) RecordHandAct(h views.HandAct) error { + *a = append(*a, h.Verb+": "+h.What+" / "+h.Why) + return nil +} + +// verbsFixture is a running pipeline of app placed from a policy with a required lint, an optional unit, an optional review +// and a required approval, in a store, with the verbs over it. +func verbsFixture() (Verbs, memStore, *[]string, *stepAsks, *actLog, *[]string) { + reg, entries := placeFixture() + reg.Policy.Steps.Review = "optional" + h, store, said, _, _, _ := fixture() + p := seat.Pipeline{Pipeline: pkey, Branch: "main", Source: 5, Module: "app", Commit: "c1", Fingerprint: "src2:aa", Repository: "o/app", + Outcome: seat.PipelineOutcomeRunning} + p.Steps, _ = Place(p, reg, entries, time.Unix(1, 0), "the-runner") + store[pkey] = struct { + p seat.Pipeline + rev uint64 + }{p, 1} + asks, acts, dropped := &stepAsks{}, &actLog{}, &[]string{} + s := &Starter{Steps: asks, Name: "the-runner", Registration: func(string) (catalog.Registration, bool) { return reg, true }, + Now: h.Now} + return Verbs{Hearer: h, Starter: s, View: func(string) (Half, bool) { return Half{Entries: entries}, true }, Acts: acts, + Drop: func(ask string) error { *dropped = append(*dropped, ask); return nil }}, store, said, asks, acts, dropped +} + +func stepNamed(p seat.Pipeline, artifact, step string) seat.StepResult { + for _, s := range p.Steps { + if s.Artifact == artifact && s.Step == step { + return s + } + } + return seat.StepResult{} +} + +// A code review ends the placed review step of the pipeline at its commit and fingerprint, whoever its author, recording the +// caller the bus proved in the step's result, its step-ended and the pipeline's show; a review no pipeline waits on is refused +// (design 47 G §70, §482). +func TestReviewEndsThePlacedReviewStepWithItsProvenCaller(t *testing.T) { + v, store, said, _, _, _ := verbsFixture() + in := seat.ReviewInput{Module: "app", Commit: "c1", Fingerprint: "src2:aa", Outcome: "passed", Said: "reads well"} + for name, bad := range map[string]seat.ReviewInput{ + "another commit": {Module: "app", Commit: "c2", Fingerprint: "src2:aa", Outcome: "passed"}, + "another fingerprint": {Module: "app", Commit: "c1", Fingerprint: "src2:bb", Outcome: "passed"}, + "an outcome review lacks": {Module: "app", Commit: "c1", Fingerprint: "src2:aa", Outcome: "waiting"}, + } { + if _, err := v.Review("agent.dev", bad); err == nil { + t.Errorf("%s: a review was taken", name) + } + } + a, err := v.Review("agent.dev", in) + if err != nil || a.Pipeline != pkey || a.Result.Outcome != "passed" || a.Result.By != "agent.dev" { + t.Fatalf("review = %+v, %v", a, err) + } + got := stepNamed(store[pkey].p, "", seat.StepResultStepReview) + if got.Outcome != "passed" || got.By != "agent.dev" || got.Why != "reads well" || got.Ended == nil { + t.Errorf("the review step = %+v", got) + } + if len(*said) != 1 || !strings.HasPrefix((*said)[0], "step-ended.") { + t.Errorf("said %v, want one step-ended", *said) + } + if _, err := v.Review("agent.dev", in); err == nil { + t.Error("a second review of an ended review step was taken") + } +} + +// accept takes only a failed optional step, writes its hand act first, and refuses without one; rerun asks the step again +// under a new ask saying it is a rerun and its caller, reopens a failed pipeline and asks nothing of a refused one; cancel ends +// a running pipeline cancelled, each step cancelled and each ask's bound dropped (design 47 G §22, §30, §325, §394). +func TestAcceptRerunAndCancelWriteByCompareAndSetAndRecordTheirHandAct(t *testing.T) { + v, store, said, asks, acts, dropped := verbsFixture() + end := func(step, outcome string) { + p, rev, _ := store.Get(pkey) + for i := range p.Steps { + if p.Steps[i].Artifact == "app:program" && p.Steps[i].Step == step { + p.Steps[i].Outcome = outcome + } + } + store.Update(pkey, p, rev) + } + end("unit", seat.StepResultOutcomeFailed) + if _, err := v.Accept("operator", seat.AcceptInput{Pipeline: pkey, Step: "lint", Artifact: "app:program", Why: "x"}); err == nil { + t.Error("a running required step was accepted") + } + if _, err := v.Accept("operator", seat.AcceptInput{Pipeline: pkey, Step: "unit", Artifact: "app:program"}); err == nil { + t.Error("an accept with no why was taken") + } + a, err := v.Accept("operator", seat.AcceptInput{Pipeline: pkey, Step: "unit", Artifact: "app:program", Why: "flaky"}) + if err != nil || a.Result.Outcome != "accepted" || a.Result.Accepted == nil || a.Result.Accepted.By != "operator" || a.Result.Accepted.Why != "flaky" { + t.Fatalf("accept = %+v, %v", a, err) + } + if len(*acts) != 1 || !strings.Contains((*acts)[0], "mesh-pipeline-runner.accept") || !strings.Contains((*acts)[0], "flaky") { + t.Errorf("hand acts = %v", *acts) + } + // A rerun of the lint step: a failed pipeline runs again. + end("lint", seat.StepResultOutcomeFailed) + p, rev, _ := store.Get(pkey) + p.Outcome = seat.PipelineOutcomeFailed + store.Update(pkey, p, rev) + old := stepNamed(p, "app:program", "lint").Ask + r, err := v.Rerun("seat:mesh-healers", seat.RerunInput{Pipeline: pkey, Step: "lint", Artifact: "app:program", Why: "retry"}) + if err != nil || r.Result == nil || r.Result.Outcome != "running" || r.Result.Ask == old || !r.Result.Rerun || r.Result.By != "seat:mesh-healers" { + t.Fatalf("rerun = %+v, %v", r.Result, err) + } + if len(asks.asks) != 1 || asks.ids[0] != r.Result.Ask { + t.Fatalf("asked %v under %v, want one ask under %s", asks.asks, asks.ids, r.Result.Ask) + } + if b, ok := asks.asks[0].(nodebuildgo.StepAsk); !ok || b.ID != r.Result.Ask || b.Step == nil || !b.Step.Rerun || b.Step.By != "seat:mesh-healers" || b.Step.Step != "lint" { + t.Errorf("the ask = %+v", asks.asks[0]) + } + if got, _, _ := store.Get(pkey); got.Outcome != seat.PipelineOutcomeRunning { + t.Errorf("the failed pipeline is %s, want running again", got.Outcome) + } + if _, err := v.Rerun("operator", seat.RerunInput{Pipeline: pkey, Step: "lint", Artifact: "app:program", Why: "again"}); err == nil { + t.Error("a running step was asked again") + } + if _, err := v.Rerun("operator", seat.RerunInput{Pipeline: pkey, Step: "approval", Why: "again"}); err == nil { + t.Error("an approval was asked again by rerun") + } + // A cancel ends the pipeline cancelled and drops the bounds of the asks still held. + c, err := v.Cancel("operator", seat.CancelInput{Pipeline: pkey, Why: "wrong branch"}) + if err != nil || c.Pipeline.Outcome != "cancelled" || c.Pipeline.EndCount != 1 || len(*dropped) == 0 { + t.Fatalf("cancel = %+v, %v, dropped %v", c.Pipeline, err, *dropped) + } + if s := stepNamed(c.Pipeline, "app:program", "lint"); s.Outcome != "cancelled" || s.By != "operator" { + t.Errorf("the lint step = %+v", s) + } + if _, err := v.Cancel("operator", seat.CancelInput{Pipeline: pkey, Why: "again"}); err == nil { + t.Error("a cancelled pipeline was cancelled again") + } + if _, err := v.Rerun("operator", seat.RerunInput{Pipeline: pkey, Step: "lint", Artifact: "app:program", Why: "again"}); err == nil { + t.Error("a step of a cancelled pipeline was asked again") + } + last := (*said)[len(*said)-1] + if !strings.HasPrefix(last, "pipeline-ended.") { + t.Errorf("last said %q, want pipeline-ended", last) + } + if n := len(*acts); n != 3 { + t.Errorf("%d hand acts recorded, want accept, rerun, cancel (3): %v", n, *acts) + } +} + +// A hand act the log does not take is a verb that does nothing (ADR 0565 clause 11). +func TestAVerbWhoseHandActIsNotRecordedDoesNothing(t *testing.T) { + v, store, _, _, _, _ := verbsFixture() + v.Acts = failingActs{} + if _, err := v.Cancel("operator", seat.CancelInput{Pipeline: pkey, Why: "x"}); err == nil { + t.Fatal("cancel did something") + } + if p, _, _ := store.Get(pkey); p.Outcome != seat.PipelineOutcomeRunning { + t.Errorf("the pipeline is %s", p.Outcome) + } +} + +type failingActs struct{} + +func (failingActs) RecordHandAct(views.HandAct) error { return errors.New("no holder") }