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) + } +}