diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go index 2581c317..288ab266 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers.go @@ -7,7 +7,7 @@ import ( "github.com/nats-io/nats.go/jetstream" - "mesh-delivery-pipeline/internal/pipeline" + "git.novox.be/novox/mesh-sdk/go/process" ) // handlers is where a part of the runner registers its handler of the events it hears: the event hub, or agedHub over it. @@ -15,43 +15,43 @@ type handlers interface { Handle(pattern string, handle func(jetstream.Msg)) error } -// agedHub registers each handler on the hub so that every message it is handed is noted delivered, and acknowledged -// once the handler acknowledges, refuses or ends it, so the runner's state consumers holds when the consumer's oldest -// unacknowledged message was delivered (design 47 G §76, §468, §495; A-164, A-166). +// agedHub registers each handler on the hub so that every message it is handed is noted delivered on the consumer's +// ledger, and applied once the handler acknowledges or ends it, so the runner's state consumers holds when the +// consumer's oldest unacknowledged message was delivered (design 47 G §76, §468, §495; A-164, A-166, A-176). A message +// the handler asks again (a refusal for now) stays unacknowledged. type agedHub struct { - hub handlers - ages *pipeline.ConsumerAges + hub handlers + ledger *process.ConsumerLedger } func (a agedHub) Handle(pattern string, handle func(jetstream.Msg)) error { return a.hub.Handle(pattern, func(m jetstream.Msg) { - handle(&agedMsg{Msg: m, ages: a.ages, mark: a.ages.Delivered()}) + a.ledger.Delivered(m) + handle(&agedMsg{Msg: m, ledger: a.ledger}) }) } -// agedMsg is a message whose acknowledgement, refusal or end is noted on its consumer's ages, once. +// agedMsg is a message whose acknowledgement or end is noted applied on its consumer's ledger, once. type agedMsg struct { jetstream.Msg - ages *pipeline.ConsumerAges - mark uint64 - once sync.Once + ledger *process.ConsumerLedger + once sync.Once } -func (m *agedMsg) done() { m.once.Do(func() { m.ages.Acknowledged(m.mark) }) } +func (m *agedMsg) applied() { m.once.Do(func() { m.ledger.Applied(m.Msg, time.Time{}) }) } -func (m *agedMsg) Ack() error { defer m.done(); return m.Msg.Ack() } +func (m *agedMsg) Ack() error { defer m.applied(); return m.Msg.Ack() } -func (m *agedMsg) DoubleAck(ctx context.Context) error { defer m.done(); return m.Msg.DoubleAck(ctx) } +func (m *agedMsg) DoubleAck(ctx context.Context) error { + defer m.applied() + return m.Msg.DoubleAck(ctx) +} -func (m *agedMsg) Nak() error { defer m.done(); return m.Msg.Nak() } +func (m *agedMsg) Term() error { defer m.applied(); return m.Msg.Term() } -func (m *agedMsg) NakWithDelay(d time.Duration) error { defer m.done(); return m.Msg.NakWithDelay(d) } +func (m *agedMsg) TermWithReason(r string) error { defer m.applied(); return m.Msg.TermWithReason(r) } -func (m *agedMsg) Term() error { defer m.done(); return m.Msg.Term() } - -func (m *agedMsg) TermWithReason(r string) error { defer m.done(); return m.Msg.TermWithReason(r) } - -// consumersPut writes a consumer's age in the state consumers. -func consumersPut(ctx context.Context, kv jetstream.KeyValue) func(string, []byte) error { - return func(k string, v []byte) error { _, err := kv.Put(ctx, k, v); return err } +// consumersSink writes a consumer's facts in the state consumers. +func consumersSink(kv jetstream.KeyValue) process.ConsumerSink { + return func(ctx context.Context, k string, v []byte) error { _, err := kv.Put(ctx, k, v); return err } } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go index 8dd8db43..91ddcd8c 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/consumers_test.go @@ -1,41 +1,62 @@ package main import ( + "context" "encoding/json" "testing" "github.com/nats-io/nats.go/jetstream" - "mesh-delivery-pipeline/internal/pipeline" + "git.novox.be/novox/mesh-sdk/go/process" ) type oneHub struct{ handle func(jetstream.Msg) } func (h *oneHub) Handle(_ string, handle func(jetstream.Msg)) error { h.handle = handle; return nil } -type ackedMsg struct{ jetstream.Msg } +type ackedMsg struct { + jetstream.Msg + seq uint64 +} func (ackedMsg) Ack() error { return nil } +func (ackedMsg) Nak() error { return nil } +func (m ackedMsg) Metadata() (*jetstream.MsgMetadata, error) { + return &jetstream.MsgMetadata{Sequence: jetstream.SequencePair{Stream: m.seq}}, nil +} +func (ackedMsg) Subject() string { return "mesh.seat.x.event.y" } -// TestAgedHub: a message handed to a handler is noted delivered, and acknowledged when the handler acknowledges it, so -// the state consumers holds the consumer's oldest unacknowledged delivery (§76, §468; RN25, A-166). +// TestAgedHub: a message handed to a handler is noted delivered on the consumer's ledger, applied when the handler +// acknowledges it, and left unacknowledged when the handler asks it again (§76, §468; RN25, A-166, A-176). func TestAgedHub(t *testing.T) { - var last pipeline.ConsumerAge - ages := &pipeline.ConsumerAges{Consumer: "n_runner", Put: func(_ string, v []byte) error { last = pipeline.ConsumerAge{}; return json.Unmarshal(v, &last) }} + var last process.ConsumerFacts + ledger := process.NewConsumerLedger(process.EventsStream, "n_runner", func(_ context.Context, _ string, v []byte) error { + last = process.ConsumerFacts{} + return json.Unmarshal(v, &last) + }, nil, nil) hub := &oneHub{} - var inside *pipeline.ConsumerAge - if err := (agedHub{hub: hub, ages: ages}).Handle("x.>", func(m jetstream.Msg) { - seen := last - inside = &seen + var inside process.ConsumerFacts + nak := false + if err := (agedHub{hub: hub, ledger: ledger}).Handle("x.>", func(m jetstream.Msg) { + inside = last + if nak { + _ = m.Nak() + return + } _ = m.Ack() }); err != nil { t.Fatal(err) } - hub.handle(ackedMsg{}) - if inside == nil || inside.Oldest == nil { - t.Fatalf("inside the handler the consumer's oldest unacknowledged delivery = %+v, want it written", inside) + hub.handle(ackedMsg{seq: 1}) + if inside.OldestUnacknowledged.IsZero() { + t.Fatalf("inside the handler the consumer's facts = %+v, want its delivery written", inside) } - if last.Oldest != nil { - t.Errorf("after the handler acknowledged: oldest %v, want none", last.Oldest) + if !last.OldestUnacknowledged.IsZero() || last.Acknowledged != 1 { + t.Errorf("after the handler acknowledged: %+v, want none unacknowledged and 1 acknowledged", last) + } + nak = true + hub.handle(ackedMsg{seq: 2}) + if last.OldestUnacknowledged.IsZero() { + t.Errorf("a message asked again is noted applied: %+v", last) } } diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go index 4e073a16..9712039e 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/main.go @@ -71,15 +71,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer calls.Close() hub := process.NewEventHub(logger.Printf) - // The age of the durable consumer's oldest unacknowledged message, read by the tool runner_health (§76, §468; A-164, - // A-166): written when a message is delivered or acknowledged. + // The durable consumer's facts, among them when its oldest unacknowledged message was delivered, read by the tool + // runner_health (§76, §468; A-164, A-166, A-176): written when a message is delivered or applied. consumers, err := process.EssentialBucket(ctx, js, "pipeline runner", ownBucket(cred.Module, pipeline.StateConsumers), "say how far its consumer is behind") if err != nil { return err } - ages := &pipeline.ConsumerAges{Consumer: cred.EventConsumerName(), Put: consumersPut(ctx, consumers), Logf: logger.Printf} - ages.Begin() - heard := agedHub{hub: hub, ages: ages} + heard := agedHub{hub: hub, ledger: process.NewConsumerLedger(process.EventsStream, cred.EventConsumerName(), consumersSink(consumers), nil, logger.Printf)} // A branch's view holding its first half asks again the steps placed in its running pipelines that were never asked // (§88); the starter is made after the views, so it is reached through a pointer set once it is. var resumer atomic.Pointer[func(string)] diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go index 61ce0a1f..feb5d854 100644 --- a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go @@ -167,10 +167,10 @@ func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *tes if err != nil { t.Fatal(err) } - var age pipeline.ConsumerAge + var facts process.ConsumerFacts waitFor(t, ctx, logs, func() bool { e, err := consumers.Get(ctx, "n1_"+module) - return err == nil && json.Unmarshal(e.Value(), &age) == nil && age.Oldest == nil && age.Consumer == "n1_"+module + return err == nil && json.Unmarshal(e.Value(), &facts) == nil && facts.OldestUnacknowledged.IsZero() && facts.Acknowledged > 0 }) if a := pipeline.Health(kvKept{ctx, pipelines}, kvKept{ctx, consumers}, time.Now()); !a.Healthy { t.Errorf("runner_health over the bus: %+v", a) diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go b/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go index 05f8b95a..d7a85fab 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/consumers.go @@ -5,103 +5,20 @@ import ( "fmt" "sort" "strings" - "sync" "time" "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/process" ) -// StateConsumers is the runner's old state section of its durable consumers' ages, a key per consumer (design 47 G -// §468, §76, §495; A-164, A-166). -const StateConsumers = "consumers" +// StateConsumers is the runner's old state section of its durable consumers' facts, a key per consumer, written by +// mesh-sdk's consumer ledger (design 47 G §468, §76, §495; A-164, A-166, A-176). +const StateConsumers = process.ConsumersState // BehindAfter is how long a consumer's oldest unacknowledged message may wait before the runner's health check names it // behind (§76). const BehindAfter = 5 * time.Minute -// ConsumerAge is one durable consumer's value in the state consumers: when its oldest unacknowledged message was -// delivered, none when every delivered message is acknowledged (A-166). -type ConsumerAge struct { - Consumer string `json:"consumer"` - Oldest *time.Time `json:"oldest-unacknowledged,omitempty"` -} - -// ConsumerAges follows one durable consumer's delivered and acknowledged messages and writes its ConsumerAge through -// Put each time the oldest unacknowledged delivery changes (A-166). -type ConsumerAges struct { - Consumer string - Put func(key string, value []byte) error - Now func() time.Time - Logf func(string, ...any) - - mu sync.Mutex - next uint64 - pending map[uint64]time.Time - written *time.Time - wrote bool -} - -// Begin writes the consumer's value at the runner's start, before any message is delivered, so its health check reads -// every consumer the process holds. -func (c *ConsumerAges) Begin() { - c.mu.Lock() - defer c.mu.Unlock() - c.writeLocked() -} - -// Delivered notes a message delivered now and answers its mark, given to Acknowledged once the message is acknowledged. -func (c *ConsumerAges) Delivered() uint64 { - c.mu.Lock() - defer c.mu.Unlock() - if c.pending == nil { - c.pending = map[uint64]time.Time{} - } - c.next++ - now := time.Now().UTC() - if c.Now != nil { - now = c.Now().UTC() - } - c.pending[c.next] = now - c.writeLocked() - return c.next -} - -// Acknowledged notes the message of mark acknowledged (or refused for good): it waits no more. -func (c *ConsumerAges) Acknowledged(mark uint64) { - c.mu.Lock() - defer c.mu.Unlock() - if _, ok := c.pending[mark]; !ok { - return - } - delete(c.pending, mark) - c.writeLocked() -} - -// writeLocked writes the consumer's oldest unacknowledged delivery when it differs from the one last written. -func (c *ConsumerAges) writeLocked() { - var oldest *time.Time - for _, at := range c.pending { - if oldest == nil || at.Before(*oldest) { - at := at - oldest = &at - } - } - if c.wrote && ((oldest == nil && c.written == nil) || (oldest != nil && c.written != nil && oldest.Equal(*c.written))) { - return - } - raw, err := json.Marshal(ConsumerAge{Consumer: c.Consumer, Oldest: oldest}) - if err == nil { - err = c.Put(c.Consumer, raw) - } - if err != nil { - if c.Logf != nil { - c.Logf("the age of the consumer %s is not written in %s: %v", c.Consumer, StateConsumers, err) - } - return - } - c.written, c.wrote = oldest, true -} - // Health is the runner's health as its tool runner_health answers it (§468, §76, §495; A-164): healthy when its state // pipelines answers, as show reads it, and each durable consumer's oldest unacknowledged message waits less than // BehindAfter; otherwise unhealthy naming each consumer behind and how long. @@ -114,7 +31,7 @@ func Health(pipelines, consumers Kept, now time.Time) health.Answer { return health.Unhealthy(fmt.Sprintf("The ages of the runner's consumers cannot be read from its state %s (%v).", StateConsumers, err)) } if len(keys) == 0 { - return health.Unhealthy(fmt.Sprintf("The runner has written no consumer's age in its state %s: it has heard no event yet.", StateConsumers)) + return health.Healthy(fmt.Sprintf("The runner answers show and has heard no event yet: its state %s holds no consumer.", StateConsumers)) } sort.Strings(keys) var behind []string @@ -123,14 +40,14 @@ func Health(pipelines, consumers Kept, now time.Time) health.Answer { if err != nil { return health.Unhealthy(fmt.Sprintf("The age of the consumer %s cannot be read (%v).", k, err)) } - var a ConsumerAge + var a process.ConsumerFacts if raw != nil { if err := json.Unmarshal(raw, &a); err != nil { return health.Unhealthy(fmt.Sprintf("The age of the consumer %s does not read (%v).", k, err)) } } - if a.Oldest != nil { - if age := now.Sub(*a.Oldest); age >= BehindAfter { + if !a.OldestUnacknowledged.IsZero() { + if age := now.Sub(a.OldestUnacknowledged); age >= BehindAfter { behind = append(behind, fmt.Sprintf("%s (its oldest unacknowledged message waits %s)", k, age.Round(time.Second))) } } diff --git a/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go b/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go index 8cfac785..8decf31c 100644 --- a/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go +++ b/modules/mesh-delivery-pipeline/internal/pipeline/consumers_test.go @@ -6,6 +6,8 @@ import ( "strings" "testing" "time" + + "git.novox.be/novox/mesh-sdk/go/process" ) type mapKept map[string]json.RawMessage @@ -24,52 +26,14 @@ type failingKept struct{} func (failingKept) Keys() ([]string, error) { return nil, errors.New("no bucket") } func (failingKept) Value(string) (json.RawMessage, error) { return nil, errors.New("no bucket") } -// TestConsumerAges: the process writes when its consumer's oldest unacknowledged message was delivered, each time it -// changes, and none once all are acknowledged (§468, §495; A-164, A-166). -func TestConsumerAges(t *testing.T) { - at := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC) - now := at - st := mapKept{} - writes := 0 - c := &ConsumerAges{Consumer: "n_mesh-delivery-pipeline", Now: func() time.Time { return now }, - Put: func(k string, v []byte) error { writes++; st[k] = v; return nil }} - oldest := func() *time.Time { - var a ConsumerAge - if err := json.Unmarshal(st["n_mesh-delivery-pipeline"], &a); err != nil { - t.Fatal(err) - } - return a.Oldest - } - c.Begin() - if writes != 1 || oldest() != nil { - t.Fatalf("at start: %d writes, oldest %v; want one write of none", writes, oldest()) - } - first := c.Delivered() - now = at.Add(time.Minute) - second := c.Delivered() - if writes != 2 || !oldest().Equal(at) { - t.Errorf("two delivered: %d writes, oldest %v; want 2 writes, oldest %v (a later delivery leaves the oldest)", writes, oldest(), at) - } - c.Acknowledged(first) - if !oldest().Equal(at.Add(time.Minute)) { - t.Errorf("the first acknowledged: oldest %v, want %v", oldest(), at.Add(time.Minute)) - } - c.Acknowledged(second) - c.Acknowledged(second) - if oldest() != nil || writes != 4 { - t.Errorf("all acknowledged: oldest %v after %d writes, want none after 4", oldest(), writes) - } -} - // TestHealth: runner_health is healthy when the runner answers show and each consumer is less than five minutes behind, // and otherwise names each consumer behind (§76, §468, §495; RN25, A-164). func TestHealth(t *testing.T) { now := time.Date(2026, 11, 12, 10, 0, 0, 0, time.UTC) age := func(name string, since time.Duration) mapKept { - a := ConsumerAge{Consumer: name} + a := process.ConsumerFacts{Name: name} if since > 0 { - t := now.Add(-since) - a.Oldest = &t + a.OldestUnacknowledged = now.Add(-since) } raw, _ := json.Marshal(a) return mapKept{name: raw} @@ -85,7 +49,7 @@ func TestHealth(t *testing.T) { {"four minutes", mapKept{}, age("n_runner", 4*time.Minute), true, ""}, {"six minutes names the consumer", mapKept{}, age("n_runner", 6*time.Minute), false, "n_runner (its oldest unacknowledged message waits 6m0s)"}, {"show does not answer", failingKept{}, age("n_runner", 0), false, "does not answer show"}, - {"no consumer written", mapKept{}, mapKept{}, false, "no consumer's age"}, + {"no event heard yet", mapKept{}, mapKept{}, true, "has heard no event yet"}, } for _, r := range rows { a := Health(r.pipelines, r.consumers, now)