diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index a1ee06a..5480691 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -1098,6 +1098,17 @@ type Beat struct { // the bus's own heartbeat to the loop stopped arriving, the bus ended the loop or no longer answers. A Beat that is // not Missed follows the first answer after one that was. A nil beat says nothing. func (c *Conn) ConsumeAsBeating(module string, deliver func(Envelope) error, beat func(Beat)) (func(), error) { + return c.ConsumeAsBeatingWhen(module, deliver, beat, nil) +} + +// tickHook, when set before a consume starts, is called after each idle heartbeat interval was handled (tests only, +// through SetTickHook), so a test counts beats instead of sleeping. A consume reads it once, when it starts. +var tickHook func() + +// ConsumeAsBeatingWhen is ConsumeAsBeating that asks the bus about the consumer, and says anything, only while wanted +// answers true (nil is always): a module whose children asked for no heartbeat costs the bus no consumer-info request. +func (c *Conn) ConsumeAsBeatingWhen(module string, deliver func(Envelope) error, beat func(Beat), wanted func() bool) (func(), error) { + asked := func() bool { return beat != nil && (wanted == nil || wanted()) } durable := ConsumerOf(c.node, module) js, err := jetstream.New(c.nc) if err != nil { @@ -1106,7 +1117,7 @@ func (c *Conn) ConsumeAsBeating(module string, deliver func(Envelope) error, bea var lastEvent atomic.Int64 var missed atomic.Bool say := func(why string) { - if beat != nil && missed.CompareAndSwap(false, true) { + if asked() && missed.CompareAndSwap(false, true) { beat(Beat{At: time.Now().UTC(), Missed: true, Why: why}) } } @@ -1130,6 +1141,30 @@ func (c *Conn) ConsumeAsBeating(module string, deliver func(Envelope) error, bea h.start() if beat != nil { every := IdleHeartbeat + hook := tickHook + tickOnce := func() { + select { + case <-events.Closed(): + say("the bus ended reading the consumer " + durable) + return + default: + } + if !asked() { + return + } + ctx, cancel := context.WithTimeout(context.Background(), every) + _, err := js.Consumer(ctx, EventsStream, durable) + cancel() + if err != nil { + say("the bus does not answer about the consumer " + durable + ": " + err.Error()) + return + } + if time.Since(time.Unix(0, lastEvent.Load())) < every { + return // an event arrived in this interval: it was the word + } + missed.Store(false) + beat(Beat{At: time.Now().UTC()}) + } go func() { tick := time.NewTicker(every) defer tick.Stop() @@ -1139,24 +1174,10 @@ func (c *Conn) ConsumeAsBeating(module string, deliver func(Envelope) error, bea return case <-tick.C: } - select { - case <-events.Closed(): - say("the bus ended reading the consumer " + durable) - continue - default: + tickOnce() + if hook != nil { + hook() } - ctx, cancel := context.WithTimeout(context.Background(), every) - _, err := js.Consumer(ctx, EventsStream, durable) - cancel() - if err != nil { - say("the bus does not answer about the consumer " + durable + ": " + err.Error()) - continue - } - if time.Since(time.Unix(0, lastEvent.Load())) < every { - continue // an event arrived in this interval: it was the word - } - missed.Store(false) - beat(Beat{At: time.Now().UTC()}) } }() } diff --git a/node-tools/internal/bus/consume_test.go b/node-tools/internal/bus/consume_test.go index 66acd3c..e5cc2e5 100644 --- a/node-tools/internal/bus/consume_test.go +++ b/node-tools/internal/bus/consume_test.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "sync" + "sync/atomic" "testing" "time" @@ -68,6 +69,13 @@ func TestAConsumerSaysItIsBoundWhileIdleAndOnceWhenTheBusEndedIt(t *testing.T) { restore := bus.IdleHeartbeat bus.IdleHeartbeat = 200 * time.Millisecond t.Cleanup(func() { bus.IdleHeartbeat = restore }) + ticks := make(chan struct{}, 64) + t.Cleanup(bus.SetTickHook(func() { + select { + case ticks <- struct{}{}: + default: + } + })) mesh.Consumer(t, "anchor", "quiet", []string{"mesh.mod.alpha.event.>"}, time.Second) c, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"}) if err != nil { @@ -87,6 +95,14 @@ func TestAConsumerSaysItIsBoundWhileIdleAndOnceWhenTheBusEndedIt(t *testing.T) { t.Fatal(err) } t.Cleanup(stop) + // A consume nobody asked a heartbeat of says nothing and asks the bus nothing, however many intervals pass. + var unaskedBeats atomic.Int32 + unasked, err := c.ConsumeAsBeatingWhen("quiet", func(bus.Envelope) error { return nil }, + func(bus.Beat) { unaskedBeats.Add(1) }, func() bool { return false }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(unasked) last := func() (bus.Beat, int) { mu.Lock() defer mu.Unlock() @@ -116,7 +132,13 @@ func TestAConsumerSaysItIsBoundWhileIdleAndOnceWhenTheBusEndedIt(t *testing.T) { } } mu.Unlock() - time.Sleep(600 * time.Millisecond) + for n := 0; n < 3; n++ { // three more intervals pass, counted by the loop's own tick + select { + case <-ticks: + case <-time.After(10 * time.Second): + t.Fatal("the heartbeat loop handled no more intervals") + } + } mu.Lock() defer mu.Unlock() after := 0 @@ -125,6 +147,9 @@ func TestAConsumerSaysItIsBoundWhileIdleAndOnceWhenTheBusEndedIt(t *testing.T) { after++ } } + if unaskedBeats.Load() != 0 { + t.Errorf("a consume with no heartbeat asked said %d words", unaskedBeats.Load()) + } if after != missed || missed != 1 { t.Errorf("Missed is said once, not at every beat: %d then %d", missed, after) } diff --git a/node-tools/internal/bus/export_test.go b/node-tools/internal/bus/export_test.go new file mode 100644 index 0000000..92420cd --- /dev/null +++ b/node-tools/internal/bus/export_test.go @@ -0,0 +1,8 @@ +package bus + +// SetTickHook makes the next consumes call f after each idle heartbeat interval was handled; the restore puts it back. +func SetTickHook(f func()) (restore func()) { + old := tickHook + tickHook = f + return func() { tickHook = old } +} diff --git a/node-tools/internal/bus/seat.go b/node-tools/internal/bus/seat.go index 5a84212..1ef7fc1 100644 --- a/node-tools/internal/bus/seat.go +++ b/node-tools/internal/bus/seat.go @@ -370,7 +370,13 @@ func (c *Conn) SeatRecord(module, bucket, key string) (*StateEntry, error) { } var revision uint64 fmt.Sscan(reply.Header.Get("Nats-Sequence"), &revision) - return &StateEntry{Key: module + "." + key, Value: valueOf(reply.Data), Revision: revision}, nil + var at time.Time + if stamp := reply.Header.Get("Nats-Time-Stamp"); stamp != "" { + if t, err := time.Parse(time.RFC3339Nano, stamp); err == nil { + at = t.UTC() + } + } + return &StateEntry{Key: module + "." + key, Value: valueOf(reply.Data), Revision: revision, At: at}, nil } // StateCreate writes one key only when it has no value, and answers its revision; a key that has one is diff --git a/node-tools/internal/bus/seat_test.go b/node-tools/internal/bus/seat_test.go index fe49a19..d9de59d 100644 --- a/node-tools/internal/bus/seat_test.go +++ b/node-tools/internal/bus/seat_test.go @@ -213,6 +213,9 @@ func TestAnAskerReadsItsOwnRecordAndNoOther(t *testing.T) { if err != nil || got == nil || string(got.Value) != `{"state":"open"}` { t.Fatalf("the asker could not read its own ask: %v %v", got, err) } + if got.At.IsZero() || time.Since(got.At) > time.Minute { + t.Errorf("the record does not say when the bus wrote it: at %v", got.At) + } if missing, err := conn.SeatRecord("mesh-delivery", "messenger_asks", "a2"); err != nil || missing != nil { t.Fatalf("a missing ask read as %v, %v", missing, err) } diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index 78fc4a8..c4b23c9 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -72,8 +72,13 @@ type Deps struct { // Start asks the four first views, each until answered, and follows the owners' views families on the runner's own // consumers. It returns at once: the picture fills as the owners answer. func Start(parent context.Context, conn *bus.Conn, logf func(string, ...any)) (*Views, func()) { + return StartWith(parent, DepsOf(conn), logf) +} + +// DepsOf is the views' reach through the runner's own connection: its tool calls, its consumer and its findings state. +func DepsOf(conn *bus.Conn) Deps { me := conn.Module() - return StartWith(parent, Deps{Module: me, + return Deps{Module: me, Ask: func(address string, body any) (json.RawMessage, error) { got, err := conn.Ask(address, body, "") return got.Result, err @@ -84,7 +89,7 @@ func Start(parent context.Context, conn *bus.Conn, logf func(string, ...any)) (* return err }, DeleteFinding: func(key string) error { return conn.StateDelete(me, FindingsState, key) }, - }, logf) + } } // StartWith is Start over deps. diff --git a/node-tools/internal/firstview/firstview_integration_test.go b/node-tools/internal/firstview/firstview_integration_test.go new file mode 100644 index 0000000..c76a5f3 --- /dev/null +++ b/node-tools/internal/firstview/firstview_integration_test.go @@ -0,0 +1,129 @@ +//go:build integration + +package firstview + +import ( + "context" + "encoding/json" + "os" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/views" + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// issue 1966 TLS-5 [i] (ADR 0615 §2, §4, §6): against the real bus, with the owners down and then up, the runner's own +// consumer carries the manifest's four `consumes` to the owners' views events: the first views are asked until answered, +// the unanswered ones are findings that the answer clears, and each owner's next event reaches its picture. +func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThenUp(t *testing.T) { + mesh := mt.New(t) + url := mt.URL(t) + + // The consumer is the mesh's, made from the manifest's `consumes` as the mesh makes it: one filter per family. + raw, err := os.ReadFile("../../module.json") + if err != nil { + t.Fatal(err) + } + var manifest struct { + Consumes []string `json:"consumes"` + } + if err := json.Unmarshal(raw, &manifest); err != nil { + t.Fatal(err) + } + var filters []string + for _, c := range manifest.Consumes { + emitter, event, _ := strings.Cut(c, ".views.") + filters = append(filters, "mesh.mod."+emitter+".event.views."+event) + } + if len(filters) != len(Specs) { + t.Fatalf("the manifest consumes %d families, the runner holds %d first views", len(filters), len(Specs)) + } + mesh.Consumer(t, "anchor", "node-tools", filters, 2*time.Second) + + conn, err := bus.Connect(bus.Credential{URL: url, Module: "node-tools", Node: "anchor"}) + if err != nil { + t.Fatal(err) + } + conn.Logf = func(string, ...any) {} + t.Cleanup(conn.Close) + + var mu sync.Mutex + findings := map[string]bool{} + deps := DepsOf(conn) + deps.Bound, deps.Wait = 100*time.Millisecond, 20*time.Millisecond + deps.PutFinding = func(key string, _ json.RawMessage) error { mu.Lock(); findings[key] = true; mu.Unlock(); return nil } + deps.DeleteFinding = func(key string) error { mu.Lock(); delete(findings, key); mu.Unlock(); return nil } + v, stop := StartWith(context.Background(), deps, func(string, ...any) {}) + t.Cleanup(stop) + + // The owners are down: no tool answers, nothing is claimed, and past the bound each view is a finding. + if _, err := v.State(Assigner + ".placement"); err == nil { + t.Fatal("a state its owner has not answered is claimed") + } + mt.Until(t, func() error { + mu.Lock() + defer mu.Unlock() + if len(findings) != len(Specs) { + return errFindings(len(findings)) + } + return nil + }) + + // The owners come up and answer their first views. + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + for _, s := range Specs { + s := s + if _, err := nc.Subscribe("mesh.mod."+s.Owner+".tool."+s.Tool, func(m *nats.Msg) { + answer, _ := json.Marshal(views.Answer{V: 1, Owner: s.Owner, State: s.State, Revision: 4, Created: time.Unix(1000, 0).UTC(), + Values: map[string]json.RawMessage{"first": json.RawMessage(`{"n":1}`)}}) + body, _ := json.Marshal(map[string]json.RawMessage{"result": answer}) + _ = m.Respond(body) + }); err != nil { + t.Fatal(err) + } + } + _ = nc.Flush() + for _, s := range Specs { + name := s.Name() + mt.Until(t, func() error { _, err := v.Entry(name, "first"); return err }) + } + mt.Until(t, func() error { + mu.Lock() + defer mu.Unlock() + if len(findings) != 0 { + return errFindings(len(findings)) + } + return nil + }) + + // Each owner's next event arrives on the runner's consumer: all four families are bound. + for _, s := range Specs { + ev, _ := json.Marshal(views.Event{V: 1, Owner: s.Owner, State: s.State, Key: "next", Revision: 5, Created: time.Unix(1000, 0).UTC(), + Value: json.RawMessage(`{"n":2}`)}) + mesh.Emit(t, "mesh.mod."+s.Owner+".event.views."+s.State, json.RawMessage(ev)) + } + for _, s := range Specs { + name := s.Name() + mt.Until(t, func() error { + got, err := v.Entry(name, "next") + if err != nil || string(got) != `{"n":2}` { + return errFindings(-1) + } + return nil + }) + } +} + +type errFindings int + +func (e errFindings) Error() string { return "not yet" } diff --git a/node-tools/internal/runtime/events_test.go b/node-tools/internal/runtime/events_test.go index bea321b..b2b7ebf 100644 --- a/node-tools/internal/runtime/events_test.go +++ b/node-tools/internal/runtime/events_test.go @@ -1,11 +1,13 @@ package runtime import ( + "encoding/json" "fmt" "os" "path/filepath" "strings" "sync" + "sync/atomic" "testing" "time" @@ -157,3 +159,30 @@ func TestALongRunningBundleThatExitsIsStartedAgain(t *testing.T) { func fmtSprintf(format string, args ...any) string { return fmt.Sprintf(format, args...) } func errorf(format string, args ...any) error { return fmt.Errorf(format, args...) } + +// A bundle loaded again sets its heartbeat handler while the consumer's loop hands a word on: no data race (-race), +// and the heartbeat is asked of the bus only while some bundle holds a handler. +func TestAHeartbeatHandlerSetAgainWhileWordsAreHandedOnIsNotARace(t *testing.T) { + m := &moduleEvents{} + all := &consumers{logf: t.Logf, of: map[string]*moduleEvents{"mod": m}} + slot := &beatSlot{} + m.beats = []*beatSlot{slot} + if m.beatAsked() { + t.Fatal("a heartbeat is asked while no bundle gave a handler") + } + var got atomic.Int32 + done := make(chan struct{}) + go func() { + defer close(done) + for i := 0; i < 200; i++ { + slot.set(func(json.RawMessage) { got.Add(1) }) + } + }() + for i := 0; i < 200; i++ { + all.handBeat(m, bus.Beat{At: time.Now()}) + } + <-done + if !m.beatAsked() { + t.Error("a bundle gave a handler and no heartbeat is asked") + } +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 2960102..597fffb 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -775,7 +775,36 @@ type consumers struct { type moduleEvents struct { stop func() delivers []*func(json.RawMessage) error - beats []*func(json.RawMessage) + beats []*beatSlot +} + +// beatSlot is where a bundle's heartbeat handler is kept; a bundle loaded again sets it while the consumer's loop +// reads it, so both go through its own lock. +type beatSlot struct { + mu sync.Mutex + f func(json.RawMessage) +} + +func (s *beatSlot) get() func(json.RawMessage) { + s.mu.Lock() + defer s.mu.Unlock() + return s.f +} + +func (s *beatSlot) set(f func(json.RawMessage)) { + s.mu.Lock() + defer s.mu.Unlock() + s.f = f +} + +// beatAsked says whether any bundle of the module asked for the heartbeat; the caller holds c.mu. +func (m *moduleEvents) beatAsked() bool { + for _, t := range m.beats { + if t.get() != nil { + return true + } + } + return false } // stopAll unbinds every module's consumer. @@ -823,7 +852,7 @@ type moduleBus struct { module string mu sync.Mutex deliver *func(json.RawMessage) error - beat *func(json.RawMessage) + beat *beatSlot } // Publish emits an event as the module (ADR 0193). @@ -871,11 +900,9 @@ func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error, beat func(jso } mine := b.deliver if b.beat == nil { - x := beat - b.beat = &x - } else { - *b.beat = beat + b.beat = &beatSlot{} } + b.beat.set(beat) mineBeat := b.beat b.mu.Unlock() @@ -901,8 +928,13 @@ func (b *moduleBus) Subscribe(deliver func(json.RawMessage) error, beat func(jso return nil } module := b.module - stop, err := c.conn.ConsumeAsBeating(module, func(env bus.Envelope) error { return c.handOut(module, m, env) }, - func(b bus.Beat) { c.handBeat(m, b) }) + // The bus is asked about the consumer only while a bundle of the module asked for the heartbeat. + stop, err := c.conn.ConsumeAsBeatingWhen(module, func(env bus.Envelope) error { return c.handOut(module, m, env) }, + func(b bus.Beat) { c.handBeat(m, b) }, func() bool { + c.mu.Lock() + defer c.mu.Unlock() + return m.beatAsked() + }) if err != nil { return err } @@ -918,10 +950,10 @@ func (c *consumers) handBeat(m *moduleEvents, b bus.Beat) { return } c.mu.Lock() - targets := append([](*func(json.RawMessage))(nil), m.beats...) + targets := append([]*beatSlot(nil), m.beats...) c.mu.Unlock() for _, t := range targets { - if f := *t; f != nil { + if f := t.get(); f != nil { f(raw) } }