RN4: the runner's process answers review, rerun, accept and cancel
A review ends a placed review step with its proven caller; rerun asks a step again under a new ask saying it is a rerun and by whom; accept takes a failed optional step; cancel ends a running pipeline cancelled and drops its asks' bounds. Each writes `pipelines` by compare-and-set, says step-ended, step-started or pipeline-ended, and records its hand act first, through AnswerSeat. Builds: design 47 G §22, §70, §482, §30, §325, §394, A-107 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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()
|
||||
|
||||
@@ -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)}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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:
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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") }
|
||||
Reference in New Issue
Block a user