Merge branch 'feat/47-G-fix11-b1e' into feat/47-G-join11

2026-11-12 10:24:52 +01:00
parent 1e1ca6b14f 0e1aa6dea3
commit 8cc9837f3a
10 changed files with 259 additions and 13 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)
+	}
+}
@@ -142,8 +142,59 @@ func subAModuleHearsASeatEventKeptOnItsOwnStream(t *testing.T) {
 	}
 }

+// A-181: a module hears another module's own stream kept last per subject (a manifest's `streams` entry with `last-per-subject`) as
+// it hears a seat's: the last message of each subject read whole at its start, though its durable began after them, then the durable.
+func subAModuleHearsAnotherModulesStreamKeptLastPerSubject(t *testing.T) {
+	nc, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer nc.Close()
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	stream, subject := "BUILDER_BUILDS", process.ModuleStreamSubject("builder", "build.ended.>")
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil {
+		t.Fatal(err)
+	}
+	for _, m := range []struct{ event, v string }{{"build.ended.web.fp1", "old"}, {"build.ended.web.fp1", "new"}, {"build.ended.web.fp2", "two"}} {
+		if _, err := js.Publish(ctx, process.ModuleStreamSubject("builder", m.event), []byte(m.v)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	cred := process.Account{Node: "ace", Module: "assigner"}
+	durable := cred.SeatEventConsumerConfig([]string{subject}, true)
+	durable.DeliverPolicy = jetstream.DeliverNewPolicy
+	if _, err := js.CreateOrUpdateConsumer(ctx, stream, durable); err != nil {
+		t.Fatal(err)
+	}
+	got := make(chan string, 4)
+	var reads []string
+	cc, err := cred.HearStream(ctx, js, stream, func(m jetstream.Msg) { got <- string(m.Data()); _ = m.Ack() }, t.Logf, process.OnRead(func(s string) { reads = append(reads, s) }))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer cc.Stop()
+	heard := map[string]bool{}
+	for len(heard) < 2 {
+		select {
+		case h := <-got:
+			heard[h] = true
+		case <-ctx.Done():
+			t.Fatalf("heard %v", heard)
+		}
+	}
+	if !heard["new"] || !heard["two"] || heard["old"] || len(reads) != 2 {
+		t.Errorf("heard %v with reads %v, want the last of each subject, read whole", heard, reads)
+	}
+}
+
 // TestTheHubBindsALateConsumerAndSaysItIsLive runs the tests of this package that were merged to keep the module within its test budget (design 47 G §419), each as a subtest.
 func TestTheHubBindsALateConsumerAndSaysItIsLive(t *testing.T) {
 	t.Run("TheHubBindsALateConsumerAndSaysItIsLive", subTheHubBindsALateConsumerAndSaysItIsLive)
 	t.Run("AModuleHearsASeatEventKeptOnItsOwnStream", subAModuleHearsASeatEventKeptOnItsOwnStream)
+	t.Run("AModuleHearsAnotherModulesStreamKeptLastPerSubject", subAModuleHearsAnotherModulesStreamKeptLastPerSubject)
 }
@@ -97,6 +97,13 @@ func (a Account) HearSeatEvents(ctx context.Context, js jetstream.JetStream, str
 		return nil, fmt.Errorf("%q is not the bus's name of a seat event's stream: a contract's Stream<E> is %s<seat>_<event>, "+
 			"with no dot (seatdef.KeptStream)", stream, seatdef.KeptStreamPrefix)
 	}
+	return a.HearStream(ctx, js, stream, handle, logf, opts...)
+}
+
+// HearStream is HearSeatEvents for the stream of a module's own events (a manifest's `streams` entry, named as
+// catalogue.StreamBusName names it), which the module is granted through its `consumes`: the same one rule, a stream the bus's own
+// info says is kept last per subject (`last-per-subject`) read whole first, then the durable followed (design 47 G §556, §703, A-181).
+func (a Account) HearStream(ctx context.Context, js jetstream.JetStream, stream string, handle func(jetstream.Msg), logf func(string, ...any), opts ...HearOption) (jetstream.ConsumeContext, error) {
 	if handle == nil {
 		return nil, fmt.Errorf("hearing %s needs a handler", stream)
 	}