diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go index 6ddf5016..1be958c2 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go @@ -32,15 +32,23 @@ func verbOf[In any](do func(caller string, in In) (any, error)) process.SeatHand } } -// 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. +// answerVerbs answers the seat's write verbs over the bus, each writing the state pipelines by compare-and-set (design 47 +// G §22, §70, §306, §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) }), - }) + return conn.AnswerSeat(seat.Seat, writeHandlers(v)) +} + +// writeHandlers are the seat's write verbs the process answers: review, rerun, accept, cancel, start and ask-again +// (§306; RN14). +func writeHandlers(v pipeline.Verbs) map[string]process.SeatHandler { + return 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) }), + seat.VerbStart: verbOf(func(c string, in seat.StartInput) (any, error) { return v.Start(c, in) }), + seat.VerbAskAgain: verbOf(func(c string, in seat.AskAgainInput) (any, error) { return v.AskAgain(c, in) }), + } } // dropBound deletes the bound of an ask a cancelled pipeline no longer waits on. 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 index 9b5fb4d6..6b1db4e8 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs_test.go @@ -2,11 +2,29 @@ package main import ( "encoding/json" + "slices" "testing" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + + "mesh-delivery-pipeline/internal/pipeline" ) +// RN12, RN14 (§306): the process answers every write verb of the seat, the bundle every read verb, so each verb the +// manifest claims is served. +func TestEveryWriteVerbIsServed(t *testing.T) { + var got []string + for v := range writeHandlers(pipeline.Verbs{}) { + got = append(got, v) + } + slices.Sort(got) + want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbReview, seat.VerbStart} + slices.Sort(want) + if !slices.Equal(got, want) { + t.Errorf("the process answers %v, want %v", got, want) + } +} + // 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 diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/approval.go b/modules/mesh-delivery-pipeline/internal/pipeline/approval.go index 2756fa75..6497dfe5 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/approval.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/approval.go @@ -51,7 +51,15 @@ func approvalDue(p seat.Pipeline) int { // 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)) + // An approval asked again (ask-again) is an ask of its own, its id following from when it was asked again, so the + // ask rebuilt from the step is the same and the earlier, decided ask is not its duplicate. + idOf := p.Pipeline + "\n" + p.Commit + for _, st := range p.Steps { + if st.Step == seat.StepResultStepApproval && st.Rerun { + idOf += "\nasked again\n" + from.UTC().Format(time.RFC3339Nano) + } + } + sum := sha256.Sum256([]byte(idOf)) act := func(do string) (string, error) { return operatorchannelclient.ActDigest(operatorchannelclient.Act{"pipeline": p.Pipeline, "commit": p.Commit, "fingerprint": p.Fingerprint, "do": do}) @@ -98,6 +106,10 @@ func askApproval(key string, p *seat.Pipeline, from time.Time, asker Asker) (boo if i < 0 || asker == nil { return false, nil } + // An approval asked again keeps the time ask-again wrote, so an ask published and then not written is the same ask. + if s := p.Steps[i]; s.Rerun && s.Started != nil { + from = *s.Started + } a, err := ApprovalAsk(*p, from) if err != nil { return false, fmt.Errorf("the approval of %s cannot be asked: %w", key, err) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go index 65a924a6..e3caef80 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go @@ -334,3 +334,68 @@ func (v Verbs) askOf(p seat.Pipeline, s seat.StepResult, caller string, now time } return nil, fmt.Errorf("the %s step of %s is placed no ask now", s.Step, p.Pipeline) } + +// Start starts by hand a delivering pipeline for the head of a watched branch the runner's catalog half holds, recorded in +// the hand-act log first (§306, §394): the pipeline of that branch's source revision and the module, as a take-in would +// start it. A head whose pipeline is written already starts nothing and is refused, naming the pipeline. +func (v Verbs) Start(_ string, in seat.StartInput) (seat.StartAnswer, error) { + if in.Branch == "" || in.Module == "" { + return seat.StartAnswer{}, errors.New("start names the watched branch and the module") + } + if v.Starter == nil || v.View == nil { + return seat.StartAnswer{}, errors.New("the runner cannot start a pipeline: it holds no starter or no catalog half") + } + half, ok := v.View(in.Branch) + if !ok { + return seat.StartAnswer{}, fmt.Errorf("the catalog half of %s is not held yet: nothing was started", in.Branch) + } + e, ok := half.Entries[in.Module] + if !ok || half.Source <= 0 { + return seat.StartAnswer{}, fmt.Errorf("the catalog half of %s holds no entry of %s at a source revision: nothing was started", in.Branch, in.Module) + } + key := Key(in.Branch, half.Source, in.Module) + if err := v.record(seat.VerbStart, "started the pipeline "+key+" by hand", in.Why); err != nil { + return seat.StartAnswer{}, err + } + started, err := v.Starter.Start(Begin{Branch: in.Branch, Source: half.Source, Module: in.Module, Commit: e.Commit, Fp: e.Fp, + Repository: e.Repository, Revision: half.Revision, Entries: half.Entries}) + if err != nil { + return seat.StartAnswer{}, err + } + if !started { + return seat.StartAnswer{}, fmt.Errorf("the pipeline %s of the head is started already: a step of it is asked again by rerun", key) + } + return seat.StartAnswer{Pipeline: key}, nil +} + +// AskAgain asks a declined approval step once more (§306, §36), recorded in the hand-act log first: the step waits again, +// marked asked again by the caller the bus proved, and is asked at once when the pipeline's every other required step +// passed, under an ask of its own (the earlier ask's decision ended it). +func (v Verbs) AskAgain(caller string, in seat.AskAgainInput) (seat.AskAgainAnswer, error) { + p, i, err := v.change(in.Pipeline, seat.EventStepStarted, func() error { + return v.record(seat.VerbAskAgain, "asked the approval of "+in.Pipeline+" again", in.Why) + }, func(p *seat.Pipeline, now time.Time) (int, error) { + switch p.Outcome { + case seat.PipelineOutcomeCancelled, seat.PipelineOutcomeSuperseded: + return -1, fmt.Errorf("the pipeline %s is %s: its approval is not asked again", p.Pipeline, p.Outcome) + } + i, err := stepOf(*p, seat.StepResultStepApproval, "") + if err != nil { + return -1, err + } + s := &p.Steps[i] + if s.Outcome != seat.StepResultOutcomeDeclined { + return -1, fmt.Errorf("only a declined approval is asked again: the approval of %s is %s", p.Pipeline, s.Outcome) + } + *s = seat.StepResult{Step: s.Step, Artifact: s.Artifact, Status: s.Status, Fingerprint: s.Fingerprint, + Outcome: seat.StepResultOutcomeWaiting, By: caller, Rerun: true, Started: &now} + if p.Outcome != seat.PipelineOutcomeRunning { + p.Outcome, p.Why, p.Ended = seat.PipelineOutcomeRunning, "", nil + } + return i, nil + }) + if err != nil { + return seat.AskAgainAnswer{}, err + } + return seat.AskAgainAnswer{Result: p.Steps[i]}, nil +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go index 602f33c2..633f120b 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs_test.go @@ -170,3 +170,67 @@ func TestAVerbWhoseHandActIsNotRecordedDoesNothing(t *testing.T) { type failingActs struct{} func (failingActs) RecordHandAct(views.HandAct) error { return errors.New("no holder") } + +// RN14 (§306, §394): start starts by hand the pipeline of a watched branch's head, recorded in the hand-act log, and a head +// already started is refused; ask-again asks a declined approval once more under an ask of its own, recorded, and an +// approval not declined is refused. +func TestStartAndAskAgain(t *testing.T) { + v, store, _, _, acts, _ := verbsFixture() + made := created{} + v.Starter.Pipelines, v.Starter.Say = made, func(string, any) error { return nil } + _, entries := placeFixture() + v.View = func(string) (Half, bool) { return Half{Source: 7, Revision: 3, Entries: entries}, true } + a, err := v.Start("operator", seat.StartInput{Branch: "main", Module: "app", Why: "no pipeline for the head"}) + if err != nil || a.Pipeline != "main.7.app" || made["main.7.app"] == nil { + t.Fatalf("start = %+v, %v; written %v", a, err, len(made)) + } + if _, err := v.Start("operator", seat.StartInput{Branch: "main", Module: "app", Why: "again"}); err == nil { + t.Error("a second start of the same head was not refused") + } + if _, err := v.Start("operator", seat.StartInput{Branch: "main", Module: "absent", Why: "x"}); err == nil { + t.Error("a start of a module the half does not hold was not refused") + } + if len(*acts) != 2 || !strings.HasPrefix((*acts)[0], "mesh-pipeline-runner.start: ") { + t.Errorf("hand acts %v, want the two starts recorded", *acts) + } + + // ask-again: the approval of the fixture's pipeline is declined; asked again it is asked under a new ask. + h := v.Hearer + asker := h.Asker.(*askRecorder) + e := store[pkey] + first, err := ApprovalAsk(e.p, time.Unix(1, 0)) + if err != nil { + t.Fatal(err) + } + for i := range e.p.Steps { + s := &e.p.Steps[i] + switch s.Step { + case seat.StepResultStepApproval: + s.Outcome, s.Ask = seat.StepResultOutcomeDeclined, first.ID + default: + s.Outcome = seat.StepResultOutcomePassed + } + } + e.p.Outcome = seat.PipelineOutcomeFailed + store[pkey] = e + if _, err := v.AskAgain("operator", seat.AskAgainInput{Pipeline: pkey}); err == nil { + t.Error("an ask-again saying no why was taken") + } + got, err := v.AskAgain("operator", seat.AskAgainInput{Pipeline: pkey, Why: "asked by mistake"}) + if err != nil || got.Result.Outcome != seat.StepResultOutcomeRunning || got.Result.Ask == "" || got.Result.Ask == first.ID || !got.Result.Rerun { + t.Fatalf("ask-again = %+v, %v", got, err) + } + if len(asker.asked) != 1 || asker.asked[0].ID != got.Result.Ask { + t.Errorf("asked %v, want one ask %s", asker.asked, got.Result.Ask) + } + if store[pkey].p.Outcome != seat.PipelineOutcomeRunning { + t.Errorf("the pipeline is %s after ask-again, want running", store[pkey].p.Outcome) + } + // The ask rebuilt from the step, as a warrant is held to it, is the ask asked. + if again, err := ApprovalAsk(store[pkey].p, *stepNamed(store[pkey].p, "", seat.StepResultStepApproval).Started); err != nil || again.ID != got.Result.Ask { + t.Errorf("the ask rebuilt = %s, %v; want %s", again.ID, err, got.Result.Ask) + } + if _, err := v.AskAgain("operator", seat.AskAgainInput{Pipeline: pkey, Why: "twice"}); err == nil { + t.Error("an approval running was asked again") + } +}