RN22: a recovery's steps are asked on the build seats' stream delivery

The runner keeps each commit the seat mesh-delivery says a recovery merged in its state recoveries; a pipeline whose commit is
named there has its go, node and image steps asked on delivery.step, ahead of the first round, at its start, its resume and a
rerun; a recovery heard after its pipeline started asks nothing again.

Builds: design 47 G §81, §86, §415, A-207
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go

@@ -59,6 +59,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	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 {
@@ -117,7 +122,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// for its step.
 	starter := &pipeline.Starter{Pipelines: createdKV{ctx: ctx, kv: pipelines}, Logf: logger.Printf, Say: say,
 		Registration: rv.Registrations.Registration, Steps: busStepAsker{ctx: ctx, js: js, cred: cred}, Name: cred.Module,
-		Arm: armBound(ctx, bounds), Fills: fills, Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)}
+		Arm: armBound(ctx, bounds), Fills: fills,
+		Recovered: kvRecoveries{ctx, recovered}.Has, Kept: keptReader(ctx, kept), Toolchain: toolchainReader(ctx, toolchains)}
 	resume := func(branch string) {
 		half, ok := rv.Catalog.View(branch)
 		if !ok {
@@ -149,6 +155,9 @@ 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
 	}

modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/recoveries.go

@@ -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

@@ -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

@@ -57,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", "aged"} {
+	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)
 		}

modules/mesh-delivery-pipeline/go.mod

@@ -31,4 +31,4 @@ 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

modules/mesh-delivery-pipeline/internal/pipeline/place.go

@@ -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}
 
@@ -143,10 +151,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/start.go

@@ -72,6 +72,8 @@ type Starter struct {
 	Arm Arm
 	// 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)
 }
 
 // Start writes b's pipeline, running, and says pipeline-started; false, nil when its key was written already (the second
@@ -120,7 +122,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
 		}
 	}
@@ -178,7 +180,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)
@@ -186,6 +188,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_test.go

@@ -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/verbs.go

@@ -281,7 +281,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/manifest_test.go

@@ -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,7 +78,7 @@ func TestManifest(t *testing.T) {
 			"mesh-delivery-indexer.take-in.taken", "mesh-builds.package-published", "mesh-builds.toolchain-agreed", "mesh-builds.step-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", "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",
@@ -90,7 +90,9 @@ func TestManifest(t *testing.T) {
 			"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"})},

modules/mesh-delivery-pipeline/module.json

@@ -42,7 +42,8 @@
     "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",
@@ -55,7 +56,10 @@
     "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"
@@ -75,6 +79,10 @@
       "name": "aged",
       "history": 1
     },
+    {
+      "name": "recoveries",
+      "history": 1
+    },
     {
       "name": "toolchains",
       "history": 1