diff --git a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go index ca58da01..5fbc284d 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go @@ -27,6 +27,7 @@ const ( VerbAccept = "accept" VerbCancel = "cancel" VerbAskAgain = "ask-again" + VerbRelease = "release" ) const ( @@ -42,7 +43,7 @@ const ( // Verbs is every verb the seat serves, in its definition's order: a new copy each call. func Verbs() []string { - return []string{VerbRun, VerbStart, VerbShow, VerbList, VerbStands, VerbWhyCarried, VerbKept, VerbReview, VerbRerun, VerbAccept, VerbCancel, VerbAskAgain} + return []string{VerbRun, VerbStart, VerbShow, VerbList, VerbStands, VerbWhyCarried, VerbKept, VerbReview, VerbRerun, VerbAccept, VerbCancel, VerbAskAgain, VerbRelease} } // Events is every event published on the seat's subject: a new copy each call. @@ -959,6 +960,18 @@ type AskAgainAnswer struct { Result StepResult `json:"result"` } +// ReleaseInput is what release takes. +type ReleaseInput struct { + // why, written with the caller the runtime proves in the hand-act log + Why string `json:"why"` +} + +// ReleaseAnswer is what release answers. +type ReleaseAnswer struct { + // how many held pipelines were started + Released int64 `json:"released"` +} + // PipelineStartedEvent is the body of the event pipeline-started. type PipelineStartedEvent = PipelineStarted diff --git a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json index b3509f40..1a4cf069 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json @@ -501,6 +501,35 @@ ] }, "class": "act" + }, + { + "name": "release", + "description": "The pipelines the runner holds for its new version started by hand with the running runner, whatever the new version's state; written in the hand-act log (design 47 G 276, A-206).", + "input": { + "type": "object", + "properties": { + "why": { + "type": "string", + "description": "why, written with the caller the runtime proves in the hand-act log" + } + }, + "required": [ + "why" + ] + }, + "answer": { + "type": "object", + "properties": { + "released": { + "type": "integer", + "description": "how many held pipelines were started" + } + }, + "required": [ + "released" + ] + }, + "class": "act" } ], "events": [ diff --git a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts index 40972353..cb393110 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts @@ -17,6 +17,7 @@ export declare const verbs: { readonly accept: "accept"; readonly cancel: "cancel"; readonly askAgain: "ask-again"; + readonly release: "release"; }; export declare const events: { @@ -578,6 +579,18 @@ export interface AskAgainAnswer { "result": StepResult; } +/** is what release takes. */ +export interface ReleaseInput { + /** why, written with the caller the runtime proves in the hand-act log */ + "why": string; +} + +/** is what release answers. */ +export interface ReleaseAnswer { + /** how many held pipelines were started */ + "released": number; +} + /** is the body of the event pipeline-started. */ export type PipelineStartedEvent = PipelineStarted; diff --git a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js index a36f83ae..5b96596a 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js @@ -17,6 +17,7 @@ export const verbs = { accept: "accept", cancel: "cancel", askAgain: "ask-again", + release: "release", }; export const events = { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go index ed97400a..9e8adee7 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go @@ -118,10 +118,28 @@ func WaitsTool(r pipeline.Reads) stdio.Tool { Run: func(map[string]any) (any, error) { return r.Waits() }} } +// ReleaseTool is the runner's tool runner_release: it starts the pipelines held for the runner's new version with the running +// runner, through the seat's verb release, which the process that holds the pipelines answers and records in the hand-act log +// first (design 47 G §276, A-206). +func ReleaseTool(ask func(key string, body any) (json.RawMessage, error)) stdio.Tool { + return stdio.Tool{Name: Prefix + "_release", + Description: "Start now, with the running pipeline runner, the pipelines it holds until its new version runs healthy " + + "(runner_waits lists them), each wait ending released. Written in the hand-act log (mesh-hand-acts).", + Input: map[string]any{"type": "object", "required": []string{"why"}, "properties": map[string]any{ + "why": map[string]any{"type": "string", "description": "why, in plain words"}}}, + Run: func(args map[string]any) (any, error) { + in, err := as[seat.ReleaseInput](args) + if err != nil { + return nil, err + } + return ask("seat:"+seat.Seat+"."+seat.VerbRelease, in) + }} +} + func main() { r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}} tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) - tools = append(tools, WaitsTool(r)) + tools = append(tools, WaitsTool(r), ReleaseTool(stdio.Ask)) tools = append(tools, HealthTool(r.Pipelines, kept{stdio.State(pipeline.StateConsumers)}, time.Now)) if err := stdio.Serve("", tools); err != nil { fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] %v\n", err) diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go index bb3dd146..ef0bd12b 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "fmt" "strings" "testing" @@ -28,7 +29,17 @@ func TestTools(t *testing.T) { } // RN25 (§468, §76; A-164): the health check is the bundle's runner_health. names = append(names, HealthTool(none{}, none{}, nil).Name) - want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again runner_health" + // A-206: the tool runner_release asks the seat's verb release with its why. + var asked string + rel := ReleaseTool(func(key string, body any) (json.RawMessage, error) { + asked = key + fmt.Sprintf(" %+v", body) + return json.RawMessage(`{"released":1}`), nil + }) + if _, err := rel.Run(map[string]any{"why": "w"}); err != nil || asked != "seat:mesh-pipeline-runner.release {Why:w}" { + t.Errorf("runner_release asked %q, %v", asked, err) + } + names = append(names, rel.Name) + want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again runner_health runner_release" if got := strings.Join(names, " "); got != want { t.Errorf("tools %q, want %q", got, want) } 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 1be958c2..1517865c 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/verbs.go @@ -38,8 +38,8 @@ func answerVerbs(conn *process.Conn, v pipeline.Verbs) (func(), error) { 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). +// writeHandlers are the seat's write verbs the process answers: review, rerun, accept, cancel, start, ask-again and +// release (§306; RN14; A-206). 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) }), @@ -48,6 +48,7 @@ func writeHandlers(v pipeline.Verbs) map[string]process.SeatHandler { 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) }), + seat.VerbRelease: verbOf(func(c string, in seat.ReleaseInput) (any, error) { return v.Release(c, in) }), } } 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 6b1db4e8..61801856 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 @@ -18,7 +18,7 @@ func TestEveryWriteVerbIsServed(t *testing.T) { got = append(got, v) } slices.Sort(got) - want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbReview, seat.VerbStart} + want := []string{seat.VerbAccept, seat.VerbAskAgain, seat.VerbCancel, seat.VerbRerun, seat.VerbRelease, seat.VerbReview, seat.VerbStart} slices.Sort(want) if !slices.Equal(got, want) { t.Errorf("the process answers %v, want %v", got, want) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go index 5186d2a6..04148002 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go @@ -5,6 +5,7 @@ import ( "encoding/hex" "errors" "fmt" + "slices" "time" runnerclient "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshpipelinerunnerclient" @@ -161,3 +162,40 @@ func (s *Starter) ReleaseExpired(store Updater, key string, view func(branch str } return 0, nil } + +// heldFingerprints are the fingerprints of the runner the held pipelines in store await, each once. +func heldFingerprints(store Updater) []string { + pkeys, err := store.Keys("*.*.*") + if err != nil { + return nil + } + var fps []string + for _, key := range pkeys { + if p, _, err := store.Get(key); err == nil && p.Held != nil && p.Outcome == seat.PipelineOutcomeRunning && !slices.Contains(fps, p.Held.Fingerprint) { + fps = append(fps, p.Held.Fingerprint) + } + } + return fps +} + +// Release starts by hand the pipelines held for the runner's new version with the running runner, recorded in the hand-act +// log first (§276, A-206): each says wait-ended `released` and pipeline-started, and the bound of each wait is dropped. It +// says how many it started. +func (v Verbs) Release(_ string, in seat.ReleaseInput) (seat.ReleaseAnswer, error) { + if v.Starter == nil || v.View == nil { + return seat.ReleaseAnswer{}, errors.New("the runner cannot release a pipeline: it holds no starter or no catalog half") + } + if err := v.record(seat.VerbRelease, "started the pipelines held for the new runner by hand", in.Why); err != nil { + return seat.ReleaseAnswer{}, err + } + fps := heldFingerprints(v.Hearer.Store) + n, err := v.Starter.Release(v.Hearer.Store, "", "released", v.View) + if n > 0 && v.Drop != nil { + for _, fp := range fps { + if derr := v.Drop(HeldKey(fp)); derr != nil && err == nil { + err = fmt.Errorf("the bound of the wait for %s is not dropped: %w", fp, derr) + } + } + } + return seat.ReleaseAnswer{Released: int64(n)}, err +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go index c4911fea..19628ec6 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go @@ -84,3 +84,44 @@ func TestPipelinesAreHeldForTheNewRunnerAndReleasedWhenItRuns(t *testing.T) { t.Error("a pipeline is still held") } } + +// The verb release starts the held pipelines by hand, recorded in the hand-act log first, saying why and ending each wait +// `released`, and drops the bound of the wait; a release with no why does nothing (design 47 G §276, A-206). +func TestReleaseVerbStartsTheHeldPipelinesByHand(t *testing.T) { + reg, entries := placeFixture() + entries["runner"] = catalog.CatalogEntry{Module: "runner"} + store := memCreated{memStore{}} + asks := &stepAsks{} + var said []string + s := &Starter{Pipelines: store, Steps: asks, Name: "the-runner", Self: "runner", SelfFp: "src2:old", Awaiting: Awaiting(store.memStore), + Registration: func(m string) (catalog.Registration, bool) { return reg, m == "app" }, + Say: func(subject string, _ any) error { + said = append(said, subject[strings.Index(subject, ".event.")+7:]) + return nil + }} + view := func(string) (Half, bool) { return Half{Entries: entries}, true } + if _, err := s.TakenIn(catalog.TakeInTaken{Key: "m.1", Branch: "main", Source: 4, Entries: []catalog.TakeInEntry{ + {Module: "app", Commit: "c1", Fingerprint: "src2:aa"}, {Module: "runner", Commit: "c2", Fingerprint: "src2:new"}}}, view); err != nil { + t.Fatal(err) + } + acts, dropped := &actLog{}, []string{} + v := Verbs{Hearer: &Hearer{Store: store.memStore}, Starter: s, View: view, Acts: acts, + Drop: func(k string) error { dropped = append(dropped, k); return nil }} + if _, err := v.Release("operator", seat.ReleaseInput{}); err == nil || len(*acts) != 0 { + t.Fatalf("a release saying no why: %v, acts %v", err, *acts) + } + said = nil + a, err := v.Release("operator", seat.ReleaseInput{Why: "the new runner will not come"}) + if err != nil || a.Released != 2 { + t.Fatalf("release = %+v, %v", a, err) + } + if p, _, _ := store.Get("main.4.app"); p.Held != nil || len(p.Steps) == 0 || len(asks.asks) == 0 { + t.Errorf("after the release: held %+v, steps %d", p.Held, len(p.Steps)) + } + if len(*acts) != 1 || !strings.HasPrefix((*acts)[0], "mesh-pipeline-runner.release: ") || len(dropped) != 1 || dropped[0] != HeldKey("src2:new") { + t.Errorf("acts %v, dropped %v", *acts, dropped) + } + if !strings.Contains(strings.Join(said, "|"), "wait-ended.app.main.c1") { + t.Errorf("said %v", said) + } +} diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index e29f902e..c47f2bfb 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -91,7 +91,7 @@ func TestManifest(t *testing.T) { "seat:node-build-image.check.step"})}, {"uses operator-channel", equalSets(m.Uses, []string{"operator-channel"})}, // RN25 (§539): the runner's tools carry the prefix runner_. - {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health", "runner_waits"})}, + {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health", "runner_waits", "runner_release"})}, // RN25 (§468, §76): the health check, run outside the process, is the tool runner_health. {"health is runner_health", healthTool(raw) == "tool runner_health"}, // RN1/AC5 (§95, §130): the new assigner is delivered before the runner. diff --git a/modules/mesh-delivery-pipeline/module.json b/modules/mesh-delivery-pipeline/module.json index 0b100789..206245fa 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -18,7 +18,8 @@ "rerun", "accept", "cancel", - "ask-again" + "ask-again", + "release" ] } ], @@ -26,7 +27,8 @@ "runner_first_views", "runner_first_view_again", "runner_health", - "runner_waits" + "runner_waits", + "runner_release" ], "consumes": [ "mesh-delivery-indexer.take-in.taken",