From d44c45c9b97b412444e0096c76f6d9df37c61dc8 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:02:16 +0100 Subject: [PATCH] A-206: the bound's expiry writes the finding pipeline-waits through the runner's finding writer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ReleaseExpired says each pipeline it started through the starter's Late, and the process writes the finding pipeline-waits (scope module, id the pipeline's key `..`) naming the awaited runner and since when. Builds: design 47 G §266, §269, A-206 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-delivery-pipeline/findingwriter.go | 24 ++++++++++++++- .../findingwriter_test.go | 29 +++++++++++++++++++ .../cmd/mesh-delivery-pipeline/main.go | 10 ++++++- .../cmd/mesh-delivery-pipeline/views.go | 4 ++- .../internal/pipeline/hold.go | 19 +++++++++--- .../internal/pipeline/hold_test.go | 29 +++++++++++++++++++ .../internal/pipeline/start.go | 2 ++ 7 files changed, 110 insertions(+), 7 deletions(-) create mode 100644 modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go index 0fedcad4..b8d2d35e 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter.go @@ -2,6 +2,8 @@ package main import ( "context" + "fmt" + "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -9,12 +11,32 @@ import ( condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" ) +// 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] + +// 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). +const KindPipelineWaits = "pipeline-waits" + +// lateFinding is the finding pipeline-waits of one pipeline the bound's expiry started: scope module, id its key +// `..`, naming the fingerprint of the new runner it waited for and since when. +func lateFinding(p seat.Pipeline, held seat.Held, runner, node string, at time.Time) condseat.Finding { + return condseat.Finding{Scope: views.FindingScope, ID: p.Pipeline, Kind: KindPipelineWaits, Severity: condseat.FindingSeverityWarning, + Headline: "the new pipeline runner did not run healthy in time", + Summary: fmt.Sprintf("the pipeline %s was held since %s for the runner at %s to run healthy; the bound passed and the running runner runs it", + p.Pipeline, held.Since.UTC().Format(time.RFC3339), held.Fingerprint), + Source: "bound of the wait for the new runner", Observer: runner, At: at, + 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 // 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) { +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]{ 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) { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go new file mode 100644 index 00000000..8d9dd3fc --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/findingwriter_test.go @@ -0,0 +1,29 @@ +package main + +import ( + "strings" + "testing" + "time" + + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/views" + + seat "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshpipelinerunner" +) + +// The bound's expiry's finding is pipeline-waits, scope module, id the pipeline's key, naming the awaited fingerprint and since +// when (design 47 G §266, A-206); it passes the writer's own check of a finding. +func TestLateFindingIsPipelineWaitsOfTheModuleScope(t *testing.T) { + since := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC) + f := lateFinding(seat.Pipeline{Pipeline: "main.4.app", Module: "app", Branch: "main"}, seat.Held{Since: since, Fingerprint: "src2:new"}, + "mesh-delivery-pipeline", "n1", since.Add(30*time.Minute)) + if f.Kind != "pipeline-waits" || f.Scope != views.FindingScope || f.ID != "main.4.app" || f.Severity != condseat.FindingSeverityWarning { + t.Errorf("finding %+v", f) + } + if !strings.Contains(f.Summary, "src2:new") || !strings.Contains(f.Summary, "2026-11-11T12:00:00Z") { + t.Errorf("summary %q", f.Summary) + } + if err := views.CheckFinding(f); err != nil { + t.Error(err) + } +} diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index f68f0641..0be86b7f 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -13,6 +13,7 @@ import ( "os/signal" "sync/atomic" "syscall" + "time" "github.com/nats-io/nats.go" @@ -108,7 +109,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Arm: armBound(ctx, bounds), Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains), // The runner's own module and fingerprint (A-149): a take-in giving it another holds the pipelines it starts (§65, A-195). Self: cred.Module, SelfFp: getenv("MESH_FINGERPRINT"), Awaiting: pipeline.Awaiting(pipelineStore{ctx: ctx, kv: pipelines}), - Hold: armHeld(ctx, bounds)} + Hold: armHeld(ctx, bounds), + // The bound's expiry writes the finding pipeline-waits for each pipeline it started (§266, A-206). + Late: func(p seat.Pipeline, held seat.Held) { + module, node := rv.Findings.ModuleNode() + 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) + } + }} resume := func(branch string) { half, ok := rv.Catalog.View(branch) if !ok { diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go index 363c34db..1236cf75 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/views.go @@ -20,6 +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 *findingsWriter } // 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. diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go index 04148002..000f7f5a 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/hold.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold.go @@ -149,18 +149,29 @@ func (s *Starter) Release(store Updater, fp, how string, view func(branch string } // ReleaseExpired starts the pipelines held for the fingerprint whose bound key expired (§266, A-195): the 30 minutes passed -// without the new runner running healthy, so the running runner runs them. +// without the new runner running healthy, so the running runner runs them, and each started is said late through Late (A-206). func (s *Starter) ReleaseExpired(store Updater, key string, view func(branch string) (Half, bool)) (int, error) { pkeys, err := store.Keys("*.*.*") if err != nil { return 0, fmt.Errorf("the pipelines are not listed: %w", err) } + fp := "" + held := map[string]seat.Held{} for _, k := range pkeys { - if p, _, err := store.Get(k); err == nil && p.Held != nil && HeldKey(p.Held.Fingerprint) == key { - return s.Release(store, p.Held.Fingerprint, "released", view) + if p, _, err := store.Get(k); err == nil && p.Held != nil && p.Outcome == seat.PipelineOutcomeRunning && HeldKey(p.Held.Fingerprint) == key { + fp, held[k] = p.Held.Fingerprint, *p.Held } } - return 0, nil + if fp == "" { + return 0, nil + } + n, err := s.Release(store, fp, "released", view) + for k, h := range held { + if p, _, gerr := store.Get(k); gerr == nil && p.Held == nil && s.Late != nil { + s.Late(p, h) + } + } + return n, err } // heldFingerprints are the fingerprints of the runner the held pipelines in store await, each once. diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go index 19628ec6..b609a473 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/hold_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "strings" "testing" + "time" 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" @@ -125,3 +126,31 @@ func TestReleaseVerbStartsTheHeldPipelinesByHand(t *testing.T) { t.Errorf("said %v", said) } } + +// The bound's expiry starts the held pipelines and says each as late through the starter's Late, with the awaited fingerprint and +// since when; a pipeline that stays held (no view) is not said (design 47 G §266, A-206). +func TestExpiredBoundSaysEachReleasedPipelineLate(t *testing.T) { + reg, entries := placeFixture() + store := memCreated{memStore{}} + since := time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC) + for _, k := range []string{"main.4.app", "dev.4.app"} { + store.memStore[k] = struct { + p seat.Pipeline + rev uint64 + }{seat.Pipeline{Pipeline: k, Branch: strings.Split(k, ".")[0], Module: "app", Outcome: seat.PipelineOutcomeRunning, + Held: &seat.Held{Since: since, Fingerprint: "src2:new"}}, 1} + } + var late []string + s := &Starter{Pipelines: store, Steps: &stepAsks{}, Name: "the-runner", Say: func(string, any) error { return nil }, + Registration: func(string) (catalog.Registration, bool) { return reg, true }, + Late: func(p seat.Pipeline, h seat.Held) { + late = append(late, p.Pipeline+"@"+h.Fingerprint+"@"+h.Since.Format(time.RFC3339)) + }} + view := func(b string) (Half, bool) { return Half{Entries: entries}, b == "main" } + if n, err := s.ReleaseExpired(store.memStore, HeldKey("src2:new"), view); err == nil || n != 1 { + t.Fatalf("release = %d, %v (the pipeline of a branch with no view stays held, an error)", n, err) + } + if len(late) != 1 || late[0] != "main.4.app@src2:new@2026-11-11T12:00:00Z" { + t.Errorf("late %v", late) + } +} diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/start.go b/modules/mesh-delivery-pipeline/internal/pipeline/start.go index 42f23f01..f35288ba 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/start.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/start.go @@ -78,6 +78,8 @@ type Starter struct { Self, SelfFp string Awaiting func() (string, bool) Hold func(fp string) error + // 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) } func newPipeline(b Begin, key string, now time.Time) seat.Pipeline {