From 2c4931751a3fe1daf9c69d60c1ef27b6a51d99ec Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:40:57 +0100 Subject: [PATCH] A-176: a helper writes a durable consumer's facts into the module's state consumers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 --- go/process/consumers.go | 122 +++++++++++++++++++++++++++++++++++ go/process/consumers_test.go | 52 +++++++++++++++ 2 files changed, 174 insertions(+) create mode 100644 go/process/consumers.go create mode 100644 go/process/consumers_test.go diff --git a/go/process/consumers.go b/go/process/consumers.go new file mode 100644 index 00000000..80ffb536 --- /dev/null +++ b/go/process/consumers.go @@ -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) + } +} diff --git a/go/process/consumers_test.go b/go/process/consumers_test.go new file mode 100644 index 00000000..5f5fb9ce --- /dev/null +++ b/go/process/consumers_test.go @@ -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) + } +}