novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch 'feat/47-G-fix11-b1f' into feat/47-G-join11

Browse Source 
# Conflicts:
#	modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go
#	modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go
#	modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go
#	modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go
#	modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go
#	modules/mesh-delivery-pipeline/go.mod
#	modules/mesh-delivery-pipeline/internal/pipeline/read_test.go
#	modules/mesh-delivery-pipeline/internal/pipeline/start.go
#	modules/mesh-delivery-pipeline/manifest_test.go
#	modules/mesh-delivery-pipeline/module.json
This commit is contained in:
jschoubben
2026-11-12 10:30:22 +01:00
parent 4aaa805f2a a8314676c9
commit 5db642e161
41 changed files with 2284 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-contracts/go/seats/meshpipelinerunner/contract.go
+29 -2
View File
@@ -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:<hex> 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, <level>.<digest>, 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 <branch>.<revision>.<module>.
type PipelinesByBranchRevisionModuleValue = Pipeline
modules/mesh-delivery-contracts/seats/mesh-pipeline-runner/seat.json
+37
View File
@@ -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, <level>.<digest>, which the runner ages by their last-read (design 47 G sections 330, 470)"
}
},
"description": "one entry of the step log",
modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.d.ts
Vendored
+24
View File
@@ -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:<hex> 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, <level>.<digest>, 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 <branch>.<revision>.<module>. */
export type PipelinesByBranchRevisionModuleValue = Pipeline;
modules/mesh-delivery-contracts/ts/seats/meshpipelinerunner/index.js
+9
View File
@@ -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",
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main.go
+44 -1
View File
@@ -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)
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/main_test.go
+56 -1
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline-tools/unlogged.go
+105
View File
@@ -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
}
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/aged.go
+214
View File
@@ -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, `<pipeline>`
// 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()
}
}
}
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/asks.go
+25
View File
@@ -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
}
}
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go
+5 -7
View File
@@ -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
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go
+27 -1
View File
@@ -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)
}
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries.go
+55
View File
@@ -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()
})
}
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries_test.go
+35
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go
+12 -1
View File
@@ -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")
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/steps.go
+1 -3
View File
@@ -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 {
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go
+2 -1
View File
@@ -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
}
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/widens.go
+81
View File
@@ -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
}
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/widens_test.go
+35
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/go.mod
+5 -1
View File
@@ -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
modules/mesh-delivery-pipeline/internal/pipeline/aged.go
+177
View File
@@ -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
}
modules/mesh-delivery-pipeline/internal/pipeline/aged_test.go
+144
View File
@@ -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"])
}
}
modules/mesh-delivery-pipeline/internal/pipeline/approval.go
+5 -2
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/internal/pipeline/bounds.go
+5 -1
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/internal/pipeline/failed.go
+188
View File
@@ -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:<account>`; 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:<account>` 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
}
modules/mesh-delivery-pipeline/internal/pipeline/failed_test.go
+166
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/internal/pipeline/fill.go
+134
View File
@@ -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)
}
modules/mesh-delivery-pipeline/internal/pipeline/fill_test.go
+52
View File
@@ -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 }
modules/mesh-delivery-pipeline/internal/pipeline/hold.go
+1 -1
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/internal/pipeline/place.go
+12 -1
View File
@@ -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).
modules/mesh-delivery-pipeline/internal/pipeline/read.go
+32 -2
View File
@@ -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) {
modules/mesh-delivery-pipeline/internal/pipeline/read_test.go
+23
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/internal/pipeline/start.go
+22 -2
View File
@@ -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) {
modules/mesh-delivery-pipeline/internal/pipeline/steps.go
+31 -11
View File
@@ -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})
modules/mesh-delivery-pipeline/internal/pipeline/steps_test.go
+29 -2
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/internal/pipeline/unlogged.go
+84
View File
@@ -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
}
modules/mesh-delivery-pipeline/internal/pipeline/unlogged_test.go
+59
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/internal/pipeline/verbs.go
+7 -4
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/internal/pipeline/widens.go
+179
View File
@@ -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
}
modules/mesh-delivery-pipeline/internal/pipeline/widens_test.go
+108
View File
@@ -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)
}
}
modules/mesh-delivery-pipeline/manifest_test.go
+10 -4
View File
@@ -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.
modules/mesh-delivery-pipeline/module.json
+15 -3
View File
@@ -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