RN17 (unlogged): a step's end and its log entry are one write, appended after the end is said
The pipeline gains `unlogged`: a step's end is written in its key with its step-log entry in one compare-and-set, said as step-ended, then appended to the step log and cleared; a log that refuses an append ends no step, says full and starts no new pipeline; at its start the runner appends every unlogged entry in order after reading back the log's last entry of that pipeline, and once a minute while the log is full. The tools runner_unlogged and runner_unlogged_append (a hand act). Builds: design 47 G §486, §606, §549 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -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)
|
||||
|
||||
@@ -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*)",
|
||||
|
||||
@@ -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) */
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user