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

2026-11-12 10:24:52 +01:00
parent 602f6cfe0b 9f2d1df1c9
commit b325ba0120
4 changed files with 101 additions and 4 deletions
@@ -59,7 +59,7 @@ func NewConsumerLedger(stream, name string, sink ConsumerSink, now func() time.T
 		facts: ConsumerFacts{Stream: stream, Name: name}}
 }

-// Delivered notes that m was given to the consumer, and writes the facts.
+// Delivered notes that m was given to the consumer, and writes the facts; a message given again keeps its first delivery's time.
 func (l *ConsumerLedger) Delivered(m jetstream.Msg) {
 	md, err := m.Metadata()
 	if err != nil {
@@ -67,7 +67,16 @@ func (l *ConsumerLedger) Delivered(m jetstream.Msg) {
 		return
 	}
 	l.mu.Lock()
-	l.unacks[md.Sequence.Stream] = l.now().UTC()
+	// A message delivered again keeps the time it was first delivered, so one that never settles grows old (A-187): asked again
+	// in this run, the time kept; after a restart, which kept none, the time the bus stored it, never later than its first
+	// delivery.
+	if _, known := l.unacks[md.Sequence.Stream]; !known {
+		first := l.now().UTC()
+		if md.NumDelivered > 1 && !md.Timestamp.IsZero() && md.Timestamp.Before(first) {
+			first = md.Timestamp.UTC()
+		}
+		l.unacks[md.Sequence.Stream] = first
+	}
 	l.facts.Pending = md.NumPending
 	l.mu.Unlock()
 	l.write()
@@ -15,11 +15,12 @@ type ledgerMsg struct {
 	seq     uint64
 	pending uint64
 	stored  time.Time
+	times   uint64
 }

 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
+	return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: m.seq, Consumer: m.seq}, NumPending: m.pending, Timestamp: m.stored, NumDelivered: max(m.times, 1)}, nil
 }

 func TestConsumerLedgerKeepsTheFactsAfterEachMessage(t *testing.T) {
@@ -40,6 +41,20 @@ func TestConsumerLedgerKeepsTheFactsAfterEachMessage(t *testing.T) {
 	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)
 	}
+	// A message delivered again keeps its first delivery's time (A-187): asked again in this run, and after a restart, where the
+	// new ledger knows none, the time the bus stored it.
+	now = now.Add(time.Minute)
+	l.Delivered(ledgerMsg{subject: a.subject, seq: 7, pending: 0, times: 2})
+	if f := read(); !f.OldestUnacknowledged.Equal(now.Add(-2 * time.Minute)) {
+		t.Errorf("after a delivery again: %+v, want the first delivery kept as the oldest unacknowledged", f)
+	}
+	stored := now.Add(-time.Hour)
+	again := NewConsumerLedger("EVENTS", "n1_m", func(_ context.Context, key string, value []byte) error { kept[key] = value; return nil }, func() time.Time { return now }, nil)
+	again.Delivered(ledgerMsg{subject: a.subject, seq: 7, stored: stored, times: 3})
+	if f := read(); !f.OldestUnacknowledged.Equal(stored) {
+		t.Errorf("after a restart's delivery again: %+v, want the time the bus stored it", f)
+	}
+	now = now.Add(-time.Minute)
 	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) {
@@ -179,6 +179,32 @@ func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context
 	return p, nil
 }

