From ebd71c5f15cd9e1e26b1cec29818f715637f191c Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 05:52:33 +0100 Subject: [PATCH 1/3] A-187: the consumer ledger keeps a message's first delivery time when it is delivered again MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A message delivered again in the same run keeps its kept time; after a restart, which kept none, it takes the time the bus stored it, never later than its first delivery, so a message that never settles grows old and the runner's health check fires for it. Builds: design 47 G A-187, §76 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/process/consumers.go | 13 +++++++++++-- go/process/consumers_test.go | 17 ++++++++++++++++- 2 files changed, 27 insertions(+), 3 deletions(-) diff --git a/go/process/consumers.go b/go/process/consumers.go index 80ffb536..1fc03324 100644 --- a/go/process/consumers.go +++ b/go/process/consumers.go @@ -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() diff --git a/go/process/consumers_test.go b/go/process/consumers_test.go index 5f5fb9ce..2044289d 100644 --- a/go/process/consumers_test.go +++ b/go/process/consumers_test.go @@ -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) { From c4521a2ba6346b3d858f225aecf6dd1e26f6a840 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:03:13 +0100 Subject: [PATCH 2/3] RN12: a process reader hears a seat's event with no picture on the event's one subscription MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AdoptEvent hands each message of the event to its handler beside the adopted views, and makes the subscription at once when no view made it, so the runner hears a run's candidate revision even when it follows no watched branch. Builds: design 47 G A-185, A-199, A-193, §532 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- go/views/process_reader.go | 26 ++++++++++++++++++++++++++ go/views/views_test.go | 20 ++++++++++++++++++++ 2 files changed, 46 insertions(+) diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 5115032f..3ac45459 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -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. 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() diff --git a/go/views/views_test.go b/go/views/views_test.go index 07bdd0ac..e0fcc69a 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -800,6 +800,26 @@ 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) + } pf.WaitFor(context.Background(), "4") ended <- true if fd := pfs.next(t); fd.Kind != ViewWaitsKind || fd.ID != "reader.s.revisions" { From 9f2d1df1c9deca9d249dd62f3fe42fa1af1e19cc Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 06:11:26 +0100 Subject: [PATCH 3/3] B1g-1: an adopted event whose handler fails is left unacknowledged, the bus asks again Builds: design 47 G A-222, A-193, A-199 --- go/views/process_reader.go | 9 +++++++-- go/views/views_test.go | 22 ++++++++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 3ac45459..2e98cba0 100644 --- a/go/views/process_reader.go +++ b/go/views/process_reader.go @@ -181,7 +181,7 @@ func (r *ProcessReader) Adopt(f *Follower, seat, event string, hear func(context // 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. The subscription is made at once when no follower made it, so a part that follows no +// 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 { @@ -365,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 { diff --git a/go/views/views_test.go b/go/views/views_test.go index e0fcc69a..1deb04b1 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -820,6 +820,28 @@ func subViews(t *testing.T) { 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" {