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) { diff --git a/go/views/process_reader.go b/go/views/process_reader.go index 6a999fef..1edd9a31 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, 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 { diff --git a/go/views/views_test.go b/go/views/views_test.go index 1e3c5429..32f945b4 100644 --- a/go/views/views_test.go +++ b/go/views/views_test.go @@ -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" {