A-206: the bound's expiry writes the finding pipeline-waits through the runner's finding writer

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 `<branch>.<id>.<module>`) naming the awaited runner and since when. Builds: design 47 G §266, §269, A-206 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 06:02:16 +01:00
parent 6635dda18f
commit d44c45c9b9
7 changed files with 110 additions and 7 deletions
@@ -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
+// `<branch>.<id>.<module>`, 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) {
@@ -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)
+	}
+}
@@ -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 {
@@ -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.
@@ -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.
@@ -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)
+	}
+}
@@ -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 {