jschoubben
631654eec6
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
53 lines
1.8 KiB
Go
53 lines
1.8 KiB
Go
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)
|
|
}
|
|
}
|