RN25: runner_health reads the consumer facts mesh-sdk's ledger writes

The runner's own writer of its consumer's oldest unacknowledged delivery is
replaced by mesh-sdk's consumer ledger: each message the hub hands a handler is
noted delivered, and applied when the handler acknowledges or ends it; one it
asks again stays unacknowledged. The key form is the same, a key per consumer
by its name in the state consumers. A runner that heard no event yet is healthy.

Builds: design 47 G §76, §468, §495; lead answers A-166, A-176

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

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

@@ -7,7 +7,7 @@ import (
 
 	"github.com/nats-io/nats.go/jetstream"
 
-	"mesh-delivery-pipeline/internal/pipeline"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 // handlers is where a part of the runner registers its handler of the events it hears: the event hub, or agedHub over it.
@@ -15,43 +15,43 @@ type handlers interface {
 	Handle(pattern string, handle func(jetstream.Msg)) error
 }
 
-// agedHub registers each handler on the hub so that every message it is handed is noted delivered, and acknowledged
-// once the handler acknowledges, refuses or ends it, so the runner's state consumers holds when the consumer's oldest
-// unacknowledged message was delivered (design 47 G §76, §468, §495; A-164, A-166).
+// agedHub registers each handler on the hub so that every message it is handed is noted delivered on the consumer's
+// ledger, and applied once the handler acknowledges or ends it, so the runner's state consumers holds when the
+// consumer's oldest unacknowledged message was delivered (design 47 G §76, §468, §495; A-164, A-166, A-176). A message
+// the handler asks again (a refusal for now) stays unacknowledged.
 type agedHub struct {
-	hub  handlers
-	ages *pipeline.ConsumerAges
+	hub    handlers
+	ledger *process.ConsumerLedger
 }
 
 func (a agedHub) Handle(pattern string, handle func(jetstream.Msg)) error {
 	return a.hub.Handle(pattern, func(m jetstream.Msg) {
-		handle(&agedMsg{Msg: m, ages: a.ages, mark: a.ages.Delivered()})
+		a.ledger.Delivered(m)
+		handle(&agedMsg{Msg: m, ledger: a.ledger})
 	})
 }
 
-// agedMsg is a message whose acknowledgement, refusal or end is noted on its consumer's ages, once.
+// agedMsg is a message whose acknowledgement or end is noted applied on its consumer's ledger, once.
 type agedMsg struct {
 	jetstream.Msg
-	ages *pipeline.ConsumerAges
-	mark uint64
-	once sync.Once
+	ledger *process.ConsumerLedger
+	once   sync.Once
 }
 
-func (m *agedMsg) done() { m.once.Do(func() { m.ages.Acknowledged(m.mark) }) }
+func (m *agedMsg) applied() { m.once.Do(func() { m.ledger.Applied(m.Msg, time.Time{}) }) }
 
-func (m *agedMsg) Ack() error { defer m.done(); return m.Msg.Ack() }
+func (m *agedMsg) Ack() error { defer m.applied(); return m.Msg.Ack() }
 
-func (m *agedMsg) DoubleAck(ctx context.Context) error { defer m.done(); return m.Msg.DoubleAck(ctx) }
+func (m *agedMsg) DoubleAck(ctx context.Context) error {
+	defer m.applied()
+	return m.Msg.DoubleAck(ctx)
+}
 
-func (m *agedMsg) Nak() error { defer m.done(); return m.Msg.Nak() }
+func (m *agedMsg) Term() error { defer m.applied(); return m.Msg.Term() }
 
-func (m *agedMsg) NakWithDelay(d time.Duration) error { defer m.done(); return m.Msg.NakWithDelay(d) }
+func (m *agedMsg) TermWithReason(r string) error { defer m.applied(); return m.Msg.TermWithReason(r) }
 
-func (m *agedMsg) Term() error { defer m.done(); return m.Msg.Term() }
-
-func (m *agedMsg) TermWithReason(r string) error { defer m.done(); return m.Msg.TermWithReason(r) }
-
-// consumersPut writes a consumer's age in the state consumers.
-func consumersPut(ctx context.Context, kv jetstream.KeyValue) func(string, []byte) error {
-	return func(k string, v []byte) error { _, err := kv.Put(ctx, k, v); return err }
+// consumersSink writes a consumer's facts in the state consumers.
+func consumersSink(kv jetstream.KeyValue) process.ConsumerSink {
+	return func(ctx context.Context, k string, v []byte) error { _, err := kv.Put(ctx, k, v); return err }
 }

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

@@ -1,41 +1,62 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
 	"testing"
 
 	"github.com/nats-io/nats.go/jetstream"
 
-	"mesh-delivery-pipeline/internal/pipeline"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 type oneHub struct{ handle func(jetstream.Msg) }
 
 func (h *oneHub) Handle(_ string, handle func(jetstream.Msg)) error { h.handle = handle; return nil }
 
-type ackedMsg struct{ jetstream.Msg }
+type ackedMsg struct {
+	jetstream.Msg
+	seq uint64
+}
 
 func (ackedMsg) Ack() error { return nil }
+func (ackedMsg) Nak() error { return nil }
+func (m ackedMsg) Metadata() (*jetstream.MsgMetadata, error) {
+	return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: m.seq}}, nil
+}
+func (ackedMsg) Subject() string { return "mesh.seat.x.event.y" }
 