+// AdoptEvent hears a seat's event on the one subscription the followers adopted on that event share, with no picture and no
+// first view (design 47 G A-185, A-199): hear is handed each message of the event beside every adopted follower's handler,
+// and the message is acknowledged, unless hear fails: then it is left unacknowledged and the bus asks again (A-222). The subscription is made at once when no follower made it, so a part that follows no
+// view of the seat still hears the event; adopted before Start, it is made at Start.
+func (r *ProcessReader) AdoptEvent(seat, event string, hear func(context.Context, []byte) error) error {
+	if seat == "" || event == "" || hear == nil {
+		return errors.New("views: an adopted event names its seat, its event and its handler")
+	}
+	subject := "mesh.seat." + seat + ".event." + event
+	r.mu.Lock()
+	if r.adopted == nil {
+		r.adopted = map[string][]adoptedHear{}
+	}
+	fresh := len(r.adopted[subject]) == 0
+	r.adopted[subject] = append(r.adopted[subject], adoptedHear{hear: hear})
+	hub, handle := r.hub, r.handle
+	r.mu.Unlock()
+	if fresh && hub != nil {
+		// one subscription per event, with every further part (design 47 G §532)
+		if err := hub.Handle(subject+".>", handle); err != nil {
+			r.logf("views: %s's event %s is not heard (%v)", r.Module, subject, err)
+		}
+	}
+	return nil
+}
+
 // adopt is Adopt's part under the lock; subject is the event's subject when no follower adopted it before.
 func (r *ProcessReader) adopt(f *Follower, seat, event string, hear func(context.Context, []byte) error) (*Picture, string, error) {
 	r.mu.Lock()
@@ -339,12 +365,17 @@ func (r *ProcessReader) Hear(ctx context.Context) func(jetstream.Msg) {
 		}
 		r.mu.Unlock()
 		if hears != nil {
+			failed := false
 			for _, h := range hears {
 				if err := h.hear(ctx, m.Data()); err != nil {
 					r.logf("views: %s is not taken: %v", m.Subject(), err)
+					// an event adopted with no picture is asked again by the bus (design 47 G A-222)
+					failed = failed || h.bucket == ""
 				}
 			}
-			_ = m.Ack()
+			if !failed {
+				_ = m.Ack()
+			}
 			return
 		}
 		if owned != nil {
@@ -801,6 +801,48 @@ func subViews(t *testing.T) {
 		if len(a) != 1 || len(b) != 1 {
 			t.Errorf("two adopted views on one process reader heard %v and %v", a, b)
 		}
+		// A-185, A-199: an event adopted with no picture is heard once on the same subscription, beside the views; a reader
+		// following no view hears it too.
+		var hooked []string
+		hook := func(_ context.Context, d []byte) error { hooked = append(hooked, string(d)); return nil }
+		if err := prr.AdoptEvent("s", "revision-said", hook); err != nil || prr.AdoptEvent("s", "", hook) == nil {
+			t.Fatalf("AdoptEvent: %v, and one naming no event was taken", err)
+		}
+		prr.Hear(context.Background())(&seatMsg{subject: "mesh.seat.s.event.revision-said.main", data: []byte(`{"revision":8}`)})
+		if len(a) != 2 || len(b) != 2 || len(hooked) != 1 || len(prr.Followers()) != 2 {
+			t.Errorf("beside two views the adopted event heard %v (views %v, %v)", hooked, a, b)
+		}
+		alone := &ProcessReader{Module: "reader", Now: now}
+		if err := alone.AdoptEvent("s", "revision-said", hook); err != nil {
+			t.Fatal(err)
+		}
+		am := &seatMsg{subject: "mesh.seat.s.event.revision-said.next", data: []byte(`{"revision":9}`)}
+		alone.Hear(context.Background())(am)
+		if len(hooked) != 2 || !am.acked {
+			t.Errorf("a reader following no view heard %v, acked %v", hooked, am.acked)
+		}
+		// A-222: an adopted event whose handler fails is left unacknowledged, so the bus asks again.
+		failing := &ProcessReader{Module: "reader", Now: now}
+		var tries int
+		if err := failing.AdoptEvent("s", "revision-said", func(context.Context, []byte) error {
+			tries++
+			if tries == 1 {
+				return errors.New("not placed yet")
+			}
+			return nil
+		}); err != nil {
+			t.Fatal(err)
+		}
+		fm := &seatMsg{subject: "mesh.seat.s.event.revision-said.next", data: []byte(`{"revision":10}`)}
+		failing.Hear(context.Background())(fm)
+		if fm.acked {
+			t.Error("a failed adopted event was acknowledged")
+		}
+		fm2 := &seatMsg{subject: fm.subject, data: fm.data}
+		failing.Hear(context.Background())(fm2)
+		if !fm2.acked || tries != 2 {
+			t.Errorf("the event asked again: acked %v after %d tries", fm2.acked, tries)
+		}
 		pf.WaitFor(context.Background(), "4")
 		ended <- true
 		if fd := pfs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" {