diff --git a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go index 8cadf686..136a6ec4 100644 --- a/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go +++ b/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go @@ -460,6 +460,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) 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 4d82a907..51433279 100644 --- a/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json +++ b/modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json @@ -1025,6 +1025,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*)", 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 579c3a87..c96b0029 100644 --- a/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts +++ b/modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts @@ -219,6 +219,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) */ 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 4b0559b7..444dc4b8 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 @@ -111,6 +111,8 @@ func HealthTool(pipelines, consumers pipeline.Kept, now func() time.Time) stdio. func main() { r := pipeline.Reads{Pipelines: kept{stdio.State(seat.StatePipelines)}, Kept: kept{stdio.State(seat.StateKept)}} tools := append(Tools(r), FirstViewTools(stdio.State(views.StatusState), meshhandactsclient.Recorder{Ask: stdio.Ask}, time.Now)...) + tools = append(tools, 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 bb3dd146..fddd68ee 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go @@ -2,6 +2,7 @@ package main import ( "encoding/json" + "errors" "strings" "testing" @@ -28,8 +29,32 @@ func TestTools(t *testing.T) { } // RN25 (§468, §76; A-164): the health check is the bundle's runner_health. names = append(names, HealthTool(none{}, none{}, nil).Name) - want := "mesh-pipeline-runner.show mesh-pipeline-runner.list mesh-pipeline-runner.stands mesh-pipeline-runner.why-carried mesh-pipeline-runner.kept runner_first_views runner_first_view_again runner_health" + // §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_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) + } +} 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 index e0dca28d..9a780571 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "strconv" "time" @@ -175,3 +176,39 @@ func storeFills(ctx context.Context, js jetstream.JetStream, cred process.Accoun 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/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index fbc8b195..acb10be9 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -116,6 +116,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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 + 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 @@ -161,6 +162,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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/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 f23d4781..918adf93 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/bounds.go @@ -77,19 +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) - h.Ager.Age() - h.Fills.Check() 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/fill.go b/modules/mesh-delivery-pipeline/internal/pipeline/fill.go index bee8d812..506ea84b 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/fill.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/fill.go @@ -75,6 +75,27 @@ func (f *Fills) Check() { } } +// 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 { diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go index 45c50d41..f94e9c6f 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/steps.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/steps.go @@ -94,6 +94,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 @@ -166,6 +168,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() @@ -180,11 +188,9 @@ func (h *Hearer) OnStepEnded(bs StepSeat, subject string, ev StepEnded) (seat.Pi } else if err != nil { return p, fmt.Errorf("pipeline %s cannot be written: %w", ev.Pipeline, err) } - var keptKeys []string if h.Keep != nil { for _, k := range ev.Kept { kept := seat.KeptResult{Digest: k.Digest, Level: k.Level, Package: k.Package, Pipeline: k.Pipeline, LastRead: &now} - keptKeys = append(keptKeys, seat.KeptByLevelDigestKey(keys.EscapeName(k.Level), keys.EscapeName(k.Digest))) if k.Toolchain != nil { kept.Toolchain = &seat.ToolchainUsed{Language: k.Toolchain.Language, Recipe: k.Toolchain.Recipe, Version: k.Toolchain.Version} } @@ -196,13 +202,14 @@ func (h *Hearer) OnStepEnded(bs StepSeat, subject string, ev StepEnded) (seat.Pi } } } - h.logEnd(ev.Pipeline, *s, now, keptKeys...) h.Failures.StepEnded(h.Store, p, *s) - h.Ager.Age() - h.Fills.Check() 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) } } @@ -222,14 +229,10 @@ func (h *Hearer) said(p seat.Pipeline, asked, ended bool) error { return nil } -func (h *Hearer) logEnd(key string, s seat.StepResult, at time.Time, kept ...string) { - 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, Kept: kept, 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) { 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 9f4c587c..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) } diff --git a/modules/mesh-delivery-pipeline/manifest_test.go b/modules/mesh-delivery-pipeline/manifest_test.go index 191c9b9d..f97f5e97 100644 --- a/modules/mesh-delivery-pipeline/manifest_test.go +++ b/modules/mesh-delivery-pipeline/manifest_test.go @@ -95,7 +95,7 @@ func TestManifest(t *testing.T) { "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"})}, + {"first-view tools", equalSets(m.Tools, []string{"runner_first_views", "runner_first_view_again", "runner_health", "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 6d2ca9cc..ad4d41c5 100644 --- a/modules/mesh-delivery-pipeline/module.json +++ b/modules/mesh-delivery-pipeline/module.json @@ -25,7 +25,9 @@ "tools": [ "runner_first_views", "runner_first_view_again", - "runner_health" + "runner_health", + "runner_unlogged", + "runner_unlogged_append" ], "consumes": [ "mesh-delivery-indexer.take-in.taken",