-// TestAgedHub: a message handed to a handler is noted delivered, and acknowledged when the handler acknowledges it, so
-// the state consumers holds the consumer's oldest unacknowledged delivery (§76, §468; RN25, A-166).
+// TestAgedHub: a message handed to a handler is noted delivered on the consumer's ledger, applied when the handler
+// acknowledges it, and left unacknowledged when the handler asks it again (§76, §468; RN25, A-166, A-176).
 func TestAgedHub(t *testing.T) {
-	var last pipeline.ConsumerAge
-	ages := &pipeline.ConsumerAges{Consumer: "n_runner", Put: func(_ string, v []byte) error { last = pipeline.ConsumerAge{}; return json.Unmarshal(v, &last) }}
+	var last process.ConsumerFacts
+	ledger := process.NewConsumerLedger(process.EventsStream, "n_runner", func(_ context.Context, _ string, v []byte) error {
+		last = process.ConsumerFacts{}
+		return json.Unmarshal(v, &last)
+	}, nil, nil)
 	hub := &oneHub{}
-	var inside *pipeline.ConsumerAge
-	if err := (agedHub{hub: hub, ages: ages}).Handle("x.>", func(m jetstream.Msg) {
-		seen := last
-		inside = &seen
+	var inside process.ConsumerFacts
+	nak := false
+	if err := (agedHub{hub: hub, ledger: ledger}).Handle("x.>", func(m jetstream.Msg) {
+		inside = last
+		if nak {
+			_ = m.Nak()
+			return
+		}
 		_ = m.Ack()
 	}); err != nil {
 		t.Fatal(err)
 	}
-	hub.handle(ackedMsg{})
-	if inside == nil || inside.Oldest == nil {
-		t.Fatalf("inside the handler the consumer's oldest unacknowledged delivery = %+v, want it written", inside)
+	hub.handle(ackedMsg{seq: 1})
+	if inside.OldestUnacknowledged.IsZero() {
+		t.Fatalf("inside the handler the consumer's facts = %+v, want its delivery written", inside)
 	}
-	if last.Oldest != nil {
-		t.Errorf("after the handler acknowledged: oldest %v, want none", last.Oldest)
+	if !last.OldestUnacknowledged.IsZero() || last.Acknowledged != 1 {
+		t.Errorf("after the handler acknowledged: %+v, want none unacknowledged and 1 acknowledged", last)
+	}
+	nak = true
+	hub.handle(ackedMsg{seq: 2})
+	if last.OldestUnacknowledged.IsZero() {
+		t.Errorf("a message asked again is noted applied: %+v", last)
 	}
 }

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

@@ -71,15 +71,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	}
 	defer calls.Close()
 	hub := process.NewEventHub(logger.Printf)
-	// The age of the durable consumer's oldest unacknowledged message, read by the tool runner_health (§76, §468; A-164,
-	// A-166): written when a message is delivered or acknowledged.
+	// The durable consumer's facts, among them when its oldest unacknowledged message was delivered, read by the tool
+	// runner_health (§76, §468; A-164, A-166, A-176): written when a message is delivered or applied.
 	consumers, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, pipeline.StateConsumers), "say how far its consumer is behind")
 	if err != nil {
 		return err
 	}
-	ages := &pipeline.ConsumerAges{Consumer: cred.EventConsumerName(), Put: consumersPut(ctx, consumers), Logf: logger.Printf}
-	ages.Begin()
-	heard := agedHub{hub: hub, ages: ages}
+	heard := agedHub{hub: hub, ledger: process.NewConsumerLedger(process.EventsStream, cred.EventConsumerName(), consumersSink(consumers), nil, logger.Printf)}
 	// A branch's view holding its first half asks again the steps placed in its running pipelines that were never asked
 	// (§88); the starter is made after the views, so it is reached through a pointer set once it is.
 	var resumer atomic.Pointer[func(string)]

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

@@ -167,10 +167,10 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes
 	if err != nil {
 		t.Fatal(err)
 	}
-	var age pipeline.ConsumerAge
+	var facts process.ConsumerFacts
 	waitFor(t, ctx, logs, func() bool {
 		e, err := consumers.Get(ctx, "n1_"+module)
-		return err == nil && json.Unmarshal(e.Value(), &age) == nil && age.Oldest == nil && age.Consumer == "n1_"+module
+		return err == nil && json.Unmarshal(e.Value(), &facts) == nil && facts.OldestUnacknowledged.IsZero() && facts.Acknowledged > 0
 	})
 	if a := pipeline.Health(kvKept{ctx, pipelines}, kvKept{ctx, consumers}, time.Now()); !a.Healthy {
 		t.Errorf("runner_health over the bus: %+v", a)

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

@@ -5,103 +5,20 @@ import (
 	"fmt"
 	"sort"
 	"strings"
-	"sync"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/health"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
-// StateConsumers is the runner's old state section of its durable consumers' ages, a key per consumer (design 47 G
-// §468, §76, §495; A-164, A-166).
-const StateConsumers = "consumers"
+// StateConsumers is the runner's old state section of its durable consumers' facts, a key per consumer, written by
+// mesh-sdk's consumer ledger (design 47 G §468, §76, §495; A-164, A-166, A-176).
+const StateConsumers = process.ConsumersState
 
 // BehindAfter is how long a consumer's oldest unacknowledged message may wait before the runner's health check names it
 // behind (§76).
 const BehindAfter = 5 * time.Minute
 
-// ConsumerAge is one durable consumer's value in the state consumers: when its oldest unacknowledged message was
-// delivered, none when every delivered message is acknowledged (A-166).
-type ConsumerAge struct {
-	Consumer string     `json:"consumer"`
-	Oldest   *time.Time `json:"oldest-unacknowledged,omitempty"`
-}
-
-// ConsumerAges follows one durable consumer's delivered and acknowledged messages and writes its ConsumerAge through
-// Put each time the oldest unacknowledged delivery changes (A-166).
-type ConsumerAges struct {
-	Consumer string
-	Put      func(key string, value []byte) error
-	Now      func() time.Time
-	Logf     func(string, ...any)
-
-	mu      sync.Mutex
-	next    uint64
-	pending map[uint64]time.Time
-	written *time.Time
-	wrote   bool
-}
-
-// Begin writes the consumer's value at the runner's start, before any message is delivered, so its health check reads
-// every consumer the process holds.
-func (c *ConsumerAges) Begin() {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	c.writeLocked()
-}
-
-// Delivered notes a message delivered now and answers its mark, given to Acknowledged once the message is acknowledged.
-func (c *ConsumerAges) Delivered() uint64 {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	if c.pending == nil {
-		c.pending = map[uint64]time.Time{}
-	}
-	c.next++
-	now := time.Now().UTC()
-	if c.Now != nil {
-		now = c.Now().UTC()
-	}
-	c.pending[c.next] = now
-	c.writeLocked()
-	return c.next
-}
-
-// Acknowledged notes the message of mark acknowledged (or refused for good): it waits no more.
-func (c *ConsumerAges) Acknowledged(mark uint64) {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	if _, ok := c.pending[mark]; !ok {
-		return
-	}
-	delete(c.pending, mark)
-	c.writeLocked()
-}
-
-// writeLocked writes the consumer's oldest unacknowledged delivery when it differs from the one last written.
-func (c *ConsumerAges) writeLocked() {
-	var oldest *time.Time
-	for _, at := range c.pending {
-		if oldest == nil || at.Before(*oldest) {
-			at := at
-			oldest = &at
-		}
-	}
-	if c.wrote && ((oldest == nil && c.written == nil) || (oldest != nil && c.written != nil && oldest.Equal(*c.written))) {
-		return
-	}
-	raw, err := json.Marshal(ConsumerAge{Consumer: c.Consumer, Oldest: oldest})
-	if err == nil {
-		err = c.Put(c.Consumer, raw)
-	}
-	if err != nil {
-		if c.Logf != nil {
-			c.Logf("the age of the consumer %s is not written in %s: %v", c.Consumer, StateConsumers, err)
-		}
-		return
-	}
-	c.written, c.wrote = oldest, true
-}
-
 // Health is the runner's health as its tool runner_health answers it (§468, §76, §495; A-164): healthy when its state
 // pipelines answers, as show reads it, and each durable consumer's oldest unacknowledged message waits less than
 // BehindAfter; otherwise unhealthy naming each consumer behind and how long.
@@ -114,7 +31,7 @@ func Health(pipelines, consumers Kept, now time.Time) health.Answer {
 		return health.Unhealthy(fmt.Sprintf("The ages of the runner's consumers cannot be read from its state %s (%v).", StateConsumers, err))
 	}
 	if len(keys) == 0 {
-		return health.Unhealthy(fmt.Sprintf("The runner has written no consumer's age in its state %s: it has heard no event yet.", StateConsumers))
+		return health.Healthy(fmt.Sprintf("The runner answers show and has heard no event yet: its state %s holds no consumer.", StateConsumers))
 	}
 	sort.Strings(keys)
 	var behind []string
@@ -123,14 +40,14 @@ func Health(pipelines, consumers Kept, now time.Time) health.Answer {
 		if err != nil {
 			return health.Unhealthy(fmt.Sprintf("The age of the consumer %s cannot be read (%v).", k, err))
 		}
-		var a ConsumerAge
+		var a process.ConsumerFacts
 		if raw != nil {
 			if err := json.Unmarshal(raw, &a); err != nil {
 				return health.Unhealthy(fmt.Sprintf("The age of the consumer %s does not read (%v).", k, err))
 			}
 		}
-		if a.Oldest != nil {
-			if age := now.Sub(*a.Oldest); age >= BehindAfter {
+		if !a.OldestUnacknowledged.IsZero() {
+			if age := now.Sub(a.OldestUnacknowledged); age >= BehindAfter {
 				behind = append(behind, fmt.Sprintf("%s (its oldest unacknowledged message waits %s)", k, age.Round(time.Second)))
 			}
 		}

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

@@ -6,6 +6,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/process"
 )
 
 type mapKept map[string]json.RawMessage
@@ -24,52 +26,14 @@ type failingKept struct{}
 func (failingKept) Keys() ([]string, error)               { return nil, errors.New("no bucket") }
 func (failingKept) Value(string) (json.RawMessage, error) { return nil, errors.New("no bucket") }
 
-// TestConsumerAges: the process writes when its consumer's oldest unacknowledged message was delivered, each time it
-// changes, and none once all are acknowledged (§468, §495; A-164, A-166).
-func TestConsumerAges(t *testing.T) {
-	at := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC)
-	now := at
-	st := mapKept{}
-	writes := 0
-	c := &ConsumerAges{Consumer: "n_mesh-delivery-pipeline", Now: func() time.Time { return now },
-		Put: func(k string, v []byte) error { writes++; st[k] = v; return nil }}
-	oldest := func() *time.Time {
-		var a ConsumerAge
-		if err := json.Unmarshal(st["n_mesh-delivery-pipeline"], &a); err != nil {
-			t.Fatal(err)
-		}
-		return a.Oldest
-	}
-	c.Begin()
-	if writes != 1 || oldest() != nil {
-		t.Fatalf("at start: %d writes, oldest %v; want one write of none", writes, oldest())
-	}
-	first := c.Delivered()
-	now = at.Add(time.Minute)
-	second := c.Delivered()
-	if writes != 2 || !oldest().Equal(at) {
-		t.Errorf("two delivered: %d writes, oldest %v; want 2 writes, oldest %v (a later delivery leaves the oldest)", writes, oldest(), at)
-	}
-	c.Acknowledged(first)
-	if !oldest().Equal(at.Add(time.Minute)) {
-		t.Errorf("the first acknowledged: oldest %v, want %v", oldest(), at.Add(time.Minute))
-	}
-	c.Acknowledged(second)
-	c.Acknowledged(second)
-	if oldest() != nil || writes != 4 {
-		t.Errorf("all acknowledged: oldest %v after %d writes, want none after 4", oldest(), writes)
-	}
-}
-
 // TestHealth: runner_health is healthy when the runner answers show and each consumer is less than five minutes behind,
 // and otherwise names each consumer behind (§76, §468, §495; RN25, A-164).
 func TestHealth(t *testing.T) {
 	now := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC)
 	age := func(name string, since time.Duration) mapKept {
-		a := ConsumerAge{Consumer: name}
+		a := process.ConsumerFacts{Name: name}
 		if since > 0 {
-			t := now.Add(-since)
-			a.Oldest = &t
+			a.OldestUnacknowledged = now.Add(-since)
 		}
 		raw, _ := json.Marshal(a)
 		return mapKept{name: raw}
@@ -85,7 +49,7 @@ func TestHealth(t *testing.T) {
 		{"four minutes", mapKept{}, age("n_runner", 4*time.Minute), true, ""},
 		{"six minutes names the consumer", mapKept{}, age("n_runner", 6*time.Minute), false, "n_runner (its oldest unacknowledged message waits 6m0s)"},
 		{"show does not answer", failingKept{}, age("n_runner", 0), false, "does not answer show"},
-		{"no consumer written", mapKept{}, mapKept{}, false, "no consumer's age"},
+		{"no event heard yet", mapKept{}, mapKept{}, true, "has heard no event yet"},
 	}
 	for _, r := range rows {
 		a := Health(r.pipelines, r.consumers, now)