A-176: a helper writes a durable consumer's facts into the module's state consumers

process.ConsumerLedger writes stream, name, acknowledged sequence, pending, last event applied with its at and the delivery of the oldest unacknowledged message into one key per consumer, after each message delivered and applied, so a tool reads them. Builds: design 47 G §579, A-166, A-176 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 05:40:57 +01:00
parent df3b732eca
commit 2c4931751a
2 changed files with 174 additions and 0 deletions
@@ -0,0 +1,122 @@
+package process
+
+import (
+	"context"
+	"encoding/json"
+	"sync"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// ConsumersState is the name of the state a module keeps its durable consumers' facts in, one key per consumer by its name,
+// history 1 (design 47 G A-166, A-176): what a module's tool reads of a consumer, since a tool runs in a runtime of its own and
+// cannot read the consumer's information.
+const ConsumersState = "consumers"
+
+// ConsumerFacts is one durable consumer as ConsumerLedger keeps it.
+type ConsumerFacts struct {
+	// Stream and Name are the consumer's stream and its name.
+	Stream string `json:"stream"`
+	Name   string `json:"name"`
+	// Acknowledged is the stream sequence of the last message applied.
+	Acknowledged uint64 `json:"acknowledged"`
+	// Pending is the messages the consumer has not been given yet, as the last message said.
+	Pending uint64 `json:"pending"`
+	// LastEvent is the subject of the last event applied and LastAt its own `at`, or when the bus stored it where it has none.
+	LastEvent string    `json:"last-event,omitempty"`
+	LastAt    time.Time `json:"last-at,omitempty"`
+	// OldestUnacknowledged is when the oldest message delivered and not yet applied was delivered; zero when none is.
+	OldestUnacknowledged time.Time `json:"oldest-unacknowledged,omitempty"`
+}
+
+// ConsumerSink stores one key of the module's state consumers.
+type ConsumerSink func(ctx context.Context, key string, value []byte) error
+
+// ConsumerLedger writes a durable consumer's facts into the module's state consumers after each message delivered and each message
+// applied, so a tool reads them and nothing is read on a timer (design 47 G A-166, A-176). One ledger serves one consumer.
+type ConsumerLedger struct {
+	stream, name string
+	sink         ConsumerSink
+	now          func() time.Time
+	logf         func(string, ...any)
+
+	mu     sync.Mutex
+	facts  ConsumerFacts
+	unacks map[uint64]time.Time // delivered and not applied, by stream sequence
+}
+
+// NewConsumerLedger is the ledger of the consumer name on stream, writing through sink; a nil now is the clock and a nil logf says
+// nothing.
+func NewConsumerLedger(stream, name string, sink ConsumerSink, now func() time.Time, logf func(string, ...any)) *ConsumerLedger {
+	if now == nil {
+		now = time.Now
+	}
+	if logf == nil {
+		logf = func(string, ...any) {}
+	}
+	return &ConsumerLedger{stream: stream, name: name, sink: sink, now: now, logf: logf, unacks: map[uint64]time.Time{},
+		facts: ConsumerFacts{Stream: stream, Name: name}}
+}
+
+// Delivered notes that m was given to the consumer, and writes the facts.
+func (l *ConsumerLedger) Delivered(m jetstream.Msg) {
+	md, err := m.Metadata()
+	if err != nil {
+		l.logf("consumers: %s's message has no metadata, its delivery is not kept: %v", l.name, err)
+		return
+	}
+	l.mu.Lock()
+	l.unacks[md.Sequence.Stream] = l.now().UTC()
+	l.facts.Pending = md.NumPending
+	l.mu.Unlock()
+	l.write()
+}
+
+// Applied notes that m was applied and acknowledged, at the event's own `at` (zero: when the bus stored it), and writes the facts.
+func (l *ConsumerLedger) Applied(m jetstream.Msg, at time.Time) {
+	md, err := m.Metadata()
+	if err != nil {
+		l.logf("consumers: %s's message has no metadata, its application is not kept: %v", l.name, err)
+		return
+	}
+	if at.IsZero() {
+		at = md.Timestamp
+	}
+	l.mu.Lock()
+	delete(l.unacks, md.Sequence.Stream)
+	if md.Sequence.Stream > l.facts.Acknowledged {
+		l.facts.Acknowledged = md.Sequence.Stream
+	}
+	l.facts.Pending = md.NumPending
+	l.facts.LastEvent, l.facts.LastAt = m.Subject(), at.UTC()
+	l.mu.Unlock()
+	l.write()
+}
+
+// Facts are the facts now.
+func (l *ConsumerLedger) Facts() ConsumerFacts {
+	l.mu.Lock()
+	defer l.mu.Unlock()
+	f := l.facts
+	f.OldestUnacknowledged = time.Time{}
+	for _, t := range l.unacks {
+		if f.OldestUnacknowledged.IsZero() || t.Before(f.OldestUnacknowledged) {
+			f.OldestUnacknowledged = t
+		}
+	}
+	return f
+}
+
+func (l *ConsumerLedger) write() {
+	if l.sink == nil {
+		return
+	}
+	data, err := json.Marshal(l.Facts())
+	if err == nil {
+		err = l.sink(context.Background(), l.name, data)
+	}
+	if err != nil {
+		l.logf("consumers: the facts of %s are not written to the state %s: %v", l.name, ConsumersState, err)
+	}
+}
@@ -0,0 +1,52 @@
+package process
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+type ledgerMsg struct {
+	jetstream.Msg
+	subject string
+	seq     uint64
+	pending uint64
+	stored  time.Time
+}
+
+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
+}
+
+func TestConsumerLedgerKeepsTheFactsAfterEachMessage(t *testing.T) {
+	kept := map[string][]byte{}
+	now := time.Date(2026, 11, 12, 9, 0, 0, 0, time.UTC)
+	l := NewConsumerLedger("EVENTS", "n1_m", func(_ context.Context, key string, value []byte) error { kept[key] = value; return nil }, func() time.Time { return now }, nil)
+	read := func() ConsumerFacts {
+		var f ConsumerFacts
+		if err := json.Unmarshal(kept["n1_m"], &f); err != nil {
+			t.Fatal(err)
+		}
+		return f
+	}
+	a, b := ledgerMsg{subject: "mesh.seat.s.event.a", seq: 7, pending: 1}, ledgerMsg{subject: "mesh.seat.s.event.b", seq: 8, pending: 0}
+	l.Delivered(a)
+	now = now.Add(time.Minute)
+	l.Delivered(b)
+	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)
+	}
+	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) {
+		t.Errorf("after the first applied: %+v, want sequence 7, its event and at, and the second one's delivery as the oldest", f)
+	}
+	l.Applied(b, time.Time{})
+	if f := read(); f.Acknowledged != 8 || !f.OldestUnacknowledged.IsZero() {
+		t.Errorf("after both applied: %+v, want sequence 8 and none unacknowledged", f)
+	}
+}