A reader asks again when its consumer was made after its first view, says every silent drop, and carries what it heard in its status (issue 1966)
mesh/delivery merged: source.main moved to novox/mesh-sdk@55f71b78114f
mesh/delivery-group group fix/1966-views-durable merged: every member is merged
mesh/check checking 38659c3937e7 on 55f71b78114f | novox/mesh-sdk#174 in group fix/1966-views-durable

A consumer is made from now (ADR 0443): an event said before it was made is never delivered to it and no later event
shows the loss. The event hub tells a part when the bus made its consumer, and a follower asked before that asks
again. Events passed over, at or below the held revision, or of a state not followed are now said in the log. The
status carries the highest revision heard and why the picture is behind it.

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
2026-11-09 02:26:10 +01:00
parent 4c8ebfba6d
commit 38659c3937
5 changed files with 113 additions and 11 deletions
+18
View File
@@ -23,6 +23,7 @@ type EventHub struct {
mu sync.RWMutex
routes []hubRoute
cons jetstream.ConsumeContext
bound []func(created time.Time)
}
type hubRoute struct {
@@ -56,6 +57,15 @@ func (h *EventHub) Handle(pattern string, handle func(jetstream.Msg)) error {
return nil
}
// OnBound registers f, told each time the hub binds the module's consumer, with when the bus made it. A consumer is made
// from now (ADR 0443), so an event said before it was made is not delivered to it: a part that took a picture before
// that must ask its owner again (views.Follower.ConsumerMade). Register before Start.
func (h *EventHub) OnBound(f func(created time.Time)) {
h.mu.Lock()
h.bound = append(h.bound, f)
h.mu.Unlock()
}
// Patterns are the registered subject patterns, in order.
func (h *EventHub) Patterns() []string {
h.mu.RLock()
@@ -83,6 +93,14 @@ func (h *EventHub) Start(ctx context.Context, js jetstream.JetStream, cred Accou
if err != nil {
return fmt.Errorf("the consumer %s on %s is not heard: %w", cred.EventConsumerName(), EventsStream, err)
}
if info, ierr := cons.Info(ctx); ierr != nil {
h.logf("events: the consumer %s on %s is heard, but its information is not read (%v): a first view taken before it was made is not asked again", cred.EventConsumerName(), EventsStream, ierr)
} else {
h.logf("events: the consumer %s on %s is heard; the bus made it at %s", cred.EventConsumerName(), EventsStream, info.Created.UTC().Format(time.RFC3339))
for _, f := range h.bound {
f(info.Created.UTC())
}
}
return nil
}