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

RN11: the runner says a failed head as the finding pipeline-failed

Browse Source 
A required step that failed or could not run writes the finding of its commit, owned by the commit's author as the forge
records it (read through the seat git's pull); one with no known author has no owner and says why, written at the next step
end of that commit; a newer revision passing the steps clears it. The runner's finding writer is the held writer.

Builds: design 47 G §33, §540, A-191
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 05:53:20 +01:00
parent 4b47e88369
commit 267bd0a9cb
11 changed files with 397 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
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
@@ -11,15 +11,13 @@ import (
"git.novox.be/novox/mesh-sdk/go/views"
)
// 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)) (*views.Writer[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) {
w, err := views.NewWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{
func findingWriter(ctx context.Context, cred process.Account, conn *nats.Conn, js jetstream.JetStream, state process.State, logf func(string, ...any)) (*views.HeldWriter[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent], error) {
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
+1
View File
@@ -101,6 +101,7 @@ 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.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.
starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say,
modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go
+3 -1
View File
@@ -20,6 +20,8 @@ import (
type runnerViews struct {
Registrations *meshcatalogclient.Registrations
Catalog *pipeline.Views
// Findings is the runner's held finding writer: it raises the finding pipeline-failed (§33).
Findings pipeline.Findings
}
// startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or
@@ -53,7 +55,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, conn *nats.Conn, cr
return nil, err
}
reader.Start(ctx, js, cred, status)
return &runnerViews{Registrations: regs, Catalog: cv}, nil
return &runnerViews{Registrations: regs, Catalog: cv, Findings: findings}, nil
}
// catalogView is a branch's meshcatalogclient.Catalog as the runner reads it.
modules/mesh-delivery-pipeline/go.mod
+2
View File
@@ -10,7 +10,9 @@ 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-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
modules/mesh-delivery-pipeline/internal/pipeline/bounds.go
+1
View File
@@ -84,6 +84,7 @@ func (h *Hearer) endNotTaken(key, ask string, bound time.Duration) error {
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)
}
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/steps.go
+3
View File
@@ -107,6 +107,8 @@ 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
}
func (h *Hearer) now() time.Time {
@@ -189,6 +191,7 @@ 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)
}
modules/mesh-delivery-pipeline/manifest_test.go
+2
View File
@@ -84,6 +84,8 @@ func TestManifest(t *testing.T) {
{"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",
// 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.
modules/mesh-delivery-pipeline/module.json
+1
View File
@@ -49,6 +49,7 @@
"seat:mesh-catalog.revision",
"seat:mesh-hand-acts.hand-act",
"seat:mesh-catalog.outdated",
"seat:git.pull",
"seat:mesh-module-assigner.assignments",
"seat:mesh-builds.step",
"seat:mesh-wants.step",