B1g-4: an unanswered drop is said call-unanswered on its five-minute bound key's expiry

Builds: design 47 G A-222, A-77, §350
2026-11-12 06:17:05 +01:00
parent cbb678f42e
commit 8e7b8ab1d1
8 changed files with 220 additions and 9 deletions
@@ -0,0 +1,92 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	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"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+// callFindings is the runner's finding writer as its drops use it (design 47 G A-222, §350).
+type callFindings interface {
+	Write(ctx context.Context, f condseat.Finding) error
+	Clear(ctx context.Context, scope, id, kind, why string) error
+}
+
+// dropFindingID is the id of the call-unanswered finding of buildSeat's drop of ask.
+func dropFindingID(buildSeat, ask string) string { return buildSeat + ".drop." + ask }
+
+// deleteBound deletes a key of the state bounds; a key already gone is no fault.
+func deleteBound(ctx context.Context, bounds jetstream.KeyValue) func(key string) error {
+	return func(key string) error {
+		if err := bounds.Delete(ctx, key); err != nil && !errors.Is(err, jetstream.ErrKeyNotFound) {
+			return err
+		}
+		return nil
+	}
+}
+
+// armDropBound arms the five-minute bound of a drop, a key of the state bounds with its own time to live (A-77, A-222).
+func armDropBound(ctx context.Context, bounds jetstream.KeyValue) func(key string) error {
+	return func(key string) error {
+		_, err := bounds.Create(ctx, key, []byte(`{"drop":"`+key+`"}`), jetstream.KeyTTL(pipeline.AskBound))
+		if errors.Is(err, jetstream.ErrKeyExists) {
+			return nil
+		}
+		return err
+	}
+}
+
+// dropAnswered ends the wait of an answered drop: its bound is deleted and a call-unanswered finding it raised before is
+// cleared.
+func dropAnswered(ctx context.Context, unbound func(key string) error, w callFindings) func(buildSeat, ask string) error {
+	return func(buildSeat, ask string) error {
+		if err := unbound(pipeline.DropBoundKey(buildSeat, ask)); err != nil {
+			return err
+		}
+		return w.Clear(ctx, mcc.ScopeCall, dropFindingID(buildSeat, ask), mcc.KindCallUnanswered, "the drop is answered")
+	}
+}
+
+// dropUnanswered says the drop of ask on buildSeat unanswered, the finding call-unanswered in the runner's findings, on the
+// expiry of its five-minute bound (A-222, §350, ADR 0668 §3).
+func dropUnanswered(ctx context.Context, w callFindings, module string, now func() time.Time) func(buildSeat, ask string) error {
+	return func(buildSeat, ask string) error {
+		if w == nil {
+			return errors.New("the runner has no finding writer")
+		}
+		id := dropFindingID(buildSeat, ask)
+		call := buildSeat + ".drop"
+		return w.Write(ctx, condseat.Finding{Scope: mcc.ScopeCall, ID: id, Kind: mcc.KindCallUnanswered, Severity: condseat.FindingSeverityWarning,
+			Summary:     fmt.Sprintf("the call %s of ask %s is unanswered", call, ask),
+			Headline:    "A call whose answer is kept is not answered",
+			Explanation: fmt.Sprintf("The drop of the superseded step's ask %s was asked of the seat %s as a kept call and no holder answered it within %s, so the build it names may still run.", ask, buildSeat, pipeline.AskBound),
+			Source:      module + "." + mcc.KindCallUnanswered, Observer: module, At: now().UTC(), Subject: condseat.Subject{Scope: mcc.ScopeCall, ID: id}})
+	}
+}
+
+// boundExpired takes the expiry of a key of the state bounds: a drop's says the drop unanswered, an ask's ends the step no
+// holder took (§32, §535, A-77); another key is none of the runner's.
+func boundExpired(h *pipeline.Hearer, unanswered func(buildSeat, ask string) error, logf func(string, ...any)) func(key string) {
+	return func(key string) {
+		if s, ask, ok := pipeline.ParseDropBound(key); ok {
+			if err := unanswered(s, ask); err != nil {
+				logf("the drop of ask %s on %s, unanswered, is not said: %v", ask, s, err)
+			}
+			return
+		}
+		if !pipeline.IsAskKey(key) {
+			return
+		}
+		if err := h.NotTaken(key, pipeline.AskBound); err != nil {
+			logf("the step of ask %s, which no holder took, is not ended: %v", key, err)
+		}
+	}
+}
@@ -0,0 +1,52 @@
+package main
+
+import (
+	"context"
+	"testing"
+	"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"
+
+	"mesh-delivery-pipeline/internal/pipeline"
+)
+
+type fakeFindings struct {
+	written []condseat.Finding
+	cleared []string
+}
+
+func (f *fakeFindings) Write(_ context.Context, x condseat.Finding) error {
+	f.written = append(f.written, x)
+	return nil
+}
+
+func (f *fakeFindings) Clear(_ context.Context, scope, id, kind, _ string) error {
+	f.cleared = append(f.cleared, scope+"."+id+"."+kind)
+	return nil
+}
+
+// A-222, A-77: the expiry of a drop's five-minute bound says the drop call-unanswered in the runner's findings, naming the
+// seat and the ask; an answered drop clears it; a key that is no drop's is not said.
+func TestADropBoundsExpirySaysCallUnanswered(t *testing.T) {
+	w := &fakeFindings{}
+	var logged int
+	expired := boundExpired(nil, dropUnanswered(context.Background(), w, "the-runner", func() time.Time { return time.Unix(100, 0) }), func(string, ...any) { logged++ })
+	expired("not-a-bound")
+	if len(w.written) != 0 {
+		t.Fatalf("a key of no drop was said: %+v", w.written)
+	}
+	expired(pipeline.DropBoundKey("node-build-go", "step-ab"))
+	if len(w.written) != 1 {
+		t.Fatalf("written %+v, want one finding", w.written)
+	}
+	f := w.written[0]
+	if f.Scope != mcc.ScopeCall || f.Kind != mcc.KindCallUnanswered || f.ID != "node-build-go.drop.step-ab" || f.Observer != "the-runner" || f.Subject.ID != f.ID {
+		t.Errorf("the finding = %+v", f)
+	}
+	var unbound []string
+	if err := dropAnswered(context.Background(), func(k string) error { unbound = append(unbound, k); return nil }, w)("node-build-go", "step-ab"); err != nil ||
+		len(w.cleared) != 1 || w.cleared[0] != "call.node-build-go.drop.step-ab.call-unanswered" || len(unbound) != 1 || unbound[0] != "drop.node-build-go.step-ab" {
+		t.Errorf("an answered drop: %v, cleared %v, unbound %v", err, w.cleared, unbound)
+	}
+}
@@ -14,6 +14,7 @@ import (
 	"strings"
 	"sync/atomic"
 	"syscall"
+	"time"

 	"github.com/nats-io/nats.go"

@@ -160,7 +161,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// A newer commit or publish supersedes the older running pipelines it replaces, their asks dropped as kept calls (§31,
 	// §350; A-182, A-202).
 	starter.Supersede = func(newer seat.Pipeline) {
-		n, err := hearer.Supersede(newer, dropAsk(calls, cred.Module), dropBound(ctx, bounds))
+		n, err := hearer.Supersede(newer, pipeline.BoundDrop(dropAsk(calls, cred.Module), armDropBound(ctx, bounds), dropAnswered(ctx, deleteBound(ctx, bounds), rv.Findings)), dropBound(ctx, bounds))
 		if err != nil {
 			logger.Printf("the pipelines %s supersedes: %d ended, then %v", newer.Pipeline, n, err)
 		} else if n > 0 {
@@ -176,7 +177,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err := hearAskTaken(ctx, heard, bounds, hearer.Taken, logger.Printf); err != nil {
 		return err
 	}
-	if err := watchBounds(ctx, bounds, hearer, logger.Printf); err != nil {
+	if err := watchBounds(ctx, bounds, boundExpired(hearer, dropUnanswered(ctx, rv.Findings, cred.Module, time.Now), logger.Printf), logger.Printf); err != nil {
 		return err
 	}
 	if err := hearDecided(heard, hearer, cred.Module, logger.Printf); err != nil {
@@ -179,8 +179,8 @@ func armBound(ctx context.Context, bounds jetstream.KeyValue) pipeline.Arm {
 }

 // watchBounds hears the end of each ask's bound, the bus's marker of the key's expiry, and ends the step no holder took (§32,
-// §535, A-77): no timer in the process.
-func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hearer, logf func(string, ...any)) error {
+// §535, A-77), and says a drop no holder answered unanswered (A-222): no timer in the process.
+func watchBounds(ctx context.Context, bounds jetstream.KeyValue, expired func(key string), logf func(string, ...any)) error {
 	w, err := bounds.WatchAll(ctx, jetstream.UpdatesOnly())
 	if err != nil {
 		return err
@@ -195,12 +195,10 @@ func watchBounds(ctx context.Context, bounds jetstream.KeyValue, h *pipeline.Hea
 				if !ok {
 					return
 				}
-				if e == nil || e.Operation() != jetstream.KeyValuePurge || !pipeline.IsAskKey(e.Key()) {
+				if e == nil || e.Operation() != jetstream.KeyValuePurge {
 					continue
 				}
-				if err := h.NotTaken(e.Key(), pipeline.AskBound); err != nil {
-					logf("the step of ask %s, which no holder took, is not ended: %v", e.Key(), err)
-				}
+				expired(e.Key())
 			}
 		}
 	}()
@@ -20,6 +20,8 @@ import (
 type runnerViews struct {
 	Registrations *meshcatalogclient.Registrations
 	Catalog       *pipeline.Views
+	// Findings is the runner's finding writer, the reader's and the runner's own findings through it.
+	Findings callFindings
 }

 // startViews wires the runner's views on hub and asks each first view; register every other handler on hub before or
@@ -59,7 +61,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.
@@ -10,6 +10,7 @@ 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/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
@@ -93,3 +93,36 @@ func (h *Hearer) endNotTaken(key, ask string, bound time.Duration) error {

 // IsAskKey says whether a key of the state `bounds` is an ask's bound.
 func IsAskKey(key string) bool { return strings.HasPrefix(key, "step-") && !strings.Contains(key, ".") }
+
+// DropBoundPrefix begins the key of a superseded step's drop in the state `bounds`, `drop.<build seat>.<ask>` (design 47 G
+// A-222, A-77): its five-minute expiry says the drop unanswered.
+const DropBoundPrefix = "drop."
+
+// DropBoundKey is the key of the bound of buildSeat's drop of ask.
+func DropBoundKey(buildSeat, ask string) string { return DropBoundPrefix + buildSeat + "." + ask }
+
+// ParseDropBound reads a key of the state `bounds` as a drop's bound: the build seat and the ask it names.
+func ParseDropBound(key string) (buildSeat, ask string, ok bool) {
+	rest, found := strings.CutPrefix(key, DropBoundPrefix)
+	if !found {
+		return "", "", false
+	}
+	buildSeat, ask, found = strings.Cut(rest, ".")
+	return buildSeat, ask, found && buildSeat != "" && ask != "" && !strings.Contains(ask, ".")
+}
+
+// BoundDrop is drop with a bound of five minutes (A-222, A-77): the bound key is armed, with its own time to live, before
+// the drop is asked, and answered is told when the drop is answered, to delete the key, so an answer ends the wait with no
+// finding; a drop never answered leaves its key to expire, and the expiry says it unanswered. A bound that cannot be armed
+// is returned and the drop not asked.
+func BoundDrop(drop Drop, arm func(key string) error, answered func(buildSeat, ask string) error) Drop {
+	return func(buildSeat, node, ask, why string) error {
+		if err := arm(DropBoundKey(buildSeat, ask)); err != nil {
+			return fmt.Errorf("the bound of the drop of ask %s is not armed: %w", ask, err)
+		}
+		if err := drop(buildSeat, node, ask, why); err != nil {
+			return err
+		}
+		return answered(buildSeat, ask)
+	}
+}
@@ -1,6 +1,7 @@
 package pipeline

 import (
+	"errors"
 	"slices"
 	"strings"
 	"testing"
@@ -91,3 +92,34 @@ func TestSupersede(t *testing.T) {
 		t.Errorf("the older publish's pipeline is %s: %q", p.Outcome, p.Why)
 	}
 }
+
+// A-222, A-77: a drop is bounded by a key of its own armed before it is asked and deleted when it is answered; one never
+// answered leaves its key to expire.
+func TestBoundDrop(t *testing.T) {
+	var armed, answered []string
+	arm := func(key string) error { armed = append(armed, key); return nil }
+	done := func(s, ask string) error { answered = append(answered, s+"/"+ask); return nil }
+	if s, ask, ok := ParseDropBound(DropBoundKey(nodebuildgo.Seat, "step-ab")); !ok || s != nodebuildgo.Seat || ask != "step-ab" {
+		t.Errorf("the drop's key reads %q, %q, %v", s, ask, ok)
+	}
+	if _, _, ok := ParseDropBound("step-ab"); ok {
+		t.Error("an ask's bound was read as a drop's")
+	}
+	if IsAskKey(DropBoundKey("s", "step-ab")) {
+		t.Error("a drop's bound was taken for an ask's")
+	}
+	err := BoundDrop(func(string, string, string, string) error { return nil }, arm, done)(nodebuildgo.Seat, "n1", "step-ab", "why")
+	if err != nil || len(armed) != 1 || armed[0] != "drop."+nodebuildgo.Seat+".step-ab" || len(answered) != 1 {
+		t.Errorf("an answered drop: %v, armed %v, answered %v", err, armed, answered)
+	}
+	answered = nil
+	err = BoundDrop(func(string, string, string, string) error { return errors.New("no responders") }, arm, done)(nodebuildgo.Seat, "", "step-cd", "why")
+	if err == nil || len(armed) != 2 || len(answered) != 0 {
+		t.Errorf("an unanswered drop: %v, armed %v, answered %v (its key is left to expire)", err, armed, answered)
+	}
+	asked := false
+	err = BoundDrop(func(string, string, string, string) error { asked = true; return nil }, func(string) error { return errors.New("full") }, done)(nodebuildgo.Seat, "", "step-ef", "why")
+	if err == nil || asked {
+		t.Errorf("a drop whose bound is not armed: %v, asked %v", err, asked)
+	}
+}