A-187: the consumer ledger keeps a message's first delivery time when it is delivered again

A message delivered again in the same run keeps its kept time; after a restart, which kept none, it takes the time
the bus stored it, never later than its first delivery, so a message that never settles grows old and the runner's
health check fires for it.

Builds: design 47 G A-187, §76

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

go/process/consumers.go

@@ -59,7 +59,7 @@ func NewConsumerLedger(stream, name string, sink ConsumerSink, now func() time.T
 		facts: ConsumerFacts{Stream: stream, Name: name}}
 }
 
-// Delivered notes that m was given to the consumer, and writes the facts.
+// Delivered notes that m was given to the consumer, and writes the facts; a message given again keeps its first delivery's time.
 func (l *ConsumerLedger) Delivered(m jetstream.Msg) {
 	md, err := m.Metadata()
 	if err != nil {
@@ -67,7 +67,16 @@ func (l *ConsumerLedger) Delivered(m jetstream.Msg) {
 		return
 	}
 	l.mu.Lock()
-	l.unacks[md.Sequence.Stream] = l.now().UTC()
+	// A message delivered again keeps the time it was first delivered, so one that never settles grows old (A-187): asked again
+	// in this run, the time kept; after a restart, which kept none, the time the bus stored it, never later than its first
+	// delivery.
+	if _, known := l.unacks[md.Sequence.Stream]; !known {
+		first := l.now().UTC()
+		if md.NumDelivered > 1 && !md.Timestamp.IsZero() && md.Timestamp.Before(first) {
+			first = md.Timestamp.UTC()
+		}
+		l.unacks[md.Sequence.Stream] = first
+	}
 	l.facts.Pending = md.NumPending
 	l.mu.Unlock()
 	l.write()

go/process/consumers_test.go

@@ -15,11 +15,12 @@ type ledgerMsg struct {
 	seq     uint64
 	pending uint64
 	stored  time.Time
+	times   uint64
 }
 
 func (m ledgerMsg) Subject() string { return m.subject }
 func (m ledgerMsg) Metadata() (*jetstream.MsgMetadata, error) {
-	return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: m.seq, Consumer: m.seq}, NumPending: m.pending, Timestamp: m.stored}, nil
+	return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: m.seq, Consumer: m.seq}, NumPending: m.pending, Timestamp: m.stored, NumDelivered: max(m.times, 1)}, nil
 }
 
 func TestConsumerLedgerKeepsTheFactsAfterEachMessage(t *testing.T) {
@@ -40,6 +41,20 @@ func TestConsumerLedgerKeepsTheFactsAfterEachMessage(t *testing.T) {
 	if f := read(); f.Pending != 0 || !f.OldestUnacknowledged.Equal(now.Add(-time.Minute)) {
 		t.Errorf("after two deliveries: %+v, want pending 0 and the first one's delivery as the oldest unacknowledged", f)
 	}
+	// A message delivered again keeps its first delivery's time (A-187): asked again in this run, and after a restart, where the
+	// new ledger knows none, the time the bus stored it.
+	now = now.Add(time.Minute)
+	l.Delivered(ledgerMsg{subject: a.subject, seq: 7, pending: 0, times: 2})
+	if f := read(); !f.OldestUnacknowledged.Equal(now.Add(-2 * time.Minute)) {
+		t.Errorf("after a delivery again: %+v, want the first delivery kept as the oldest unacknowledged", f)
+	}
+	stored := now.Add(-time.Hour)
+	again := NewConsumerLedger("EVENTS", "n1_m", func(_ context.Context, key string, value []byte) error { kept[key] = value; return nil }, func() time.Time { return now }, nil)
+	again.Delivered(ledgerMsg{subject: a.subject, seq: 7, stored: stored, times: 3})
+	if f := read(); !f.OldestUnacknowledged.Equal(stored) {
+		t.Errorf("after a restart's delivery again: %+v, want the time the bus stored it", f)
+	}
+	now = now.Add(-time.Minute)
 	at := now.Add(-time.Hour)
 	l.Applied(a, at)
 	if f := read(); f.Acknowledged != 7 || f.LastEvent != "mesh.seat.s.event.a" || !f.LastAt.Equal(at) || !f.OldestUnacknowledged.Equal(now) {