diff --git a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go index 87206148..69160e95 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go @@ -39,6 +39,9 @@ const ( EventStepEnded = "step-ended" EventWaitHeld = "wait-held" EventWaitEnded = "wait-ended" + EventFilling = "filling" + EventFull = "full" + EventRoom = "room" ) // Verbs is every verb the seat serves, in its definition's order: a new copy each call. @@ -48,7 +51,7 @@ func Verbs() []string { // Events is every event published on the seat's subject: a new copy each call. func Events() []string { - return []string{EventPipelineStarted, EventPipelineEnded, EventGroupEnded, EventRunStarted, EventStepStarted, EventStepEnded, EventWaitHeld, EventWaitEnded} + return []string{EventPipelineStarted, EventPipelineEnded, EventGroupEnded, EventRunStarted, EventStepStarted, EventStepEnded, EventWaitHeld, EventWaitEnded, EventFilling, EventFull, EventRoom} } // Family is where the seat's events are kept: the shared events stream, filtered to the seat's subjects. @@ -145,6 +148,15 @@ func SubjectWaitEnded(module, branch, commitOrGroup string) string { // SubjectWaitEndedAll hears every wait-ended. const SubjectWaitEndedAll = "mesh.seat.mesh-pipeline-runner.event.wait-ended.*.*.*" +// SubjectFilling is where filling is published. +const SubjectFilling = "mesh.seat.mesh-pipeline-runner.event.filling" + +// SubjectFull is where full is published. +const SubjectFull = "mesh.seat.mesh-pipeline-runner.event.full" + +// SubjectRoom is where room is published. +const SubjectRoom = "mesh.seat.mesh-pipeline-runner.event.room" + const ( StatePipelines = "pipelines" StateKept = "kept" @@ -360,6 +372,8 @@ type Held struct { type KeptResult struct { // sha256: of what the package's tests read Digest string `json:"digest"` + // when a carry last read it: the runner ages it 30 days after (design 47 G section 330) + LastRead *time.Time `json:"last-read,omitempty"` // unit or integration Level string `json:"level"` Package string `json:"package"` @@ -372,7 +386,9 @@ type KeptResult struct { type LogEntry struct { Artifact string `json:"artifact,omitempty"` At time.Time `json:"at"` - Outcome string `json:"outcome,omitempty"` + // the keys of the kept results the step wrote or carried, ., which the runner ages by their last-read (design 47 G sections 330, 470) + Kept []string `json:"kept,omitempty"` + Outcome string `json:"outcome,omitempty"` // the output's name in step-outputs Output string `json:"output,omitempty"` Pipeline string `json:"pipeline"` @@ -454,6 +470,8 @@ type Pipeline struct { // every step, approval last Steps []StepResult `json:"steps"` Tests *TestCounts `json:"tests,omitempty"` + // the step-log entries of the steps that ended and are not appended to the step log yet, in order; written with the step's end in one write and cleared when appended (design 47 G sections 486, 606) + Unlogged []LogEntry `json:"unlogged,omitempty"` // the layout version V int64 `json:"v,omitempty"` // the version package-published names, for a pipeline the runner started from a publication or from outdated; none for any other (design 47 G 550) @@ -997,6 +1015,15 @@ type WaitHeldEvent = Wait // WaitEndedEvent is the body of the event wait-ended. type WaitEndedEvent = Wait +// FillingEvent is the body of the event filling. +type FillingEvent = Fill + +// FullEvent is the body of the event full. +type FullEvent = Fill + +// RoomEvent is the body of the event room. +type RoomEvent = Fill + // PipelinesByBranchRevisionModuleValue is one value of the state pipelines under the key ... type PipelinesByBranchRevisionModuleValue = Pipeline 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 60db0ab2..333a66f8 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json @@ -620,6 +620,24 @@ "max-age-days": 90, "max-mib": 32 } + }, + { + "name": "filling", + "body": { + "$ref": "#/defs/Fill" + } + }, + { + "name": "full", + "body": { + "$ref": "#/defs/Fill" + } + }, + { + "name": "room", + "body": { + "$ref": "#/defs/Fill" + } } ], "family": { @@ -760,6 +778,11 @@ "pipeline": { "type": "string", "description": "the pipeline whose step kept it" + }, + "last-read": { + "type": "string", + "format": "date-time", + "description": "when a carry last read it: the runner ages it 30 days after (design 47 G section 330)" } }, "description": "a package's passing result the runner keeps, by the digest of what its tests read (ADR 0666 section 8)", @@ -1053,6 +1076,13 @@ "version": { "type": "string", "description": "the version package-published names, for a pipeline the runner started from a publication or from outdated; none for any other (design 47 G 550)" + }, + "unlogged": { + "type": "array", + "items": { + "$ref": "#/defs/LogEntry" + }, + "description": "the step-log entries of the steps that ended and are not appended to the step log yet, in order; written with the step's end in one write and cleared when appended (design 47 G sections 486, 606)" } }, "description": "one pipeline as the runner keeps it in its state pipelines, by compare-and-set (design 47 *The runner's kept data*)", @@ -1569,6 +1599,13 @@ "at": { "type": "string", "format": "date-time" + }, + "kept": { + "type": "array", + "items": { + "type": "string" + }, + "description": "the keys of the kept results the step wrote or carried, ., which the runner ages by their last-read (design 47 G sections 330, 470)" } }, "description": "one entry of the step log", 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 7ca146f9..fa38aa41 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts @@ -29,6 +29,9 @@ export declare const events: { readonly stepEnded: "step-ended"; readonly waitHeld: "wait-held"; readonly waitEnded: "wait-ended"; + readonly filling: "filling"; + readonly full: "full"; + readonly room: "room"; }; export declare const family: { readonly name: "mesh-pipeline-runner.events"; readonly maxAgeDays: 7 }; @@ -81,6 +84,12 @@ export declare function waitEndedSubject(module: string, branch: string, commitO export declare const waitEndedSubjectAll: "mesh.seat.mesh-pipeline-runner.event.wait-ended.*.*.*"; +export declare const fillingSubject: "mesh.seat.mesh-pipeline-runner.event.filling"; + +export declare const fullSubject: "mesh.seat.mesh-pipeline-runner.event.full"; + +export declare const roomSubject: "mesh.seat.mesh-pipeline-runner.event.room"; + export declare const states: { readonly pipelines: "pipelines"; readonly kept: "kept"; @@ -157,6 +166,8 @@ export interface Held { export interface KeptResult { /** sha256: of what the package's tests read */ "digest": string; + /** when a carry last read it: the runner ages it 30 days after (design 47 G section 330) */ + "last-read"?: string; /** unit or integration */ "level": string; "package": string; @@ -169,6 +180,8 @@ export interface KeptResult { export interface LogEntry { "artifact"?: string; "at": string; + /** the keys of the kept results the step wrote or carried, ., which the runner ages by their last-read (design 47 G sections 330, 470) */ + "kept"?: string[]; "outcome"?: "running" | "waiting" | "passed" | "failed" | "nothing to run" | "bypassed" | "carried" | "could not run" | "waiting for a build node" | "did not build" | "accepted" | "declined" | "superseded" | "cancelled"; /** the output's name in step-outputs */ "output"?: string; @@ -216,6 +229,8 @@ export interface Pipeline { /** every step, approval last */ "steps": StepResult[]; "tests"?: TestCounts; + /** the step-log entries of the steps that ended and are not appended to the step log yet, in order; written with the step's end in one write and cleared when appended (design 47 G sections 486, 606) */ + "unlogged"?: LogEntry[]; /** the layout version */ "v"?: number; /** the version package-published names, for a pipeline the runner started from a publication or from outdated; none for any other (design 47 G 550) */ @@ -615,6 +630,15 @@ export type WaitHeldEvent = Wait; /** is the body of the event wait-ended. */ export type WaitEndedEvent = Wait; +/** is the body of the event filling. */ +export type FillingEvent = Fill; + +/** is the body of the event full. */ +export type FullEvent = Fill; + +/** is the body of the event room. */ +export type RoomEvent = Fill; + /** is one value of the state pipelines under the key ... */ export type PipelinesByBranchRevisionModuleValue = Pipeline; diff --git a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js index 5b96596a..64a18312 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js @@ -29,6 +29,9 @@ export const events = { stepEnded: "step-ended", waitHeld: "wait-held", waitEnded: "wait-ended", + filling: "filling", + full: "full", + room: "room", }; export const family = { name: "mesh-pipeline-runner.events", maxAgeDays: 7 }; @@ -97,6 +100,12 @@ export function waitEndedSubject(module, branch, commitOrGroup) { export const waitEndedSubjectAll = "mesh.seat.mesh-pipeline-runner.event.wait-ended.*.*.*"; +export const fillingSubject = "mesh.seat.mesh-pipeline-runner.event.filling"; + +export const fullSubject = "mesh.seat.mesh-pipeline-runner.event.full"; + +export const roomSubject = "mesh.seat.mesh-pipeline-runner.event.room"; + export const states = { pipelines: "pipelines", kept: "kept", 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 9e8adee7..fea92533 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 @@ -10,6 +10,8 @@ import ( "time" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/busstate" "git.novox.be/novox/mesh-sdk/go/views" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" @@ -136,10 +138,51 @@ func ReleaseTool(ask func(key string, body any) (json.RawMessage, error)) stdio. }} } +// StoresFill is each of the runner's stores' fill through the bundle's calls (design 47 G §366, §623, A-214): the states +// pipelines and kept by `mesh/state.info`, step-outputs by the object store's fill and step-log by the log's; the runner's +// process writes nothing for it. A store whose fill cannot be read is an error naming it. +func StoresFill(state func(string) (*stdio.StateInfo, error), objects func(string) (buscalls.Fill, error), + log func(string) (*stdio.LogInfo, error), now func() time.Time) func() ([]seat.Fill, error) { + return func() ([]seat.Fill, error) { + at := now().UTC() + var out []seat.Fill + for _, name := range []string{seat.StatePipelines, seat.StateStepOutputs, seat.StateKept, seat.StateStepLog} { + f := seat.Fill{Store: name, At: &at} + switch name { + case seat.StateStepOutputs: + o, err := objects(name) + if err != nil { + return nil, fmt.Errorf("%s: %w", name, err) + } + f.Bytes, f.Cap = int64(o.Bytes), int64(o.MaxBytes) + case seat.StateStepLog: + l, err := log(name) + if err != nil { + return nil, fmt.Errorf("%s: %w", name, err) + } + f.Bytes, f.Cap = int64(l.Bytes), int64(l.MaxBytes) + default: + s, err := state(name) + if err != nil { + return nil, fmt.Errorf("%s: %w", name, err) + } + f.Bytes, f.Cap = int64(s.Bytes), s.MaxBytes + } + out = append(out, f) + } + return out, nil + } +} + func main() { - r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}} + fill := StoresFill(func(n string) (*stdio.StateInfo, error) { return stdio.State(n).Info() }, + func(n string) (buscalls.Fill, error) { return busstate.ObjectsOf(n).Fill() }, + func(n string) (*stdio.LogInfo, error) { return stdio.Log(n).Info() }, time.Now) + r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}, Fill: fill} tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) tools = append(tools, WaitsTool(r), ReleaseTool(stdio.Ask)) + tools = append(tools, UnloggedTools(stdio.State(seat.StatePipelines), stdio.Log(seat.StateStepLog), meshhandactsclient.Recorder{Ask: stdio.Ask}, + func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-delivery-pipeline-tools] "+f+"\n", a...) })...) 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 ef0bd12b..6b96f97a 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,9 +2,11 @@ package main import ( "encoding/json" + "errors" "fmt" "strings" "testing" + "time" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/views" @@ -39,8 +41,61 @@ func TestTools(t *testing.T) { 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" + // §606, §549: the unlogged entries are listed and appended by hand. + for _, tl := range UnloggedTools(stdio.State("pipelines"), stdio.Log("step-log"), nil, nil) { + names = append(names, tl.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 runner_unlogged runner_unlogged_append" if got := strings.Join(names, " "); got != want { t.Errorf("tools %q, want %q", got, want) } } + +type refusingActs struct{} + +func (refusingActs) RecordHandAct(views.HandAct) error { return errors.New("the hand-act log is down") } + +// runner_unlogged_append is a hand act: it names why, and appends nothing when the act is not recorded (design 47 G §606). +func TestUnloggedAppendIsAHandAct(t *testing.T) { + var run func(map[string]any) (any, error) + for _, tl := range UnloggedTools(stdio.State("pipelines"), stdio.Log("step-log"), refusingActs{}, nil) { + if tl.Name == "runner_unlogged_append" { + run = tl.Run + } + } + if _, err := run(map[string]any{}); err == nil || !strings.Contains(err.Error(), "why") { + t.Errorf("an append naming no why: %v", err) + } + if _, err := run(map[string]any{"why": "the log has room"}); err == nil || !strings.Contains(err.Error(), "not recorded") { + t.Errorf("an append whose hand act is not recorded: %v", err) + } +} + +// StoresFill reads the two states through the state call, the object store and the log through their own, in the seat's order, +// and names a store it could not read (design 47 G §366, §623, A-214). +func TestStoresFillIsReadThroughTheBundlesCalls(t *testing.T) { + now := func() time.Time { return time.Date(2026, 11, 12, 8, 0, 0, 0, time.UTC) } + var asked []string + fill := StoresFill( + func(n string) (*stdio.StateInfo, error) { + asked = append(asked, "state:"+n) + return &stdio.StateInfo{Bytes: 5, MaxBytes: 64}, nil + }, + func(n string) (buscalls.Fill, error) { + asked = append(asked, "objects:"+n) + return buscalls.Fill{Bytes: 7, MaxBytes: 70}, nil + }, + func(n string) (*stdio.LogInfo, error) { + asked = append(asked, "log:"+n) + return &stdio.LogInfo{Bytes: 9, MaxBytes: 90}, nil + }, now) + got, err := fill() + want := "state:pipelines objects:step-outputs state:kept log:step-log" + if err != nil || strings.Join(asked, " ") != want || len(got) != 4 || got[1].Bytes != 7 || got[3].Cap != 90 || got[2].Store != "kept" { + t.Fatalf("fill %+v, %v, asked %v; want %q", got, err, asked, want) + } + failing := StoresFill(func(string) (*stdio.StateInfo, error) { return nil, errors.New("no bus") }, nil, nil, now) + if _, err := failing(); err == nil || !strings.Contains(err.Error(), "pipelines") { + t.Errorf("a store unread: %v", err) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/unlogged.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/unlogged.go new file mode 100644 index 00000000..12c73014 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/unlogged.go @@ -0,0 +1,105 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/views" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + stdio "git.novox.be/novox/mesh-sdk/go" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// UnloggedTools are the runner's tools over the step-log entries of steps that ended and are not appended yet (design 47 G §606, +// §549): runner_unlogged lists them, runner_unlogged_append appends them by hand, written in the hand-act log; it is of class +// act and takes §391's default callers. +func UnloggedTools(pipelines stdio.KeptState, log stdio.KeptLog, acts views.HandActs, logf func(string, ...any)) []stdio.Tool { + h := &pipeline.Hearer{Store: stdioPipelines{pipelines}, Logf: logf, + Log: func(e seat.LogEntry) error { + _, err := log.Append(keys.EscapeName(e.Pipeline), e) + return err + }, + LogLast: func(key string) (*seat.LogEntry, error) { + e, err := log.Last(keys.EscapeName(key)) + if err != nil || e == nil { + return nil, err + } + var le seat.LogEntry + return &le, json.Unmarshal(e.Value, &le) + }} + return []stdio.Tool{ + {Name: Prefix + "_unlogged", Description: "The step-log entries of steps that ended and are not appended to the step log yet, by pipeline, in order.", + Input: map[string]any{"type": "object", "properties": map[string]any{}}, + Run: func(map[string]any) (any, error) { + out := map[string][]seat.LogEntry{} + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + return nil, err + } + for _, k := range pkeys { + if p, _, err := h.Store.Get(k); err == nil && len(p.Unlogged) > 0 { + out[k] = p.Unlogged + } + } + return map[string]any{"unlogged": out}, nil + }}, + {Name: Prefix + "_unlogged_append", Description: "Append the unlogged step-log entries to the step log by hand, in their order, each after reading back the log's last entry of its pipeline; it names why, and is a hand act.", + Input: map[string]any{"type": "object", "properties": map[string]any{"why": map[string]any{"type": "string"}}, "required": []string{"why"}}, + Run: func(args map[string]any) (any, error) { + why, _ := args["why"].(string) + if strings.TrimSpace(why) == "" { + return nil, errors.New("append names why") + } + if acts == nil { + return nil, errors.New("no way to record a hand act is given, so nothing is appended") + } + if err := acts.RecordHandAct(views.HandAct{What: "appended the unlogged step-log entries by hand", Why: strings.TrimSpace(why), + Cause: "unlogged-append", Verb: Prefix + "_unlogged_append"}); err != nil { + return nil, fmt.Errorf("the hand act was not recorded, so nothing was appended: %w", err) + } + return map[string]any{"appended": h.AppendUnlogged()}, nil + }}, + } +} + +// stdioPipelines is the state pipelines as the runtime serves it, for the by-hand append. +type stdioPipelines struct{ s stdio.KeptState } + +func (s stdioPipelines) Get(key string) (seat.Pipeline, uint64, error) { + e, err := s.s.Get(key) + if err != nil || e == nil { + return seat.Pipeline{}, 0, errors.Join(err, errors.New("no such pipeline")) + } + var p seat.Pipeline + return p, e.Revision, json.Unmarshal(e.Value, &p) +} + +func (s stdioPipelines) Update(key string, p seat.Pipeline, rev uint64) (uint64, error) { + r, err := s.s.Update(key, p, rev) + if err != nil && (strings.Contains(err.Error(), "wrong last sequence") || strings.Contains(err.Error(), "revision")) { + return 0, pipeline.ErrChanged + } + return r, err +} + +func (s stdioPipelines) Keys(filter string) ([]string, error) { + all, err := s.s.Keys() + var out []string + want := strings.Split(filter, ".") + for _, k := range all { + got := strings.Split(k, ".") + ok := len(got) == len(want) + for i := 0; ok && i < len(want); i++ { + ok = want[i] == "*" || want[i] == got[i] + } + if ok { + out = append(out, k) + } + } + return out, err +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go new file mode 100644 index 00000000..9a780571 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go @@ -0,0 +1,214 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "strconv" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/buscompose" + "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" +) + +// stepLog is the runner's step log, its old `logs` section (design 47 G §470): entries appended on the log's subject, `` +// as the key, and read onward by sequence from the stream the bus makes for the log. +type stepLog struct { + ctx context.Context + js jetstream.JetStream + module string +} + +func (l stepLog) stream() string { return buscompose.LogStreamName(l.module, seat.StateStepLog) } + +// Append appends one entry; the entry's identity is its pipeline, artifact, step, what and time, so the same entry appended twice +// is kept once. +func (l stepLog) Append(e seat.LogEntry) error { + raw, err := json.Marshal(e) + if err != nil { + return err + } + msg := nats.NewMsg(buscompose.LogSubject(l.module, seat.StateStepLog) + "." + seat.StepLogByPipelineKey(keys.EscapeName(e.Pipeline))) + msg.Data = raw + ctx, cancel := context.WithTimeout(l.ctx, 10*time.Second) + defer cancel() + _, err = l.js.PublishMsg(ctx, msg, jetstream.WithMsgID(fmt.Sprintf("%s|%s|%s|%s|%s", e.Pipeline, e.Artifact, e.Step, e.What, e.At.UTC().Format(time.RFC3339Nano)))) + return err +} + +// Read reads the entries from sequence from on, oldest first, at most limit of them. +func (l stepLog) Read(from uint64, limit int) ([]pipeline.LoggedEntry, error) { + if from == 0 { + from = 1 + } + st, err := l.js.Stream(l.ctx, l.stream()) + if err != nil { + return nil, fmt.Errorf("the step log %s is not open: %w", l.stream(), err) + } + info, err := st.Info(l.ctx) + if err != nil { + return nil, err + } + if info.State.Msgs == 0 || from > info.State.LastSeq { + return nil, nil + } + cons, err := st.OrderedConsumer(l.ctx, jetstream.OrderedConsumerConfig{DeliverPolicy: jetstream.DeliverByStartSequencePolicy, OptStartSeq: from}) + if err != nil { + return nil, err + } + batch, err := cons.Fetch(limit, jetstream.FetchMaxWait(2*time.Second)) + if err != nil { + return nil, err + } + var out []pipeline.LoggedEntry + for m := range batch.Messages() { + meta, err := m.Metadata() + if err != nil { + return out, err + } + var e seat.LogEntry + if err := json.Unmarshal(m.Data(), &e); err != nil { + continue + } + out = append(out, pipeline.LoggedEntry{Sequence: meta.Sequence.Stream, Entry: e}) + } + return out, batch.Error() +} + +// agedState is the state `aged`: one key per retention holding the last step-log sequence aged at it. +type agedState struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (a agedState) Get(name string) (uint64, error) { + e, err := a.kv.Get(a.ctx, name) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return 0, nil + } + if err != nil { + return 0, err + } + return strconv.ParseUint(string(e.Value()), 10, 64) +} + +func (a agedState) Put(name string, sequence uint64) error { + _, err := a.kv.Put(a.ctx, name, []byte(strconv.FormatUint(sequence, 10))) + return err +} + +// keptStore is the state kept as the ageing reads and deletes it. +type keptStore struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (k keptStore) Get(key string) (seat.KeptResult, bool, error) { + e, err := k.kv.Get(k.ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return seat.KeptResult{}, false, nil + } + var r seat.KeptResult + if err != nil { + return r, false, err + } + return r, true, json.Unmarshal(e.Value(), &r) +} + +func (k keptStore) Delete(key string) error { return k.kv.Delete(k.ctx, key) } + +// newAger is the ageing over the runner's own stores. +func newAger(ctx context.Context, js jetstream.JetStream, cred process.Account, pipelines, kept, aged jetstream.KeyValue, logf func(string, ...any)) *pipeline.Ager { + return &pipeline.Ager{Log: stepLog{ctx: ctx, js: js, module: cred.Module}.Read, Aged: agedState{ctx: ctx, kv: aged}, + Pipelines: pipelineDeleter{pipelineStore{ctx: ctx, kv: pipelines}}, Kept: keptStore{ctx: ctx, kv: kept}, Logf: logf, + Outputs: func(name string) error { + o, err := js.ObjectStore(ctx, catalogue.ObjectsBucket(cred.Module, seat.StateStepOutputs)) + if err != nil { + return err + } + if err := o.Delete(ctx, name); err != nil && !errors.Is(err, jetstream.ErrObjectNotFound) { + return err + } + return nil + }} +} + +// pipelineDeleter is the pipelines store with the delete only its owner makes (§330). +type pipelineDeleter struct{ pipelineStore } + +func (d pipelineDeleter) Delete(key string) error { return d.kv.Delete(d.ctx, key) } + +// storeFills reads the fill of the runner's stores against their caps: pipelines and kept (their key-value streams), step-outputs +// (its object store's) and step-log (its log's), as the bus's own streams say it (design 47 G §366, §486). +func storeFills(ctx context.Context, js jetstream.JetStream, cred process.Account, now func() time.Time) func() ([]seat.Fill, error) { + streams := []struct{ store, stream string }{ + {seat.StatePipelines, "KV_" + ownBucket(cred.Module, seat.StatePipelines)}, + {seat.StateKept, "KV_" + ownBucket(cred.Module, seat.StateKept)}, + {seat.StateStepOutputs, "OBJ_" + catalogue.ObjectsBucket(cred.Module, seat.StateStepOutputs)}, + {seat.StateStepLog, buscompose.LogStreamName(cred.Module, seat.StateStepLog)}, + } + return func() ([]seat.Fill, error) { + var out []seat.Fill + var first error + for _, s := range streams { + st, err := js.Stream(ctx, s.stream) + if err == nil { + var info *jetstream.StreamInfo + if info, err = st.Info(ctx); err == nil { + at := now().UTC() + out = append(out, seat.Fill{Store: s.store, Bytes: int64(info.State.Bytes), Cap: info.Config.MaxBytes, At: &at}) + } + } + if err != nil && first == nil { + first = fmt.Errorf("the fill of %s is not read: %w", s.store, err) + } + } + return out, first + } +} + +// Last is the step log's last entry of a pipeline, nil when the log holds none (§549). +func (l stepLog) Last(pipelineKey string) (*seat.LogEntry, error) { + st, err := l.js.Stream(l.ctx, l.stream()) + if err != nil { + return nil, err + } + msg, err := st.GetLastMsgForSubject(l.ctx, buscompose.LogSubject(l.module, seat.StateStepLog)+"."+seat.StepLogByPipelineKey(keys.EscapeName(pipelineKey))) + if errors.Is(err, jetstream.ErrMsgNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + var e seat.LogEntry + return &e, json.Unmarshal(msg.Data, &e) +} + +// whileTheLogIsFull appends the unlogged entries and reads the stores' fill again once a minute while the step log is full, so the +// entries the bus ages out say `room` with no step ending (design 47 G §606). +func whileTheLogIsFull(ctx context.Context, fills *pipeline.Fills, h *pipeline.Hearer, every time.Duration) { + t := time.NewTicker(every) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + if !slices.Contains(fills.Full(), seat.StateStepLog) { + continue + } + h.AppendUnlogged() + fills.Check() + } + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/asks.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/asks.go index e58e5e06..f6f6e239 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/asks.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/asks.go @@ -3,11 +3,14 @@ package main import ( "context" "encoding/json" + "fmt" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + gitseat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git" "git.novox.be/novox/mesh-sdk/go/process" "mesh-delivery-pipeline/internal/pipeline" @@ -39,3 +42,25 @@ func (b busStepAsker) AskStep(accept, id string, body any) error { } var _ pipeline.StepAsker = busStepAsker{} + +// commitAuthor reads the forge's login of the author of a commit through the seat git's verb pull, by the commit that merged it +// (design 47 G §33, A-191): a commit no pull request merged, or a read that failed, is an error, never an empty author. +func commitAuthor(call meshcatalogclient.Call) func(repository, commit string) (string, error) { + return func(repository, commit string) (string, error) { + raw, err := call("seat:"+gitseat.Seat+"."+gitseat.VerbPull, gitseat.PullInput{Repository: repository, MergeCommit: commit}) + if err != nil { + return "", err + } + var a gitseat.PullAnswer + if err := json.Unmarshal(raw, &a); err != nil { + return "", fmt.Errorf("the answer of %s.%s is not its seat's: %w", gitseat.Seat, gitseat.VerbPull, err) + } + switch { + case a.Error != "": + return "", fmt.Errorf("%s (%s)", a.Error, a.ErrorKind) + case a.Pull == nil || a.Pull.Author == "": + return "", fmt.Errorf("no pull request that merged %s names its author", commit) + } + return a.Pull.Author, nil + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go index b8d2d35e..7a92f89e 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go @@ -16,7 +16,7 @@ import ( ) // findingsWriter is the runner's finding writer as the mesh-sdk makes it. -type findingsWriter = views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent] +type findingsWriter = views.HeldWriter[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent] // KindPipelineWaits is the finding of pipelines held for the runner's new version that the wait's bound started with the running // runner (design 47 G §266, A-206). @@ -33,15 +33,13 @@ func lateFinding(p seat.Pipeline, held seat.Held, runner, node string, at time.T Subject: condseat.Subject{Scope: views.FindingScope, ID: p.Module, Machine: node}} } -// findingWriter is the runner's mesh-sdk finding writer (design 47 G §461, §689): the runner observes only what its views -// reader finds, so its own observe states none and the writer's statement covers the reader's open findings. The writer +// findingWriter is the runner's mesh-sdk held finding writer (design 47 G §461, §689, A-29): the runner raises its findings +// as they happen (a failed head, §33), so the writer holds what it has written and its statement says exactly that, with the +// reader's open findings. The writer // forms its subjects itself (design 47 G §383, A-13). func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*findingsWriter, error) { - w, err := views.NewWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ + w, err := views.NewHeldWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ Module: cred.Module, State: state, Node: cred.Node, Asked: condseat.SubjectFindingsAsked, JS: js, Conn: conn, Logf: logf, - Observe: func(context.Context) (views.Observed[condseat.Finding], error) { - return views.Observed[condseat.Finding]{}, nil - }, }) if err != nil { return nil, err 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 f6b1d924..b3bddb1c 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -54,6 +54,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + // The last step-log sequence aged at each retention (§330). + aged, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, "aged"), "age its own kept data") + if err != nil { + return err + } + // The commits named a recovery, whose steps are asked on the stream delivery (§81, §86, A-207). + recovered, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, "recoveries"), "know a recovery's commit") + if err != nil { + return err + } // The bound of each ask a build seat's holder must take: a key with its own time to live, its expiry heard (§535, A-77). bounds, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, "bounds"), "bound an ask no holder takes") if err != nil { @@ -101,7 +111,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } hearer := newHearer(ctx, js, cred, pipelines, kept, say, logger.Printf) hearer.Asker = busAsker{conn: conn, cred: cred} + hearer.Ager = newAger(ctx, js, cred, pipelines, kept, aged, logger.Printf) + // The stores' fill is said before a store is full, and a full one refuses a start (§366). + fills := &pipeline.Fills{Read: storeFills(ctx, js, cred, time.Now), Logf: logger.Printf, + Say: func(event string, f seat.Fill) error { return say("mesh.seat."+seat.Seat+".event."+event, f) }} + hearer.Fills = fills + // A pipeline's end names what its change widens (§71, A-207), computed by this running runner from the catalog and the grants part. + hearer.Widens = widenerOf(calls.Call).Widens + hearer.LogLast = stepLog{ctx: ctx, js: js, module: cred.Module}.Last hearer.Where = routerRecordOf(cred.Module, calls.Call, calls.Ask) + hearer.Failures = &pipeline.Failures{Findings: rv.Findings, Observer: cred.Module, Author: commitAuthor(calls.Call), Logf: logger.Printf} // Its asking half (§22, §24): a pipeline's steps are placed from the module's registration and each build seat is asked // for its step. ends := &pipeline.BuildEnds{} @@ -117,7 +136,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := rv.Findings.Write(ctx, lateFinding(p, held, module, node, time.Now().UTC())); err != nil { logger.Printf("the finding pipeline-waits of %s is not written: %v", p.Pipeline, err) } - }} + }, + Fills: fills, Recovered: kvRecoveries{ctx, recovered}.Has} resume := func(branch string) { half, ok := rv.Catalog.View(branch) if !ok { @@ -164,9 +184,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err := hearPublished(ctx, heard, starter, calls, rv.Catalog, logger.Printf); err != nil { return err } + if err := hearRecovery(heard, kvRecoveries{ctx, recovered}, logger.Printf); err != nil { + return err + } if err := hearTakenIn(heard, starter, rv.Catalog, logger.Printf); err != nil { return err } + // An entry whose step ended and was not appended is appended at the start, in order (§606, §549). + hearer.AppendUnlogged() + go whileTheLogIsFull(ctx, fills, hearer, time.Minute) 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) } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries.go new file mode 100644 index 00000000..468b9019 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries.go @@ -0,0 +1,55 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + + "github.com/nats-io/nats.go/jetstream" + + meshdelivery "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery" +) + +// recoveries is the runner's state `recoveries`: one key per commit named a recovery, holding the event that named it (design 47 +// G §81, §86, A-207). +type recoveries interface { + Put(commit string, raw []byte) error + Has(commit string) (bool, error) +} + +type kvRecoveries struct { + ctx context.Context + kv jetstream.KeyValue +} + +func (r kvRecoveries) Put(commit string, raw []byte) error { + _, err := r.kv.Put(r.ctx, commit, raw) + return err +} + +func (r kvRecoveries) Has(commit string) (bool, error) { + _, err := r.kv.Get(r.ctx, commit) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return false, nil + } + return err == nil, err +} + +// hearRecovery keeps each commit the seat mesh-delivery says a recovery merged in the state recoveries, which a pipeline's start +// and resume read (§81, §86): a commit heard after its pipeline started asks nothing again. +func hearRecovery(hub handlers, r recoveries, logf func(string, ...any)) error { + return hub.Handle(meshdelivery.SubjectRecoveryMerged, func(m jetstream.Msg) { + var ev meshdelivery.RecoveryMergedEvent + if err := json.Unmarshal(m.Data(), &ev); err != nil || ev.Commit == "" { + logf("%s is no recovery-merged: %v", m.Subject(), err) + _ = m.Term() + return + } + if err := r.Put(ev.Commit, m.Data()); err != nil { + logf("the recovery %s of %s is not kept, heard again: %v", ev.Commit, ev.Repository, err) + _ = m.Nak() + return + } + _ = m.Ack() + }) +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries_test.go new file mode 100644 index 00000000..476ee663 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries_test.go @@ -0,0 +1,35 @@ +package main + +import ( + "testing" +) + +type memRecoveries map[string][]byte + +func (m memRecoveries) Put(c string, raw []byte) error { m[c] = raw; return nil } +func (m memRecoveries) Has(c string) (bool, error) { _, ok := m[c]; return ok, nil } + +type dataMsg struct { + ackedMsg + data []byte + acks *[]string +} + +func (m dataMsg) Data() []byte { return m.data } +func (m dataMsg) Term() error { *m.acks = append(*m.acks, "term"); return nil } + +// A recovery-merged of the seat mesh-delivery is kept by its commit in the state recoveries; one that is no event is dropped +// (design 47 G §81, §86, A-207). +func TestARecoveryMergedIsKeptByItsCommit(t *testing.T) { + hub := &oneHub{} + kept := memRecoveries{} + if err := hearRecovery(hub, kept, func(string, ...any) {}); err != nil { + t.Fatal(err) + } + var acks []string + hub.handle(dataMsg{data: []byte(`{"v":1,"repository":"o/app","commit":"c9","why":"the build is down","by":"s@n"}`), acks: &acks}) + hub.handle(dataMsg{data: []byte(`not json`), acks: &acks}) + if ok, _ := kept.Has("c9"); !ok || len(kept) != 1 || len(acks) != 1 || acks[0] != "term" { + t.Errorf("kept %v, acks %v: want c9 kept and the other dropped", kept, acks) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go index cc28c592..293c9da3 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go @@ -19,6 +19,7 @@ import ( builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshbuilds" catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/testbus" @@ -56,7 +57,7 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes defer cancel() const module = "mesh-delivery-pipeline" - for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views", "consumers"} { + for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views", "consumers", "aged", "recoveries"} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_" + state, History: 1}); err != nil { t.Fatal(err) } @@ -64,6 +65,10 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_bounds", History: 1, LimitMarkerTTL: time.Minute}); err != nil { t.Fatal(err) } + // The step log, the runner's old logs section (§470): the stream the bus makes for it. + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buscompose.LogStreamName(module, seat.StateStepLog), Subjects: []string{buscompose.LogSubject(module, seat.StateStepLog) + ".>"}}); err != nil { + t.Fatal(err) + } events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}) if err != nil { t.Fatal(err) @@ -161,6 +166,12 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes waitFor(t, ctx, logs, func() bool { return readPipeline(ctx, pipelines, key, &p) && p.Outcome == seat.PipelineOutcomePassed }) + // §470, §330: the step log is the log's stream, read onward by sequence the way the ageing reads it. + logged, err := stepLog{ctx: ctx, js: js, module: module}.Read(1, 100) + if err != nil || len(logged) == 0 || logged[0].Entry.Pipeline != key { + t.Errorf("the step log holds %+v (%v), want the entries of %s", logged, err, key) + } + // RN25 (§76, §468; A-166): the durable consumer's age is in the state consumers, none waiting once every message heard // is acknowledged, and runner_health reads it healthy. consumers, err := js.KeyValue(ctx, module+"_consumers") diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go index 58f0ad0f..998394bd 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go @@ -79,9 +79,7 @@ func (s pipelineStore) Keys(filter string) ([]string, error) { 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: cred.Module, - Log: func(e seat.LogEntry) error { - return say(cred.EventSubject("step-log."+keys.EscapeName(e.Pipeline)), e) - }, + Log: stepLog{ctx: ctx, js: js, module: cred.Module}.Append, Keep: func(key string, k seat.KeptResult) error { raw, err := json.Marshal(k) if err == nil { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go index 1236cf75..aebeab85 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go @@ -20,7 +20,8 @@ import ( type runnerViews struct { Registrations *meshcatalogclient.Registrations Catalog *pipeline.Views - // Findings is the runner's finding writer, which also says the runner's own findings (A-206). + // Findings is the runner's held finding writer: it raises the finding pipeline-failed (§33) and says the runner's own + // findings (A-206). Findings *findingsWriter } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/widens.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/widens.go new file mode 100644 index 00000000..7d1bd889 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/widens.go @@ -0,0 +1,81 @@ +package main + +import ( + "encoding/json" + "fmt" + "strconv" + + "git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshgrants" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/keys" + "git.novox.be/novox/mesh-sdk/go/seatlist" + + "mesh-delivery-pipeline/internal/pipeline" +) + +// widenerOf is the runner's computation of `widens` (design 47 G §71, A-207, A-215, A-220) over its calls: the catalog seat's +// `revision` for the candidate and the branch's newest, the holders from the seat holders' one reading, and the grants part's +// `grants-for`; no other verb. +func widenerOf(call func(key string, body any) (json.RawMessage, error)) *pipeline.Widener { + return &pipeline.Widener{ + Revision: catalogRevision(call), + Holder: func(s string) (string, error) { return seatlist.HolderOf(call, s) }, + GrantsFor: func(branch string, source int64) (meshgrants.GrantsFor, error) { + return grantsFor(call, branch, source) + }, + } +} + +// catalogRevision reads one revision of a branch whole, page by page; revision 0 asks the newest (§195). +func catalogRevision(call func(string, any) (json.RawMessage, error)) func(string, int64) (pipeline.Half, error) { + return func(branch string, revision int64) (pipeline.Half, error) { + in := catalog.RevisionInput{Branch: keys.EscapeName(branch), Revision: revision} + var half pipeline.Half + for { + raw, err := call("seat:"+catalog.Seat+"."+catalog.VerbRevision, in) + if err != nil { + return pipeline.Half{}, err + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return pipeline.Half{}, fmt.Errorf("%s answered: %s", catalog.VerbRevision, said) + } + var a catalog.RevisionAnswer + if err := json.Unmarshal(answer, &a); err != nil { + return pipeline.Half{}, fmt.Errorf("%s answered no revision: %w", catalog.VerbRevision, err) + } + if a.NoneYet || a.Revision == nil { + return pipeline.Half{}, fmt.Errorf("%s holds no revision %d of %s", catalog.VerbRevision, revision, branch) + } + if half.Entries == nil { + half = pipeline.Half{Revision: a.Revision.Revision, Source: a.Revision.Source, Entries: map[string]catalog.CatalogEntry{}} + in.Revision = a.Revision.Revision // the pages of one revision, though a newer one is said between them + } + for _, e := range a.Entries { + half.Entries[e.Module] = e + } + if a.Next == "" { + return half, nil + } + in.After = a.Next + } + } +} + +// grantsFor asks the grants part what it would compose for the candidate's source revision on its branch. +func grantsFor(call func(string, any) (json.RawMessage, error), branch string, source int64) (meshgrants.GrantsFor, error) { + raw, err := call("seat:"+meshgrants.Seat+"."+meshgrants.VerbGrantsFor, meshgrants.GrantsForInput{Branch: branch, Source: strconv.FormatInt(source, 10)}) + if err != nil { + return meshgrants.GrantsFor{}, err + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return meshgrants.GrantsFor{}, fmt.Errorf("%s answered: %s", meshgrants.VerbGrantsFor, said) + } + var g meshgrants.GrantsFor + if err := json.Unmarshal(answer, &g); err != nil { + return meshgrants.GrantsFor{}, fmt.Errorf("%s answered no grants: %w", meshgrants.VerbGrantsFor, err) + } + return g, nil +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/widens_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/widens_test.go new file mode 100644 index 00000000..785819c9 --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/widens_test.go @@ -0,0 +1,35 @@ +package main + +import ( + "encoding/json" + "strings" + "testing" +) + +// The runner reads a revision whole, page by page, of the revision the first page named, and the candidate by its number; the grants +// part is asked with the source revision (design 47 G §195, §441, A-207). +func TestWidenerReadsARevisionWholeAndAsksGrantsFor(t *testing.T) { + var asked []string + call := func(key string, body any) (json.RawMessage, error) { + raw, _ := json.Marshal(body) + asked = append(asked, key+" "+string(raw)) + switch { + case strings.HasSuffix(key, ".grants-for"): + return json.RawMessage(`{"modules":[{"module":"app","added":[{"subject":"s","verb":"v"}],"removed":[]}]}`), nil + case strings.Contains(string(raw), `"after":"a"`): + return json.RawMessage(`{"revision":{"revision":8,"kind":"trunk","entries":[]},"entries":[{"module":"b","commit":"c","repository":"r","manifest":{}}]}`), nil + } + return json.RawMessage(`{"revision":{"revision":8,"kind":"trunk","entries":[]},"next":"a","entries":[{"module":"a","commit":"c","repository":"r","manifest":{}}]}`), nil + } + half, err := catalogRevision(call)("feat/x", 0) + if err != nil || half.Revision != 8 || len(half.Entries) != 2 { + t.Fatalf("half %+v, %v", half, err) + } + if !strings.Contains(asked[0], `"branch":"feat`) || !strings.Contains(asked[1], `"revision":8`) || strings.Contains(asked[0], `"revision"`) { + t.Errorf("asked %v: the first page names no revision, the next the one the first answered", asked) + } + g, err := grantsFor(call, "main", 31) + if err != nil || len(g.Modules) != 1 || !strings.Contains(asked[len(asked)-1], `"source":"31"`) { + t.Errorf("grants %+v, %v, asked %v", g, err, asked) + } +} diff --git a/modules/mesh-delivery-pipeline/go.mod b/modules/mesh-delivery-pipeline/go.mod index 028117ac..b0ceb01e 100644 --- a/modules/mesh-delivery-pipeline/go.mod +++ b/modules/mesh-delivery-pipeline/go.mod @@ -10,8 +10,10 @@ require ( require ( git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/git v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/modules/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 @@ -31,4 +33,6 @@ require ( require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants v0.0.0-00010101000000-000000000000 -require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 // indirect +require git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshdelivery v0.0.0-00010101000000-000000000000 + +require git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshgrants v0.0.0-00010101000000-000000000000 diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/aged.go b/modules/mesh-delivery-pipeline/internal/pipeline/aged.go new file mode 100644 index 00000000..487cc8c5 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/aged.go @@ -0,0 +1,177 @@ +package pipeline + +import ( + "fmt" + "time" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// Retention is one of the three ages the runner's own kept data are deleted at (design 47 G §330, *The runner's kept data*): its +// name is the key of the state `aged` that holds the last step-log sequence aged at it. +type Retention struct { + Name string + Age time.Duration +} + +// The retentions: a step's output and a pipeline not kept to 90 days at 7 days, a kept result not read for 30 days, a passed +// pipeline at 90 days. +var ( + Retention7 = Retention{"7d", 7 * 24 * time.Hour} + Retention30 = Retention{"30d", 30 * 24 * time.Hour} + Retention90 = Retention{"90d", 90 * 24 * time.Hour} +) + +// AgedPerStep is the most entries one step's end ages at one retention (§330). +const AgedPerStep = 20 + +// LoggedEntry is one entry of the step log with its sequence. +type LoggedEntry struct { + Sequence uint64 + Entry seat.LogEntry +} + +// Ager ages the runner's own kept data by its own deletes, in the step log's order (design 47 G §330): at each step's end it +// reads the step log onward from each retention's sequence, at most AgedPerStep entries, while the entry is past the +// retention, deletes what the retention says and moves the sequence; nothing is scanned whole, nothing waits on a timer. +type Ager struct { + // Log reads the step log onward from a sequence, oldest first, at most limit entries. + Log func(from uint64, limit int) ([]LoggedEntry, error) + // Aged is the state `aged`: the last sequence aged at a retention, 0 when none. + Aged interface { + Get(name string) (uint64, error) + Put(name string, sequence uint64) error + } + Pipelines interface { + Updater + Delete(key string) error + } + // Outputs deletes an object of step-outputs, Kept reads and deletes a key of the state kept. + Outputs func(name string) error + Kept interface { + Get(key string) (seat.KeptResult, bool, error) + Delete(key string) error + } + Now func() time.Time + // Logf is the runner's journal: each delete writes a line naming the key or object and its retention. + Logf func(string, ...any) +} + +func (a *Ager) logf(format string, args ...any) { + if a.Logf != nil { + a.Logf(format, args...) + } +} + +// Age is the ageing of one step's end; an error of one retention is said and the others go on. +func (a *Ager) Age() { + if a == nil || a.Log == nil { + return + } + for _, r := range []Retention{Retention7, Retention30, Retention90} { + if err := a.ageAt(r); err != nil { + a.logf("aged: the data past %s are not aged: %v", r.Name, err) + } + } +} + +func (a *Ager) ageAt(r Retention) error { + now := time.Now().UTC() + if a.Now != nil { + now = a.Now().UTC() + } + last, err := a.Aged.Get(r.Name) + if err != nil { + return err + } + entries, err := a.Log(last+1, AgedPerStep) + if err != nil { + return err + } + moved := last + for _, e := range entries { + if now.Sub(e.Entry.At) < r.Age { + break + } + a.ageEntry(r, e.Entry, now) + moved = e.Sequence + } + if moved == last { + return nil + } + return a.Aged.Put(r.Name, moved) +} + +func (a *Ager) ageEntry(r Retention, e seat.LogEntry, now time.Time) { + switch r.Name { + case Retention7.Name: + if e.Output != "" && a.Outputs != nil { + if err := a.Outputs(e.Output); err != nil { + a.logf("aged: the output %s is not deleted: %v", e.Output, err) + } else { + a.logf("aged: deleted the output %s (%s)", e.Output, r.Name) + } + } + a.deletePipeline(r, e.Pipeline, true) + case Retention30.Name: + for _, key := range e.Kept { + k, ok, err := a.Kept.Get(key) + if err != nil || !ok || k.LastRead == nil || now.Sub(*k.LastRead) < r.Age { + if err != nil { + a.logf("aged: the kept result %s is not read: %v", key, err) + } + continue + } + if err := a.Kept.Delete(key); err != nil { + a.logf("aged: the kept result %s is not deleted: %v", key, err) + continue + } + a.logf("aged: deleted the kept result %s (%s after its last read)", key, r.Name) + } + default: + a.deletePipeline(r, e.Pipeline, false) + } +} + +// deletePipeline deletes the pipeline key names unless it is its branch's newest for its module, and at 7 days unless it ended +// passed on a watched branch (a run on request's is no watched branch's). +func (a *Ager) deletePipeline(r Retention, key string, keepPassed bool) { + p, _, err := a.Pipelines.Get(key) + if err != nil { + return // gone already, or never kept + } + if p.Outcome == seat.PipelineOutcomeRunning || (keepPassed && p.Outcome == seat.PipelineOutcomePassed && p.Run == "") { + return + } + if !keepPassed && (p.Outcome != seat.PipelineOutcomePassed || p.Run != "") { + return // a pipeline not passed went at 7 days; a run's too + } + newest, err := a.newest(p) + if err != nil { + a.logf("aged: the pipeline %s is not judged, so it is kept: %v", key, err) + return + } + if newest { + return + } + if err := a.Pipelines.Delete(key); err != nil { + a.logf("aged: the pipeline %s is not deleted: %v", key, err) + return + } + a.logf("aged: deleted the pipeline %s (%s)", key, r.Name) +} + +// newest says whether p is its branch's newest pipeline for its module. +func (a *Ager) newest(p seat.Pipeline) (bool, error) { + pkeys, err := a.Pipelines.Keys(keys.EscapeName(p.Branch) + ".*." + p.Module) + if err != nil { + return false, fmt.Errorf("the pipelines of %s on %s are not listed: %w", p.Module, p.Branch, err) + } + for _, k := range pkeys { + if q, _, err := a.Pipelines.Get(k); err == nil && q.Source > p.Source { + return false, nil + } + } + return true, nil +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/aged_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/aged_test.go new file mode 100644 index 00000000..eb3ae6fc --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/aged_test.go @@ -0,0 +1,144 @@ +package pipeline + +import ( + "fmt" + "strconv" + "strings" + "testing" + "time" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +type agedState map[string]uint64 + +func (a agedState) Get(name string) (uint64, error) { return a[name], nil } +func (a agedState) Put(name string, s uint64) error { a[name] = s; return nil } + +// agedPipelines is memStore that deletes. +type agedPipelines struct { + memStore + deleted []string +} + +func (a *agedPipelines) Delete(key string) error { + delete(a.memStore, key) + a.deleted = append(a.deleted, key) + return nil +} + +type keptStore struct { + m map[string]seat.KeptResult + deleted []string +} + +func (k *keptStore) Get(key string) (seat.KeptResult, bool, error) { + v, ok := k.m[key] + return v, ok, nil +} +func (k *keptStore) Delete(key string) error { + delete(k.m, key) + k.deleted = append(k.deleted, key) + return nil +} + +type agedFixture struct { + ager *Ager + log []LoggedEntry + aged agedState + pipes *agedPipelines + kept *keptStore + outputs []string + journal []string + readFrom []uint64 +} + +var agedNow = time.Date(2026, 11, 12, 12, 0, 0, 0, time.UTC) + +func newAgedFixture() *agedFixture { + f := &agedFixture{aged: agedState{}, pipes: &agedPipelines{memStore: memStore{}}, kept: &keptStore{m: map[string]seat.KeptResult{}}} + f.ager = &Ager{Aged: f.aged, Pipelines: f.pipes, Kept: f.kept, Now: func() time.Time { return agedNow }, + Outputs: func(n string) error { f.outputs = append(f.outputs, n); return nil }, + Logf: func(format string, a ...any) { f.journal = append(f.journal, sprintf(format, a...)) }, + Log: func(from uint64, limit int) ([]LoggedEntry, error) { + f.readFrom = append(f.readFrom, from) + var out []LoggedEntry + for _, e := range f.log { + if e.Sequence >= from && len(out) < limit { + out = append(out, e) + } + } + return out, nil + }} + return f +} + +func sprintf(format string, a ...any) string { return fmt.Sprintf(format, a...) } + +func (f *agedFixture) pipeline(branch string, source int64, outcome string, run string) string { + key := branch + "." + strconv.FormatInt(source, 10) + ".app" + f.pipes.memStore[key] = struct { + p seat.Pipeline + rev uint64 + }{seat.Pipeline{Pipeline: key, Branch: branch, Source: source, Module: "app", Outcome: outcome, Run: run}, 1} + return key +} + +func days(n int) time.Time { return agedNow.Add(-time.Duration(n) * 24 * time.Hour) } + +// The runner ages its own kept data in the step log's order, by the retentions of the design: at 7 days an output and the +// pipeline that did not pass or is not its branch's newest, at 30 days a kept result not read for 30 days, at 90 days a passed +// pipeline that is not its branch's newest; each retention keeps its own sequence in the state aged (design 47 G §330). +func TestTheRunnerAgesItsOwnKeptDataByTheStepLogsOrder(t *testing.T) { + f := newAgedFixture() + failed := f.pipeline("main", 1, seat.PipelineOutcomeFailed, "") + passed := f.pipeline("main", 2, seat.PipelineOutcomePassed, "") + old := f.pipeline("main", 3, seat.PipelineOutcomePassed, "") + newest := f.pipeline("main", 9, seat.PipelineOutcomeFailed, "") + stale, fresh := days(40), days(2) + f.kept.m["unit.sha256:a"] = seat.KeptResult{Digest: "sha256:a", Level: "unit", LastRead: &stale} + f.kept.m["unit.sha256:b"] = seat.KeptResult{Digest: "sha256:b", Level: "unit", LastRead: &fresh} + f.log = []LoggedEntry{ + {1, seat.LogEntry{Pipeline: old, Output: "out-1", At: days(100)}}, + {2, seat.LogEntry{Pipeline: failed, Output: "out-2", Kept: []string{"unit.sha256:a", "unit.sha256:b"}, At: days(35)}}, + {3, seat.LogEntry{Pipeline: passed, Output: "out-3", At: days(10)}}, + {4, seat.LogEntry{Pipeline: newest, Output: "out-4", At: days(9)}}, + {5, seat.LogEntry{Pipeline: newest, Output: "out-5", At: days(1)}}, + } + f.ager.Age() + if got := strings.Join(f.outputs, ","); got != "out-1,out-2,out-3,out-4" { + t.Errorf("outputs deleted past 7 days: %s", got) + } + if got := strings.Join(f.pipes.deleted, ","); got != failed+","+old { + t.Errorf("pipelines deleted: %s, want %s (the failed one at 7 days, the old passed one at 90; the passed at 7 stays; the newest stays)", got, failed+","+old) + } + if got := strings.Join(f.kept.deleted, ","); got != "unit.sha256:a" { + t.Errorf("kept results deleted: %s", got) + } + if f.aged["7d"] != 4 || f.aged["30d"] != 2 || f.aged["90d"] != 1 { + t.Errorf("sequences aged: %v, want 4, 2 and 1 (the entries past each retention)", f.aged) + } + for _, want := range []string{"deleted the output out-1 (7d)", "deleted the pipeline " + failed + " (7d)", "deleted the kept result unit.sha256:a (30d", "deleted the pipeline " + old + " (90d)"} { + if !strings.Contains(strings.Join(f.journal, "\n"), want) { + t.Errorf("the journal names no %q:\n%s", want, strings.Join(f.journal, "\n")) + } + } + // A second step's end reads onward from the moved sequences and deletes nothing again. + f.readFrom, f.outputs = nil, nil + f.ager.Age() + if len(f.outputs) != 0 || f.readFrom[0] != 5 { + t.Errorf("the second ageing read from %v and deleted %v; want it to read from 5", f.readFrom, f.outputs) + } +} + +// One step's end ages at most 20 entries per retention. +func TestOneStepEndAgesAtMostTwentyEntries(t *testing.T) { + f := newAgedFixture() + for i := uint64(1); i <= 25; i++ { + f.log = append(f.log, LoggedEntry{i, seat.LogEntry{Pipeline: "x", Output: "o" + strconv.FormatUint(i, 10), At: days(50)}}) + } + f.ager.Age() + if len(f.outputs) != AgedPerStep || f.aged["7d"] != 20 { + t.Fatalf("%d outputs deleted, sequence %d; want 20 and 20", len(f.outputs), f.aged["7d"]) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/approval.go b/modules/mesh-delivery-pipeline/internal/pipeline/approval.go index 6497dfe5..beeff984 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/approval.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/approval.go @@ -211,18 +211,21 @@ func (h *Hearer) OnDecided(event oc.Warrant) (seat.Pipeline, error) { return p, err } ended := endPipeline(&p, now) + i := stepOfAsk(p, w.Ask) + if i >= 0 { + p.Unlogged = append(p.Unlogged, endEntry(key, p.Steps[i], 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) } } + h.appendUnlogged(key) return p, h.said(p, false, ended) } } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go b/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go index c4a107b2..918adf93 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go @@ -77,16 +77,20 @@ func (h *Hearer) endNotTaken(key, ask string, bound time.Duration) error { s := &p.Steps[i] s.Outcome, s.Ended = seat.StepResultOutcomeCouldNotRun, &now s.Why = fmt.Sprintf("the seat %s has no holder that took ask %s within %s", s.Seat, ask, bound) + p.Unlogged = append(p.Unlogged, endEntry(key, *s, now)) ended := endPipeline(&p, now) if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { continue } else if err != nil { return fmt.Errorf("pipeline %s cannot be written: %w", key, err) } - h.logEnd(key, *s, now) + h.Failures.StepEnded(h.Store, p, *s) if err := h.say(seat.EventStepEnded, p, s); err != nil { return fmt.Errorf("pipeline %s's %s is written but not said: %w", key, seat.EventStepEnded, err) } + h.appendUnlogged(key) + h.Ager.Age() + h.Fills.Check() return h.said(p, false, ended) } } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/failed.go b/modules/mesh-delivery-pipeline/internal/pipeline/failed.go new file mode 100644 index 00000000..35fb6d12 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/failed.go @@ -0,0 +1,188 @@ +package pipeline + +import ( + "context" + "fmt" + "regexp" + "strings" + "time" + + mcc "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// PipelineFailedKind is the kind of the runner's finding of a failed head (design 47 G §33, §540). +const PipelineFailedKind = "pipeline-failed" + +// forgeOwner is the form of an owner that names a commit's author as the forge does (design 47 G §540, §678). +var forgeOwner = regexp.MustCompile(`^forge:[A-Za-z0-9._-]+$`) + +// Findings is the runner's finding writer as the failures use it: mesh-sdk's held writer of the seat mesh-conditions' finding. +type Findings interface { + Write(ctx context.Context, f condseat.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error + Held() []condseat.Finding +} + +// Failures says a failed head (design 47 G §33, §540, A-191): a `required` step that failed, or ended `could not run`, is the +// finding `pipeline-failed` of its commit, naming the module, the branch, the commit and the steps, with the commit's author as +// its owner `forge:`; a newer revision of that branch passing each of those steps clears it. The condition keeper +// raises it as the author's, and the operator's after two hours (§540). +type Failures struct { + Findings Findings + // Observer is the runner's own module name. + Observer string + // Author reads the forge's login of the commit's author through the seat git (the verb pull, by the commit that merged + // it); an error is a commit no pull request merged or a read that failed. + Author func(repository, commit string) (string, error) + Now func() time.Time + Logf func(string, ...any) +} + +func (f *Failures) logf(format string, args ...any) { + if f.Logf != nil { + f.Logf(format, args...) + } +} + +// failedSteps are the required steps of p that failed or could not run, the approval excepted: an approval's end is the +// operator's, not a failure of the commit. +func failedSteps(p seat.Pipeline) []seat.StepResult { + var out []seat.StepResult + for _, s := range p.Steps { + if s.Status == seat.StepResultStatusRequired && s.Step != seat.StepResultStepApproval && + (s.Outcome == seat.StepResultOutcomeFailed || s.Outcome == seat.StepResultOutcomeCouldNotRun) { + out = append(out, s) + } + } + return out +} + +func (f *Failures) held(commit string) *condseat.Finding { + for _, h := range f.Findings.Held() { + if h.Scope == mcc.ScopeMerge && h.ID == commit && h.Kind == PipelineFailedKind { + h := h + return &h + } + } + return nil +} + +// StepEnded is what the end of step s of pipeline p says of its head: a failed required step writes the finding of its commit, +// a step of a commit already found failed that is not one writes its owner when it had none, and a passing required step clears +// the findings of the older revisions it makes good. +func (f *Failures) StepEnded(store Updater, p seat.Pipeline, s seat.StepResult) { + if f == nil || f.Findings == nil || p.Commit == "" { + return + } + ctx := context.Background() + switch { + case len(failedSteps(p)) > 0: + f.write(ctx, p) + case s.Outcome == seat.StepResultOutcomePassed || s.Outcome == seat.StepResultOutcomeCarried: + f.clearNewer(ctx, store, p) + } +} + +// write writes p's commit's finding; the owner it held stays, and one it lacked is read again here, so the owner of a +// failed read is written at the next step end of that commit (A-191). +func (f *Failures) write(ctx context.Context, p seat.Pipeline) { + steps := failedSteps(p) + var names []string + for _, s := range steps { + names = append(names, strings.TrimSpace(s.Artifact+" "+s.Step)+" ("+s.Outcome+")") + } + now := time.Now().UTC() + if f.Now != nil { + now = f.Now().UTC() + } + owner, why := "", "" + if old := f.held(p.Commit); old != nil { + owner = old.Owner + } + if owner == "" { + owner, why = f.author(p) + } + explanation := fmt.Sprintf("The required steps %s of %s did not pass on the branch %s at the commit %s. A newer commit of the branch that passes them clears this.", + strings.Join(names, ", "), p.Module, p.Branch, p.Commit) + if why != "" { + explanation += " " + why + } + err := f.Findings.Write(ctx, condseat.Finding{Scope: mcc.ScopeMerge, ID: p.Commit, Kind: PipelineFailedKind, Severity: condseat.FindingSeverityWarning, + Summary: fmt.Sprintf("%s on %s at %s: %s", p.Module, p.Branch, p.Commit, strings.Join(names, ", ")), + Headline: fmt.Sprintf("A required step of %s failed on %s", p.Module, p.Branch), + Explanation: explanation, Needs: "a newer commit of " + p.Branch + " that passes " + strings.Join(names, ", "), + Owner: owner, Source: "pipeline-runner." + PipelineFailedKind, Observer: f.Observer, At: now, + Subject: condseat.Subject{Scope: mcc.ScopeMerge, ID: p.Commit}}) + if err != nil { + f.logf("the finding %s of %s at %s is not written: %v", PipelineFailedKind, p.Module, p.Commit, err) + } +} + +// author is the finding's owner `forge:` and, when it has none, why it has none (A-191). +func (f *Failures) author(p seat.Pipeline) (string, string) { + if f.Author == nil { + return "", "The commit's author is not read, so the finding has no owner." + } + account, err := f.Author(p.Repository, p.Commit) + if err != nil { + f.logf("the author of %s in %s is not read, so its finding %s has no owner: %v", p.Commit, p.Repository, PipelineFailedKind, err) + return "", "The commit's author is not known (" + err.Error() + "), so the finding has no owner." + } + if !forgeOwner.MatchString("forge:" + account) { + f.logf("the author %q of %s is no forge account name, so its finding %s has no owner", account, p.Commit, PipelineFailedKind) + return "", "The commit's author is not known as a forge account, so the finding has no owner." + } + return "forge:" + account, "" +} + +// clearNewer clears the finding of each older commit of p's module and branch whose every failed step a revision newer than it +// has passed (§33: cleared when a newer revision of that branch passes that step). +func (f *Failures) clearNewer(ctx context.Context, store Updater, p seat.Pipeline) { + pkeys, err := store.Keys(keys.EscapeName(p.Branch) + ".*." + p.Module) + if err != nil { + f.logf("the pipelines of %s on %s are not listed, so no finding %s is cleared: %v", p.Module, p.Branch, PipelineFailedKind, err) + return + } + var all []seat.Pipeline + for _, k := range pkeys { + if q, _, err := store.Get(k); err == nil { + all = append(all, q) + } + } + for _, old := range all { + if old.Source >= p.Source || f.held(old.Commit) == nil { + continue + } + failed := failedSteps(old) + if len(failed) == 0 { + continue + } + made := true + for _, s := range failed { + made = made && passedAfter(all, old, s) + } + if made { + if err := f.Findings.Clear(ctx, mcc.ScopeMerge, old.Commit, PipelineFailedKind, "a newer revision of "+p.Branch+" passes its steps"); err != nil { + f.logf("the finding %s of %s at %s is not cleared: %v", PipelineFailedKind, p.Module, old.Commit, err) + } + } + } +} + +// passedAfter says whether a pipeline newer than old passed the step s of old. +func passedAfter(all []seat.Pipeline, old seat.Pipeline, s seat.StepResult) bool { + for _, q := range all { + if q.Source <= old.Source { + continue + } + for _, t := range q.Steps { + if t.Artifact == s.Artifact && t.Step == s.Step && (t.Outcome == seat.StepResultOutcomePassed || t.Outcome == seat.StepResultOutcomeCarried) { + return true + } + } + } + return false +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/failed_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/failed_test.go new file mode 100644 index 00000000..eabd3339 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/failed_test.go @@ -0,0 +1,166 @@ +package pipeline + +import ( + "context" + "errors" + "strconv" + "strings" + "testing" + + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" + nodebuildgo "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" +) + +// heldFindings is a finding writer in memory, by key. +type heldFindings struct { + held map[string]condseat.Finding + cleared []string +} + +func newHeld() *heldFindings { return &heldFindings{held: map[string]condseat.Finding{}} } + +func (h *heldFindings) Write(_ context.Context, f condseat.Finding) error { + h.held[f.Scope+"."+f.ID+"."+f.Kind] = f + return nil +} + +func (h *heldFindings) Clear(_ context.Context, scope, id, kind, _ string) error { + delete(h.held, scope+"."+id+"."+kind) + h.cleared = append(h.cleared, id) + return nil +} + +func (h *heldFindings) Held() []condseat.Finding { + var out []condseat.Finding + for _, f := range h.held { + out = append(out, f) + } + return out +} + +// A required step that failed or could not run is the finding pipeline-failed of its commit, owned by the commit's author as +// the forge records it; one whose author is not known has no owner and says why, and the next step end of that commit writes it +// (design 47 G §33, §540, A-191); an optional step that failed says nothing. +func TestAFailedRequiredStepIsTheFindingOfItsCommit(t *testing.T) { + for _, c := range []struct { + name, step, outcome string + author string + authorErr error + wantFinding bool + wantOwner string + }{ + {"required failed", "lint", "failed", "alice", nil, true, "forge:alice"}, + {"required could not run", "lint", "could not run", "alice", nil, true, "forge:alice"}, + {"no pull request merged it", "lint", "failed", "", errors.New("not found"), true, ""}, + {"optional failed", "unit", "failed", "alice", nil, false, ""}, + {"required passed", "lint", "passed", "alice", nil, false, ""}, + } { + t.Run(c.name, func(t *testing.T) { + h, store, _, _, _, _ := fixture() + p := store[pkey] + p.p.Repository = "novox/app" + store[pkey] = p + f := newHeld() + h.Failures = &Failures{Findings: f, Observer: "mesh-delivery-pipeline", Now: h.Now, + Author: func(repository, commit string) (string, error) { + if repository != "novox/app" || commit != "c1" { + t.Errorf("the author of %s in %s was asked", commit, repository) + } + return c.author, c.authorErr + }} + ask := map[string]string{"lint": "a1", "unit": "a2"}[c.step] + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended(ask, c.step, c.outcome)); err != nil { + t.Fatal(err) + } + got, ok := f.held["merge.c1.pipeline-failed"] + if ok != c.wantFinding { + t.Fatalf("finding written: %v, want %v (%v)", ok, c.wantFinding, f.held) + } + if !ok { + return + } + if got.Owner != c.wantOwner || got.Observer != "mesh-delivery-pipeline" || got.Subject.ID != "c1" { + t.Fatalf("finding %+v, want owner %q", got, c.wantOwner) + } + for _, name := range []string{"app", "main", "c1", "lint"} { + if !strings.Contains(got.Explanation, name) { + t.Errorf("the explanation does not name %q: %s", name, got.Explanation) + } + } + if c.wantOwner == "" && !strings.Contains(got.Explanation, "no owner") { + t.Errorf("an ownerless finding does not say why: %s", got.Explanation) + } + }) + } +} + +// The owner of a failed head whose author could not be read is written at the next step end of that commit (A-191). +func TestTheOwnerOfAFailedHeadIsWrittenAtTheNextStepEnd(t *testing.T) { + h, store, _, _, _, _ := fixture() + f := newHeld() + known := false + h.Failures = &Failures{Findings: f, Now: h.Now, Author: func(string, string) (string, error) { + if !known { + return "", errors.New("the forge did not answer") + } + return "bob", nil + }} + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a1", "lint", "failed")); err != nil { + t.Fatal(err) + } + if f.held["merge.c1.pipeline-failed"].Owner != "" { + t.Fatalf("an owner before the author is known: %+v", f.held) + } + known = true + p := store[pkey] + p.p.Steps[1].Status = seat.StepResultStatusRequired + store[pkey] = p + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a2", "unit", "failed")); err != nil { + t.Fatal(err) + } + if got := f.held["merge.c1.pipeline-failed"].Owner; got != "forge:bob" { + t.Fatalf("owner %q, want forge:bob", got) + } +} + +// A newer revision of the branch passing the step that failed clears the finding of the older commit; one that does not pass +// it, and a newer head failing on its own, leave it, and the newer failing head is a finding of its own (§33, §540). +func TestANewerPassingRevisionClearsTheFinding(t *testing.T) { + h, store, _, _, _, _ := fixture() + f := newHeld() + h.Failures = &Failures{Findings: f, Now: h.Now, Author: func(string, string) (string, error) { return "alice", nil }} + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a1", "lint", "failed")); err != nil { + t.Fatal(err) + } + newer := func(source int64, commit, ask string) string { + key := "main." + strconv.FormatInt(source, 10) + ".app" + store[key] = struct { + p seat.Pipeline + rev uint64 + }{seat.Pipeline{V: 1, Pipeline: key, Branch: "main", Source: source, Module: "app", Commit: commit, Fingerprint: "src2:aa", + Outcome: seat.PipelineOutcomeRunning, Steps: []seat.StepResult{{Artifact: "program", Step: seat.StepResultStepLint, + Status: seat.StepResultStatusRequired, Outcome: seat.StepResultOutcomeRunning, Ask: ask, Seat: nodebuildgo.Seat}}}, 1} + return key + } + // A newer head that fails the step too is a finding of its own, and the older one stays. + key := newer(6, "c2", "b1") + ev := ended("b1", "lint", "failed") + ev.Pipeline, ev.Commit = key, "c2" + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ev); err != nil { + t.Fatal(err) + } + if _, ok := f.held["merge.c1.pipeline-failed"]; !ok || len(f.held) != 2 { + t.Fatalf("findings %v, want c1 and c2", f.held) + } + // A newest head passing it clears both older commits' findings. + key = newer(7, "c3", "b2") + ev = ended("b2", "lint", "passed") + ev.Pipeline, ev.Commit = key, "c3" + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ev); err != nil { + t.Fatal(err) + } + if len(f.held) != 0 { + t.Fatalf("findings %v stand after a newer revision passed the step", f.held) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/fill.go b/modules/mesh-delivery-pipeline/internal/pipeline/fill.go new file mode 100644 index 00000000..506ea84b --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/fill.go @@ -0,0 +1,134 @@ +package pipeline + +import ( + "errors" + "fmt" + "sync" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// The fill at which a store says `filling` (80 %, past it) and `room` (70 %, below it) on the seat (design 47 G §366, §181). +const ( + FillingPercent = 80 + RoomPercent = 70 +) + +// ErrFull is a start refused because a store of the runner is full (§366): said, and heard again by the bus. +var ErrFull = errors.New("a store of the runner is full") + +// Fills says each of the runner's stores' fill before it is full (design 47 G §366): `filling` past 80 % of the cap, `full` at +// the cap, and `room` below 70 % once it said either; a store full refuses a new pipeline, loudly, and nothing is dropped to make room. +type Fills struct { + // Read is each store's fill now: pipelines, step-outputs, kept and step-log. + Read func() ([]seat.Fill, error) + // Say says one of the seat's events, filling, full or room, with the store's fill. + Say func(event string, f seat.Fill) error + Logf func(string, ...any) + mu sync.Mutex + said map[string]string // the store's last event said, "" before the first check + full map[string]bool +} + +// Check reads each store's fill and says what changed since the last check; it is read at each step's end and at each start. +func (f *Fills) Check() { + if f == nil || f.Read == nil { + return + } + fills, err := f.Read() + if err != nil { + f.logf("the stores' fill is not read: %v", err) + } + f.mu.Lock() + defer f.mu.Unlock() + if f.said == nil { + f.said, f.full = map[string]string{}, map[string]bool{} + } + for _, s := range fills { + if s.Cap <= 0 { + continue + } + event := "" + switch last := f.said[s.Store]; { + case s.Bytes >= s.Cap: + event = seat.EventFull + case s.Bytes*100 > s.Cap*FillingPercent: + if last != seat.EventFilling && last != seat.EventFull { + event = seat.EventFilling + } + case s.Bytes*100 < s.Cap*RoomPercent: + // A store below 70 % says room once it said otherwise, and at the first check after a start: a keeper holding a + // finding of the store from before is told it cleared. + if last != seat.EventRoom { + event = seat.EventRoom + } + } + f.full[s.Store] = s.Bytes >= s.Cap + if event == "" || event == f.said[s.Store] { + continue + } + if err := f.Say(event, s); err != nil { + f.logf("%s of %s is not said: %v", event, s.Store, err) + continue + } + f.said[s.Store] = event + } +} + +// MarkFull says a store full because the bus refused a write to it (the step log, §486): it refuses starts until a check reads it +// below its cap. +func (f *Fills) MarkFull(store string) { + if f == nil { + return + } + f.mu.Lock() + if f.said == nil { + f.said, f.full = map[string]string{}, map[string]bool{} + } + f.full[store] = true + first := f.said[store] != seat.EventFull + f.said[store] = seat.EventFull + f.mu.Unlock() + if first && f.Say != nil { + if err := f.Say(seat.EventFull, seat.Fill{Store: store}); err != nil { + f.logf("full of %s is not said: %v", store, err) + } + } +} + +// Full is the stores full now, as the last check read them. +func (f *Fills) Full() []string { + if f == nil { + return nil + } + f.mu.Lock() + defer f.mu.Unlock() + var out []string + for s, full := range f.full { + if full { + out = append(out, s) + } + } + return out +} + +func (f *Fills) logf(format string, args ...any) { + if f != nil && f.Logf != nil { + f.Logf(format, args...) + } +} + +// refuseStart is the error of a start while a store is full, naming each store; it is said in the journal, and the start is heard +// again by the bus, so the runner takes it up on room (§366). +func (f *Fills) refuseStart(key string) error { + if f == nil { + return nil + } + f.Check() + full := f.Full() + if len(full) == 0 { + return nil + } + f.logf("the start of %s is refused: %v full", key, full) + return fmt.Errorf("%w: %v; %s starts on room", ErrFull, full, key) +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/fill_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/fill_test.go new file mode 100644 index 00000000..84486f21 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/fill_test.go @@ -0,0 +1,52 @@ +package pipeline + +import ( + "errors" + "testing" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// A store past 80 % of its cap says `filling` once, at its cap `full` and refuses a start, said; below 70 % it says `room` and +// the start goes (design 47 G §366). +func TestAStoreFillingIsSaidAndAFullStoreRefusesAStart(t *testing.T) { + bytes := int64(10) + var said []string + var journal []string + f := &Fills{ + Read: func() ([]seat.Fill, error) { return []seat.Fill{{Store: "pipelines", Bytes: bytes, Cap: 100}}, nil }, + Say: func(event string, fl seat.Fill) error { said = append(said, event+":"+fl.Store); return nil }, + Logf: func(format string, a ...any) { journal = append(journal, format) }} + steps := []struct { + bytes int64 + say string + }{{10, "room:pipelines"}, {75, ""}, {81, "filling:pipelines"}, {85, ""}, {100, "full:pipelines"}, {75, ""}, {69, "room:pipelines"}} + for _, st := range steps { + bytes, said = st.bytes, nil + f.Check() + got := "" + if len(said) == 1 { + got = said[0] + } + if got != st.say || len(said) > 1 { + t.Fatalf("at %d of 100 bytes the runner said %v, want %q", st.bytes, said, st.say) + } + } + bytes = 100 + s := &Starter{Pipelines: failingCreate{}, Fills: f, Say: func(string, any) error { return nil }} + ok, err := s.Start(Begin{Branch: "main", Source: 1, Module: "app"}) + if ok || !errors.Is(err, ErrFull) { + t.Fatalf("a start with pipelines full: %v, %v; want it refused with ErrFull", ok, err) + } + if len(journal) == 0 { + t.Error("the refused start is not said in the journal") + } + bytes = 10 + if ok, err := s.Start(Begin{Branch: "main", Source: 1, Module: "app"}); !ok || err != nil { + t.Fatalf("a start on room: %v, %v", ok, err) + } +} + +type failingCreate struct{} + +func (failingCreate) Create(string, []byte) error { return nil } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go index b6222c7a..1b01137e 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go @@ -191,7 +191,7 @@ func (s *Starter) Release(store Updater, fp, how string, view func(branch string } if s.Steps != nil && len(asks) > 0 { s.carry(key, asks) - if err := askAll(s.Steps, s.Arm, key, asks); err != nil { + if err := askAll(s.Steps, s.Arm, key, asks, s.recovered(key, p.Commit)); err != nil { note(err) } } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/place.go b/modules/mesh-delivery-pipeline/internal/pipeline/place.go index 946ac980..78a39188 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/place.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/place.go @@ -32,6 +32,14 @@ var acceptStep = map[string]string{ nodebuildimage.Seat: nodebuildimage.AcceptSubjectCheckStep, } +// acceptRecovery is where a recovery's step is sent on the build seats' stream `delivery`, ahead of the first round (§86, §415); +// the seats mesh-builds and mesh-wants have no such stream, so their steps stay on their own accept subject. +var acceptRecovery = map[string]string{ + nodebuildgo.Seat: nodebuildgo.AcceptSubjectDeliveryStep, + nodebuildnode.Seat: nodebuildnode.AcceptSubjectDeliveryStep, + nodebuildimage.Seat: nodebuildimage.AcceptSubjectDeliveryStep, +} + // buildSteps are the steps an artifact's build seat runs, in the order they are asked (design 47 G §22). var buildSteps = []string{seat.StepResultStepLint, seat.StepResultStepUnit, seat.StepResultStepIntegration} @@ -140,10 +148,13 @@ func stepAsk(p seat.Pipeline, e catalog.CatalogEntry, a catalog.Artifact, step, } // askAll puts each ask on its seat's work queue; the first refusal is named, the rest are still put. -func askAll(asker StepAsker, arm Arm, key string, asks []Asked) error { +func askAll(asker StepAsker, arm Arm, key string, asks []Asked, recovery bool) error { var first error for _, a := range asks { accept, ok := acceptStep[a.Seat] + if to, named := acceptRecovery[a.Seat]; recovery && named { + accept = to + } err := fmt.Errorf("the seat %s is no build seat the runner asks", a.Seat) if ok { // A build seat's holder says ask-taken: the ask's bound is armed before it is put (§535, §647). diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read.go b/modules/mesh-delivery-pipeline/internal/pipeline/read.go index 4da65cb7..a01da243 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/read.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read.go @@ -22,6 +22,21 @@ type Kept interface { type Reads struct { Pipelines Kept Kept Kept + // Fill is each of the runner's stores' fill now, pipelines, step-outputs, kept and step-log, read through the bundle's calls + // (the state's `mesh/state.info`, the object store's and the log's fill; design 47 G §366, §623, A-214); nil says none. + Fill func() ([]seat.Fill, error) +} + +// fills is the stores' fill, or an error saying it could not be read: an answer never leaves it out unsaid. +func (r Reads) fills() ([]seat.Fill, error) { + if r.Fill == nil { + return nil, nil + } + f, err := r.Fill() + if err != nil { + return nil, fmt.Errorf("the stores' fill cannot be read: %w", err) + } + return f, nil } // Show is one pipeline whole, with its steps and the names of their outputs in step-outputs (§83). @@ -33,7 +48,11 @@ func (r Reads) Show(in seat.ShowInput) (seat.ShowAnswer, error) { if !ok { return seat.ShowAnswer{}, fmt.Errorf("no pipeline %q in %s", in.Pipeline, seat.StatePipelines) } - a := seat.ShowAnswer{Pipeline: p} + fills, err := r.fills() + if err != nil { + return seat.ShowAnswer{}, err + } + a := seat.ShowAnswer{Pipeline: p, Fill: fills} for _, s := range p.Steps { if s.Output != "" { a.Outputs = append(a.Outputs, s.Output) @@ -158,7 +177,18 @@ func (r Reads) KeptResults(in seat.KeptInput) (seat.KeptAnswer, error) { } out = append(out, v) } - return seat.KeptAnswer{Kept: out}, nil + fills, err := r.fills() + if err != nil { + return seat.KeptAnswer{}, err + } + a := seat.KeptAnswer{Kept: out} + for _, f := range fills { + if f.Store == seat.StateKept { + f := f + a.Fill = &f + } + } + return a, nil } func (r Reads) pipeline(key string) (seat.Pipeline, bool, error) { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go index 5dabc4b4..3666f396 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/read_test.go @@ -2,6 +2,7 @@ package pipeline import ( "encoding/json" + "errors" "strings" "testing" "time" @@ -143,3 +144,25 @@ func TestWaitsListTheHeldPipelines(t *testing.T) { t.Fatalf("waits = %+v, %v", got, err) } } + +// show carries each store's fill (pipelines, step-outputs, kept, step-log) and kept the kept store's, read through the bundle's state +// call, never counted from the keys; a fill that cannot be read is an error, not an answer without it (design 47 G §366, §623, A-214). +func TestShowAndKeptCarryTheStoresFill(t *testing.T) { + pipes, kept := memKept{}, memKept{} + put(t, pipes, "main.7.alpha", seat.Pipeline{Pipeline: "main.7.alpha", Module: "alpha"}) + fills := []seat.Fill{{Store: seat.StatePipelines, Bytes: 10, Cap: 100}, {Store: seat.StateStepOutputs, Bytes: 20, Cap: 200}, + {Store: seat.StateKept, Bytes: 30, Cap: 300}, {Store: seat.StateStepLog, Bytes: 40, Cap: 400}} + r := Reads{Pipelines: pipes, Kept: kept, Fill: func() ([]seat.Fill, error) { return fills, nil }} + shown, err := r.Show(seat.ShowInput{Pipeline: "main.7.alpha"}) + if err != nil || len(shown.Fill) != 4 || shown.Fill[1].Store != seat.StateStepOutputs || shown.Fill[3].Cap != 400 { + t.Errorf("show: %+v, %v; want the four stores' fill", shown.Fill, err) + } + k, err := r.KeptResults(seat.KeptInput{}) + if err != nil || k.Fill == nil || k.Fill.Store != seat.StateKept || k.Fill.Bytes != 30 { + t.Errorf("kept: %+v, %v; want the kept store's fill", k.Fill, err) + } + r.Fill = func() ([]seat.Fill, error) { return nil, errors.New("the bus does not answer") } + if _, err := r.Show(seat.ShowInput{Pipeline: "main.7.alpha"}); err == nil || !strings.Contains(err.Error(), "fill") { + t.Errorf("show with the fill unread: %v", err) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 3d01f85b..ddba572e 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -82,6 +82,10 @@ type Starter struct { Ends func(module, fp string) (time.Time, bool) // Late says a pipeline the bound's expiry started with the running runner (A-206): the finding pipeline-waits. Late func(p seat.Pipeline, held seat.Held) + // Fills refuses a start while a store of the runner is full (§366); nil refuses none. + Fills *Fills + // Recovered says whether a commit is named a recovery (§81, §86): its steps are asked on the build seats' stream delivery. + Recovered func(commit string) (bool, error) } func newPipeline(b Begin, key string, now time.Time) seat.Pipeline { @@ -109,6 +113,9 @@ func (s *Starter) Start(b Begin) (bool, error) { now = s.Now().UTC() } key := Key(b.Branch, b.Source, b.Module) + if err := s.Fills.refuseStart(key); err != nil { + return false, err + } if b.Await == "" && s.Awaiting != nil { if fp, ok := s.Awaiting(); ok { b.Await = fp @@ -143,7 +150,7 @@ func (s *Starter) Start(b Begin) (bool, error) { } if s.Steps != nil && len(asks) > 0 { s.carry(key, asks) - if err := askAll(s.Steps, s.Arm, key, asks); err != nil { + if err := askAll(s.Steps, s.Arm, key, asks, s.recovered(key, b.Commit)); err != nil { return true, err } } @@ -201,7 +208,7 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) { continue } s.carry(key, due) - if err := askAll(s.Steps, s.Arm, key, due); err != nil && first == nil { + if err := askAll(s.Steps, s.Arm, key, due, s.recovered(key, p.Commit)); err != nil && first == nil { first = err } asked += len(due) @@ -209,6 +216,19 @@ func (s *Starter) Resume(store Updater, branch string, half Half) (int, error) { return asked, first } +// recovered says whether the commit of pipeline key is named a recovery; a read that fails asks on the stream check, said, since +// a step is asked either way (a recovery's only goes ahead of the first round). +func (s *Starter) recovered(key, commit string) bool { + if s.Recovered == nil || commit == "" { + return false + } + named, err := s.Recovered(commit) + if err != nil { + s.logf("whether %s of %s is a recovery is not read, so its steps are asked on check: %v", commit, key, err) + } + return named +} + // carry puts the kept results of each ask's package and the toolchain of its language on its step ask (§25, §416, A-104). A // read that fails asks the steps with none, so every package runs: slower, never wrong; it is said. func (s *Starter) carry(key string, asks []Asked) { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go index a5c60fc5..e8518f84 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go @@ -95,6 +95,8 @@ type Hearer struct { // 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 + // LogLast is the step log's last entry of a pipeline, nil when it has none (§549). + LogLast func(pipeline string) (*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 @@ -108,6 +110,14 @@ type Hearer struct { Where func(ask string) (string, *oc.Warrant, error) Logf func(string, ...any) Now func() time.Time + // Failures says a failed head as the finding pipeline-failed (§33); nil says none. + Failures *Failures + // Ager ages the runner's own kept data at each step's end (§330); nil ages none. + Ager *Ager + // Fills says the stores' fill at each step's end (§366); nil says none. + Fills *Fills + // Widens computes the widens a pipeline-ended carries (§71, A-207); nil carries none. + Widens func(seat.Pipeline) []seat.Widening } func (h *Hearer) now() time.Time { @@ -161,6 +171,12 @@ func (h *Hearer) OnStepEnded(bs StepSeat, subject string, ev StepEnded) (seat.Pi } s.Output = name } + var keptKeys []string + for _, k := range ev.Kept { + keptKeys = append(keptKeys, seat.KeptByLevelDigestKey(keys.EscapeName(k.Level), keys.EscapeName(k.Digest))) + } + unlogged := endEntry(ev.Pipeline, *s, now, keptKeys...) + p.Unlogged = append(p.Unlogged, unlogged) from := now if ev.Ended != nil { from = ev.Ended.UTC() @@ -177,7 +193,7 @@ func (h *Hearer) OnStepEnded(bs StepSeat, subject string, ev StepEnded) (seat.Pi } if h.Keep != nil { for _, k := range ev.Kept { - kept := seat.KeptResult{Digest: k.Digest, Level: k.Level, Package: k.Package, Pipeline: k.Pipeline} + kept := seat.KeptResult{Digest: k.Digest, Level: k.Level, Package: k.Package, Pipeline: k.Pipeline, LastRead: &now} if k.Toolchain != nil { kept.Toolchain = &seat.ToolchainUsed{Language: k.Toolchain.Language, Recipe: k.Toolchain.Recipe, Version: k.Toolchain.Version} } @@ -189,10 +205,14 @@ func (h *Hearer) OnStepEnded(bs StepSeat, subject string, ev StepEnded) (seat.Pi } } } - h.logEnd(ev.Pipeline, *s, now) + h.Failures.StepEnded(h.Store, p, *s) 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) } + // The step's end is written and said before its log entry is appended, so a log that refuses ends no step (§486, §606). + h.appendUnlogged(ev.Pipeline) + h.Ager.Age() + h.Fills.Check() return p, h.said(p, asked, ended) } } @@ -255,14 +275,10 @@ func (h *Hearer) groupEnded(p seat.Pipeline) error { 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) - } +// endEntry is the step log's entry of a step's end, with the keys of the kept results it wrote or carried. +func endEntry(key string, s seat.StepResult, at time.Time, kept ...string) seat.LogEntry { + return seat.LogEntry{Pipeline: key, Step: s.Step, Artifact: s.Artifact, What: seat.LogEntryWhatEnded, Outcome: s.Outcome, + Output: s.Output, Kept: kept, At: at} } func (h *Hearer) logf(format string, args ...any) { @@ -284,7 +300,11 @@ func (h *Hearer) say(event string, p seat.Pipeline, step *seat.StepResult) error 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)) + e := pipelineEnded(p) + if h.Widens != nil { + e.Widens = h.Widens(p) + } + return h.Say(subject, e) 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}) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go index 7f8e832e..23506089 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go @@ -335,12 +335,12 @@ func TestTheBoundIsArmedBeforeAnAskIsPut(t *testing.T) { asker := askerFunc(func(accept, id string, _ any) error { order = append(order, "put "+id); return nil }) arm := Arm(func(ask string) error { order = append(order, "arm "+ask); return nil }) asks := []Asked{{Seat: nodebuildgo.Seat, ID: "step-1", Ask: 1}, {Seat: wants.Seat, ID: "step-2", Ask: 2}} - if err := askAll(asker, arm, "k", asks); err != nil || strings.Join(order, ",") != "arm step-1,put step-1,put step-2" { + if err := askAll(asker, arm, "k", asks, false); err != nil || strings.Join(order, ",") != "arm step-1,put step-1,put step-2" { t.Errorf("order %v, %v; want the build seat's bound armed first and mesh-wants' ask with none", order, err) } order = nil failing := Arm(func(string) error { return errors.New("bucket gone") }) - if err := askAll(asker, failing, "k", asks[:1]); err == nil || len(order) != 0 { + if err := askAll(asker, failing, "k", asks[:1], false); err == nil || len(order) != 0 { t.Errorf("an ask whose bound cannot be armed was put or said nothing: %v %v", order, err) } } @@ -348,3 +348,30 @@ func TestTheBoundIsArmedBeforeAnAskIsPut(t *testing.T) { type askerFunc func(accept, id string, body any) error func (f askerFunc) AskStep(accept, id string, body any) error { return f(accept, id, body) } + +// A recovery's steps are asked on the build seats' stream delivery, ahead of the first round; the seats with no such stream keep +// their own accept subject; every other commit's steps go on check (design 47 G §86, §415). +func TestARecoverysStepsAreAskedOnTheStreamDelivery(t *testing.T) { + asks := []Asked{{Seat: nodebuildgo.Seat, ID: "s1", Ask: 1}, {Seat: wants.Seat, ID: "s2", Ask: 2}} + for _, c := range []struct { + recovery bool + want string + }{{true, nodebuildgo.AcceptSubjectDeliveryStep + "," + wants.AcceptSubjectStep}, {false, nodebuildgo.AcceptSubjectCheckStep + "," + wants.AcceptSubjectStep}} { + var to []string + asker := askerFunc(func(accept, _ string, _ any) error { to = append(to, accept); return nil }) + if err := askAll(asker, nil, "k", asks, c.recovery); err != nil || strings.Join(to, ",") != c.want { + t.Errorf("recovery %v: asked on %v (%v), want %s", c.recovery, to, err, c.want) + } + } + // Starter.recovered reads the commit; a failed read asks on check and says so. + var said []string + s := &Starter{Logf: func(f string, a ...any) { said = append(said, f) }, Recovered: func(c string) (bool, error) { + if c == "bad" { + return false, errors.New("unread") + } + return c == "rec", nil + }} + if !s.recovered("k", "rec") || s.recovered("k", "other") || s.recovered("k", "bad") || len(said) != 1 { + t.Errorf("recovered: said %v", said) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/unlogged.go b/modules/mesh-delivery-pipeline/internal/pipeline/unlogged.go new file mode 100644 index 00000000..95a1f882 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/unlogged.go @@ -0,0 +1,84 @@ +package pipeline + +import ( + "errors" + "slices" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// sameEntry says whether two step-log entries are one: the same pipeline, step, artifact, end or start and time. +func sameEntry(a, b seat.LogEntry) bool { + return a.Pipeline == b.Pipeline && a.Step == b.Step && a.Artifact == b.Artifact && a.What == b.What && a.At.Equal(b.At) +} + +// appendUnlogged appends the unlogged entries of pipeline key to the step log in their order and clears each appended one in a +// write of the pipeline (design 47 G §486, §606): a log that refuses an append keeps the entry in the pipeline's key, says the +// log full, and leaves it to the next step's end, the next start or the minute's check while the log is full. With readBack the +// entry is appended only after the log's last entry of that pipeline is read back and is not already it (§549). It says how +// many it appended. +func (h *Hearer) appendUnlogged(key string, readBack ...bool) int { + if h.Log == nil { + return 0 + } + appended := 0 + var done []seat.LogEntry + for { + p, rev, err := h.Store.Get(key) + if err != nil || len(p.Unlogged) == 0 { + return appended + } + for _, e := range p.Unlogged { + if slices.ContainsFunc(done, func(d seat.LogEntry) bool { return sameEntry(d, e) }) { + continue + } + if len(readBack) > 0 && readBack[0] && h.LogLast != nil { + if last, err := h.LogLast(e.Pipeline); err == nil && last != nil && sameEntry(*last, e) { + done = append(done, e) + continue + } else if err != nil { + h.logf("the step log's last entry of %s is not read, so %s of it is not appended yet: %v", e.Pipeline, e.Step, err) + break + } + } + if err := h.Log(e); err != nil { + h.logf("the step log refuses the end of %s of %s, kept unlogged in its pipeline: %v", e.Step, e.Pipeline, err) + h.Fills.MarkFull(seat.StateStepLog) + break + } + done = append(done, e) + appended++ + } + if len(done) == 0 { + return appended + } + rest := p.Unlogged[:0:0] + for _, e := range p.Unlogged { + if !slices.ContainsFunc(done, func(d seat.LogEntry) bool { return sameEntry(d, e) }) { + rest = append(rest, e) + } + } + p.Unlogged = rest + if _, err := h.Store.Update(key, p, rev); errors.Is(err, ErrChanged) { + continue + } else if err != nil { + h.logf("the appended entries of %s are not cleared, so they are read back at the next start: %v", key, err) + } + return appended + } +} + +// AppendUnlogged appends the unlogged entries of every pipeline, each after reading back the log's last entry of its pipeline (§549): +// at the runner's start, and each minute while the step log is full (§606). It says how many it appended. +func (h *Hearer) AppendUnlogged() int { + pkeys, err := h.Store.Keys("*.*.*") + if err != nil { + h.logf("the pipelines are not listed, so no unlogged entry is appended: %v", err) + return 0 + } + n := 0 + for _, k := range pkeys { + n += h.appendUnlogged(k, true) + } + return n +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/unlogged_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/unlogged_test.go new file mode 100644 index 00000000..32b5d726 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/unlogged_test.go @@ -0,0 +1,59 @@ +package pipeline + +import ( + "errors" + "testing" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// A step's end is written in the pipeline with its log entry as `unlogged` in one write and said before the entry is appended: a +// log that refuses the append ends no step, keeps the entry in the pipeline's key and says the log full; the next start appends +// it, after reading back the log's last entry of that pipeline, so a stop between the append and the clear logs it once (design +// 47 G §486, §606, §549). +func TestAStepEndedBeforeItsLogEntryIsAppended(t *testing.T) { + h, store, said, _, _, _ := fixture() + var logged []seat.LogEntry + full := true + h.Log = func(e seat.LogEntry) error { + if full { + return errors.New("maximum bytes exceeded") + } + logged = append(logged, e) + return nil + } + var last *seat.LogEntry + h.LogLast = func(string) (*seat.LogEntry, error) { return last, nil } + var events []string + h.Fills = &Fills{Say: func(event string, f seat.Fill) error { events = append(events, event+":"+f.Store); return nil }} + + if _, err := h.OnStepEnded(goSeat, goSeat.Prefix+"n1", ended("a1", "lint", "passed")); err != nil { + t.Fatalf("a refused log append ended no step's end: %v", err) + } + p := store[pkey].p + if p.Steps[0].Outcome != "passed" || len(p.Unlogged) != 1 || p.Unlogged[0].Step != "lint" || len(logged) != 0 { + t.Fatalf("the end is written with its entry unlogged: steps %+v unlogged %+v logged %v", p.Steps[0], p.Unlogged, logged) + } + if len(*said) == 0 || (*said)[0][:len(seat.EventStepEnded)] != seat.EventStepEnded { + t.Errorf("step-ended is not said first: %v", *said) + } + if len(events) != 1 || events[0] != "full:"+seat.StateStepLog || len(h.Fills.Full()) != 1 { + t.Errorf("the log's refusal says %v, full %v; want full of the step log", events, h.Fills.Full()) + } + // The log has room again: the next start appends it in order, once, and clears it. + full = false + if n := h.AppendUnlogged(); n != 1 || len(logged) != 1 || len(store[pkey].p.Unlogged) != 0 { + t.Fatalf("appended %d, logged %v, unlogged %v", n, logged, store[pkey].p.Unlogged) + } + // A stop between the append and the clear: the entry is in the log already, so a start appends nothing twice. + p = store[pkey].p + p.Unlogged = []seat.LogEntry{logged[0]} + store[pkey] = struct { + p seat.Pipeline + rev uint64 + }{p, store[pkey].rev} + last = &logged[0] + if n := h.AppendUnlogged(); n != 0 || len(logged) != 1 || len(store[pkey].p.Unlogged) != 0 { + t.Errorf("a logged entry read back was appended %d times more (%d logged, unlogged %v)", n, len(logged), store[pkey].p.Unlogged) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go index e3caef80..1c5bc55e 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/verbs.go @@ -70,18 +70,21 @@ func (v Verbs) change(key, event string, record func() error, mutate func(p *sea return p, i, err } ended := endPipeline(&p, now) + if i >= 0 && event == seat.EventStepEnded { + p.Unlogged = append(p.Unlogged, endEntry(key, p.Steps[i], 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) } + if event == seat.EventStepEnded { + h.appendUnlogged(key) + } } return p, i, h.said(p, asked, ended) } @@ -281,7 +284,7 @@ func (v Verbs) Rerun(caller string, in seat.RerunInput) (seat.RerunAnswer, error } 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 { + if err := askAll(v.Starter.Steps, v.Starter.Arm, in.Pipeline, []Asked{*ask}, v.Starter.recovered(in.Pipeline, p.Commit)); 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) } } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/widens.go b/modules/mesh-delivery-pipeline/internal/pipeline/widens.go new file mode 100644 index 00000000..971a2698 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/widens.go @@ -0,0 +1,179 @@ +package pipeline + +import ( + "fmt" + "sort" + "time" + + "git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshgrants" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// ComposerSeat is the seat of the composer, the node-engine's mesh-compose (design 47 G A-167, A-220): a part serving a step. +const ComposerSeat = "node-engine" + +// Widener computes a pipeline's `widens` (design 47 G §71, §275, §335, §441, A-207, A-215): the pipeline's own revision, the +// candidate, against the branch's newest revision, both read through the catalog seat's `revision`, and the permissions from the +// grants part's `grants-for`; no other verb. A widening is said, never refused: a read that fails says `could not run`. +type Widener struct { + // Revision reads one catalog revision of a watched branch whole; revision 0 is the branch's newest. + Revision func(branch string, revision int64) (Half, error) + // Holder is the module that holds a seat, from the seat holders' one reading; an error when it cannot be read or none holds it. + Holder func(seat string) (string, error) + // GrantsFor asks the seat mesh-grants' verb grants-for for the candidate's source revision on the branch. + GrantsFor func(branch string, source int64) (meshgrants.GrantsFor, error) +} + +// Widens names each part serving a step the change alters (the holders of the runner's seat, the catalog's, each of the pipeline's +// steps' and the composer's, whose entry's fingerprint moved), each acting rule of the pipeline's module that is gained, matches +// more or is held longer, each own secret or credential provision it names that it did not, and each grant the grants part adds. +func (w *Widener) Widens(p seat.Pipeline) []seat.Widening { + if w == nil || w.Revision == nil { + return nil + } + could := func(name, said string) []seat.Widening { + return []seat.Widening{{What: seat.WideningWhatCouldNotRun, Name: name, Said: said}} + } + cand, err := w.Revision(p.Branch, p.Revision) + if err != nil { + return could("mesh-catalog", fmt.Sprintf("the candidate's revision %d is not read: %v", p.Revision, err)) + } + newest, err := w.Revision(p.Branch, 0) + if err != nil { + return could("mesh-catalog", fmt.Sprintf("the branch's newest revision is not read: %v", err)) + } + var out []seat.Widening + for _, s := range servingSeats(p) { + holder, err := w.Holder(s) + if err != nil { + out = append(out, seat.Widening{What: seat.WideningWhatCouldNotRun, Name: s, Said: fmt.Sprintf("the holder of %s is not read: %v", s, err)}) + continue + } + before, after := newest.Entries[holder], cand.Entries[holder] + if after.Fp != "" && after.Fp != before.Fp { + out = append(out, seat.Widening{What: seat.WideningWhatServingPart, Name: holder, + Said: fmt.Sprintf("serves %s; its fingerprint is %s, was %s", s, after.Fp, before.Fp)}) + } + } + out = append(out, ruleWidens(p.Module, newest.Entries[p.Module], cand.Entries[p.Module])...) + out = append(out, credentialWidens(p.Module, newest.Entries[p.Module], cand.Entries[p.Module])...) + if w.GrantsFor != nil { + g, err := w.GrantsFor(p.Branch, p.Source) + switch { + case err != nil: + out = append(out, seat.Widening{What: seat.WideningWhatCouldNotRun, Name: "mesh-grants", Said: err.Error()}) + case g.Error != "": + out = append(out, seat.Widening{What: seat.WideningWhatCouldNotRun, Name: "mesh-grants", Said: g.Error}) + default: + for _, m := range g.Modules { + if m.Module != p.Module { + continue + } + for _, a := range m.Added { + out = append(out, seat.Widening{What: seat.WideningWhatGrant, Name: m.Module + " " + a.Subject + " " + a.Verb, + Said: "the grants part would compose it, and the branch's newest does not hold it"}) + } + } + } + } + order := map[string]int{seat.WideningWhatServingPart: 0, seat.WideningWhatActingRule: 1, seat.WideningWhatCredential: 2, + seat.WideningWhatGrant: 3, seat.WideningWhatCouldNotRun: 4} + sort.SliceStable(out, func(i, j int) bool { + if order[out[i].What] != order[out[j].What] { + return order[out[i].What] < order[out[j].What] + } + return out[i].Name < out[j].Name + }) + return out +} + +// servingSeats are the seats whose holders serve a step of p (A-220): the runner's, the catalog's, the composer's and each step's. +func servingSeats(p seat.Pipeline) []string { + set := map[string]bool{seat.Seat: true, catalog.Seat: true, ComposerSeat: true} + for _, s := range p.Steps { + if s.Seat != "" { + set[s.Seat] = true + } + } + var out []string + for s := range set { + out = append(out, s) + } + sort.Strings(out) + return out +} + +// ruleWidens names each acting rule of the candidate's entry that the newest entry has not, or that now names more entries +// (`every` where it named a language) or holds longer (A-215). +func ruleWidens(module string, before, after catalog.CatalogEntry) []seat.Widening { + on := func(s string) string { + if s == "" { + return "every" + } + return s + } + var out []seat.Widening + for _, r := range after.Acts { + name := module + " " + r.Seat + "." + r.Step + var was *catalog.ActingRule + for i := range before.Acts { + if before.Acts[i].Seat == r.Seat && before.Acts[i].Step == r.Step { + was = &before.Acts[i] + } + } + switch { + case was == nil: + out = append(out, seat.Widening{What: seat.WideningWhatActingRule, Name: name, Said: "the rule is gained"}) + case on(r.On) == "every" && on(was.On) != "every": + out = append(out, seat.Widening{What: seat.WideningWhatActingRule, Name: name, Said: "the rule acts on every entry, it acted on " + was.On}) + case longer(r.Bound, was.Bound): + out = append(out, seat.Widening{What: seat.WideningWhatActingRule, Name: name, Said: "the bound is " + r.Bound + ", it was " + was.Bound}) + } + } + return out +} + +func longer(a, b string) bool { + x, err1 := time.ParseDuration(a) + y, err2 := time.ParseDuration(b) + return err1 == nil && err2 == nil && x > y +} + +// credentialWidens names each own secret and each provision with a credential of its own that the candidate's manifest names and +// the newest entry's did not (A-215). +func credentialWidens(module string, before, after catalog.CatalogEntry) []seat.Widening { + was := map[string]bool{} + for _, n := range credentialsOf(before.Manifest) { + was[n] = true + } + var out []seat.Widening + for _, n := range credentialsOf(after.Manifest) { + if !was[n] { + out = append(out, seat.Widening{What: seat.WideningWhatCredential, Name: module + " " + n, Said: "the manifest names it, and the branch's newest does not"}) + } + } + return out +} + +// credentialsOf are the names of the own secrets a manifest declares and of the provisions that carry a credential: an own secret +// of the provider, one issued by software or one only a person gives. +func credentialsOf(manifest map[string]any) []string { + var out []string + if secrets, ok := manifest["secrets"].(map[string]any); ok { + for n := range secrets { + out = append(out, n) + } + } + if provides, ok := manifest["provides"].([]any); ok { + for _, o := range provides { + if m, ok := o.(map[string]any); ok && m["credential"] != nil { + if n, ok := m["name"].(string); ok { + out = append(out, n) + } + } + } + } + sort.Strings(out) + return out +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/widens_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/widens_test.go new file mode 100644 index 00000000..fd55b595 --- /dev/null +++ b/modules/mesh-delivery-pipeline/internal/pipeline/widens_test.go @@ -0,0 +1,108 @@ +package pipeline + +import ( + "errors" + "strings" + "testing" + + "git.novox.be/novox/mesh-contracts/identity-and-access/go/seats/meshgrants" + catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +func entry(module, fp string, acts []catalog.ActingRule, manifest map[string]any) catalog.CatalogEntry { + return catalog.CatalogEntry{Module: module, Fp: fp, Acts: acts, Manifest: manifest} +} + +func halfOf(rev int64, es ...catalog.CatalogEntry) Half { + h := Half{Revision: rev, Entries: map[string]catalog.CatalogEntry{}} + for _, e := range es { + h.Entries[e.Module] = e + } + return h +} + +// widens compares the pipeline's own revision with the branch's newest: the serving parts found by the holder of the seats +// (the runner, the catalog, each step's seat, the node-engine) by fingerprint, an acting rule gained, matching more or held longer, a +// credential named that was not, and the grants the grants part adds; each is said by name, a grants part that does not answer says +// `could not run` and nothing is refused (design 47 G §71, §275, §441, A-207, A-215, A-220). +func TestWidensIsTheCandidateAgainstTheNewest(t *testing.T) { + p := seat.Pipeline{Pipeline: "main.8.app", Module: "app", Branch: "main", Source: 31, Revision: 8, + Steps: []seat.StepResult{{Step: "unit", Seat: "node-build-go"}, {Step: "lint", Seat: "node-build-go"}, {Step: "approval"}}} + holders := map[string]string{seat.Seat: "runner", catalog.Seat: "indexer", "node-build-go": "build-go", "node-engine": "host"} + rule := func(on, bound string) []catalog.ActingRule { + return []catalog.ActingRule{{Seat: "node-build-go", Step: "check", On: on, Bound: bound}} + } + newest := halfOf(9, + entry("runner", "fp-1", nil, nil), entry("indexer", "fp-1", nil, nil), entry("build-go", "fp-1", nil, nil), entry("host", "fp-1", nil, nil), + entry("app", "fp-1", rule("go", "10m"), map[string]any{"secrets": map[string]any{"token": "/x"}, "provides": []any{"db"}})) + cases := []struct { + name string + cand Half + grants meshgrants.GrantsFor + grantsErr error + want []string + }{ + {"nothing changes", newest, meshgrants.GrantsFor{}, nil, nil}, + {"a serving part's fingerprint moves", halfOf(8, entry("build-go", "fp-2", nil, nil), entry("host", "fp-1", nil, nil), entry("indexer", "fp-1", nil, nil), + entry("runner", "fp-1", nil, nil), newest.Entries["app"]), meshgrants.GrantsFor{}, nil, []string{"serving part build-go"}}, + {"an acting rule is gained", halfOf(8, entry("app", "fp-1", append(rule("go", "10m"), catalog.ActingRule{Seat: "x", Step: "grant"}), newest.Entries["app"].Manifest)), + meshgrants.GrantsFor{}, nil, []string{"acting rule app x.grant"}}, + {"an acting rule matches more", halfOf(8, entry("app", "fp-1", rule("every", "10m"), newest.Entries["app"].Manifest)), meshgrants.GrantsFor{}, nil, + []string{"acting rule app node-build-go.check"}}, + {"an acting rule is held longer", halfOf(8, entry("app", "fp-1", rule("go", "1h"), newest.Entries["app"].Manifest)), meshgrants.GrantsFor{}, nil, + []string{"acting rule app node-build-go.check"}}, + {"an acting rule is held shorter", halfOf(8, entry("app", "fp-1", rule("go", "5m"), newest.Entries["app"].Manifest)), meshgrants.GrantsFor{}, nil, nil}, + {"a secret and a credential provision are named", halfOf(8, entry("app", "fp-1", rule("go", "10m"), map[string]any{ + "secrets": map[string]any{"token": "/x", "key": map[string]any{"path": "/k"}}, + "provides": []any{"db", map[string]any{"name": "vault", "credential": map[string]any{"own": "key"}}}})), + meshgrants.GrantsFor{}, nil, []string{"credential app key", "credential app vault"}}, + {"a grant is added", newest, meshgrants.GrantsFor{Modules: []meshgrants.GrantsForModulesItem{ + {Module: "app", Added: []meshgrants.Grant{{Subject: "mesh.seat.x", Verb: "do"}}}, {Module: "other", Added: []meshgrants.Grant{{Subject: "s", Verb: "v"}}}}}, + nil, []string{"grant app mesh.seat.x do"}}, + {"the grants part does not answer", newest, meshgrants.GrantsFor{}, errors.New("no responders"), []string{"could not run mesh-grants"}}, + {"the grants part cannot read the candidate", newest, meshgrants.GrantsFor{Error: "the catalog half is not held"}, nil, []string{"could not run mesh-grants"}}, + } + for _, c := range cases { + w := &Widener{ + Revision: func(_ string, rev int64) (Half, error) { + if rev == 0 { + return newest, nil + } + if rev != 8 { + t.Errorf("%s: the candidate asked as revision %d", c.name, rev) + } + return c.cand, nil + }, + Holder: func(s string) (string, error) { return holders[s], nil }, + GrantsFor: func(string, int64) (meshgrants.GrantsFor, error) { return c.grants, c.grantsErr }, + } + var got []string + for _, x := range w.Widens(p) { + got = append(got, x.What+" "+x.Name) + } + if strings.Join(got, ";") != strings.Join(c.want, ";") { + t.Errorf("%s: widens %q, want %q", c.name, got, c.want) + } + } +} + +// pipeline-ended carries the widens the runner computed, and says it whether or not anything widens (design 47 G §71, A-207). +func TestPipelineEndedCarriesWidens(t *testing.T) { + var body any + h := &Hearer{Say: func(_ string, b any) error { body = b; return nil }, + Widens: func(p seat.Pipeline) []seat.Widening { + return []seat.Widening{{What: seat.WideningWhatGrant, Name: p.Module + " s v"}} + }} + if err := h.say(seat.EventPipelineEnded, seat.Pipeline{Module: "app", Outcome: seat.PipelineOutcomePassed}, nil); err != nil { + t.Fatal(err) + } + if e, ok := body.(seat.PipelineEnded); !ok || len(e.Widens) != 1 || e.Widens[0].Name != "app s v" { + t.Errorf("pipeline-ended says %+v", body) + } + h.Widens = nil + _ = h.say(seat.EventPipelineEnded, seat.Pipeline{Module: "app"}, nil) + if e := body.(seat.PipelineEnded); len(e.Widens) != 0 { + t.Errorf("a runner with no widener said %v", e.Widens) + } +} diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index bcf8f548..44c9368d 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -54,7 +54,7 @@ func TestManifest(t *testing.T) { {"claims the seat serving every verb of its definition", len(m.Claims) == 1 && m.Claims[0].Name == seat.Seat && equalSets(m.Claims[0].Serves, seat.Verbs())}, // §27, §85, §330, §355, §77: the old state sections. - {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "bounds", "toolchains", "findings", "first-views", "consumers"})}, + {"old state sections", equalSets(states, []string{seat.StatePipelines, seat.StateKept, "aged", "recoveries", "bounds", "toolchains", "findings", "first-views", "consumers"})}, // §535, §647, A-77, A-104: an ask no holder takes is bound by a key with its own time to live. {"bounds is key-ttl", keyTTL["bounds"]}, // RN25 (§469, §85): an old state section declares its cap, pipelines 256 MiB and kept 128 MiB. @@ -78,20 +78,26 @@ func TestManifest(t *testing.T) { "mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-ended", "mesh-delivery-builder.build.ended.**", "operator-channel.decided", "node-build-go.step-ended", "node-build-node.step-ended", "node-build-image.step-ended", "node-build-go.ask-taken", "node-build-node.ask-taken", "node-build-image.ask-taken", "node-engine.running", "mesh-wants.step-ended", "mesh-catalog.registration-changed", - "mesh-catalog.catalog-revision", "mesh-conditions.findings-asked"})}, + "mesh-catalog.catalog-revision", "mesh-conditions.findings-asked", "mesh-delivery.recovery-merged"})}, // RN2 (§21, §195, §200, §77): the registrations and the catalog half asked by their verbs, the first views read and // asked again by hand (a hand act). {"invokes", equalSets(m.Invokes, []string{"seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act", // RN3 (§53, §334): a publication's dependents behind. "seat:mesh-catalog.outdated", + // RN10 (§71, §441, A-207): the permissions a change widens, asked of the grants part. + "seat:mesh-grants.grants-for", + // RN11 (§33, A-191): the author of a failed head, read as the forge records it. + "seat:git.pull", // A-74, A-89: who holds the seat operator-channel, whose state asks holds the router's record of the approval ask. "seat:mesh-module-assigner.assignments", // A-102 (§22, §24): the accepted kinds the runner puts its steps on. "seat:mesh-builds.step", "seat:mesh-wants.step", "seat:node-build-go.check.step", "seat:node-build-node.check.step", - "seat:node-build-image.check.step"})}, + "seat:node-build-image.check.step", + // RN22 (§86, §415, A-207): a recovery's steps on the build seats' stream delivery. + "seat:node-build-go.delivery.step", "seat:node-build-node.delivery.step", "seat:node-build-image.delivery.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", "runner_release"})}, + {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health", "runner_waits", "runner_release", "runner_unlogged", "runner_unlogged_append"})}, // 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 1bb8273d..e428eb86 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -28,7 +28,9 @@ "runner_first_view_again", "runner_health", "runner_waits", - "runner_release" + "runner_release", + "runner_unlogged", + "runner_unlogged_append" ], "consumes": [ "mesh-delivery-indexer.take-in.taken", @@ -47,19 +49,25 @@ "mesh-wants.step-ended", "mesh-catalog.registration-changed", "mesh-catalog.catalog-revision", - "mesh-conditions.findings-asked" + "mesh-conditions.findings-asked", + "mesh-delivery.recovery-merged" ], "invokes": [ "seat:mesh-catalog.registrations", "seat:mesh-catalog.revision", "seat:mesh-hand-acts.hand-act", "seat:mesh-catalog.outdated", + "seat:git.pull", "seat:mesh-module-assigner.assignments", + "seat:mesh-grants.grants-for", "seat:mesh-builds.step", "seat:mesh-wants.step", "seat:node-build-go.check.step", "seat:node-build-node.check.step", - "seat:node-build-image.check.step" + "seat:node-build-image.check.step", + "seat:node-build-go.delivery.step", + "seat:node-build-node.delivery.step", + "seat:node-build-image.delivery.step" ], "uses": [ "operator-channel" @@ -79,6 +87,10 @@ "name": "aged", "history": 1 }, + { + "name": "recoveries", + "history": 1 + }, { "name": "toolchains", "history": 1