diff --git a/go/views/writer.go b/go/views/writer.go new file mode 100644 index 00000000..9697dbaf --- /dev/null +++ b/go/views/writer.go @@ -0,0 +1,526 @@ +package views + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "slices" + "sync" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// The finding writer (design 47 G §461, §485, §503): the one writer every observer uses to keep its findings in its own +// state `findings` and to say them as events (§383, §384). It names no seat and no module: the observer hands in the +// subjects of `finding-written`, `finding-cleared`, `findings-stated` and `findings-asked`, which the generated +// contract of the seat that defines them makes from its module name and node (§461, §439). + +// The time constants of the writer (design 47 G §383, §485). +const ( + // PublishRetryFor is how long an event the bus did not acknowledge is sent again before it is logged as an error. + PublishRetryFor = time.Minute + // ObserveRetryEvery is how often a failing observe is called again until a set is said. + ObserveRetryEvery = time.Minute +) + +// FindingSubject is what a finding is about (the `Subject` of design 47 G §439's finding). +type FindingSubject struct { + Scope string `json:"scope"` + ID string `json:"id"` + Machine string `json:"machine,omitempty"` + Also []string `json:"also,omitempty"` +} + +// FindingEvidence is one observation, as it was said. +type FindingEvidence struct { + At time.Time `json:"at"` + Said string `json:"said"` +} + +// FindingAction is one answer the operator may give from a notification. +type FindingAction struct { + Label string `json:"label"` + Verb string `json:"verb"` + Machine string `json:"machine,omitempty"` + Arguments map[string]string `json:"arguments,omitempty"` + Level string `json:"level,omitempty"` +} + +// FindingBody is one finding as an observer says it: the layout whole of design 47 G §439's list, the body of +// `finding-written`, of each entry of `findings-stated` and the value of its key `..` in the state `findings`. +type FindingBody struct { + Kind string `json:"kind"` + Scope string `json:"scope"` + ID string `json:"id"` + At time.Time `json:"at"` + Since time.Time `json:"since"` + Severity string `json:"severity"` + Summary string `json:"summary"` + Headline string `json:"headline,omitempty"` + Explanation string `json:"explanation,omitempty"` + Subject FindingSubject `json:"subject"` + Evidence []FindingEvidence `json:"evidence,omitempty"` + Source string `json:"source"` + Observer string `json:"observer"` + Owner string `json:"owner,omitempty"` + WaitingOn string `json:"waiting-on,omitempty"` + Actions []FindingAction `json:"actions,omitempty"` + Needs string `json:"needs,omitempty"` + Resolver string `json:"resolver,omitempty"` + Ref string `json:"ref,omitempty"` + Run string `json:"run,omitempty"` +} + +// Key is the finding's key in the state `findings`: `..` (novox/hq ADR 0564 §2). +func (f FindingBody) Key() string { return FindingKey(f.Scope, f.ID, f.Kind) } + +// FindingKey is the key of the finding of scope, id and kind. +func FindingKey(scope, id, kind string) string { return scope + "." + id + "." + kind } + +// FindingCleared is the body of `finding-cleared`: an observer no longer observes the finding. +type FindingCleared struct { + Kind string `json:"kind"` + Scope string `json:"scope"` + ID string `json:"id"` + At time.Time `json:"at"` + Why string `json:"why"` +} + +// PendingKey is a key the observer cannot judge yet since its start: its `since` and its bound (a duration such as 5m). +type PendingKey struct { + Key string `json:"key"` + Since time.Time `json:"since"` + Bound string `json:"bound"` +} + +// FindingsStated is the body of `findings-stated`: every finding the observer observes now, and what it cannot judge yet. +type FindingsStated struct { + At time.Time `json:"at"` + Asked string `json:"asked,omitempty"` + Findings []FindingBody `json:"findings"` + Pending []PendingKey `json:"pending"` +} + +// FindingsAsked is the body of `findings-asked`: the keeper asks every observer, or the one named, to state its findings. +type FindingsAsked struct { + Ask string `json:"ask"` + At time.Time `json:"at"` + By string `json:"by"` + Module string `json:"module,omitempty"` +} + +// Observed is what an observer's own truth says now: the findings it observes and the keys it cannot judge yet. +type Observed struct { + Findings []FindingBody + Pending []PendingKey +} + +// FindingState is the observer's own state `findings` (stdio.State and process.State both are one). +type FindingState interface { + Get(key string) (*stdio.StateEntry, error) + Put(key string, value any) (uint64, error) + Delete(key string) error + Keys() ([]string, error) +} + +// WriterSubjects are the subjects of the four events, as the seat's generated contract makes them for the observer's +// module and node. +type WriterSubjects struct { + Written, Cleared, Stated, Asked string +} + +// WriterConfig makes a Writer. +type WriterConfig struct { + // Module is the observer's module name: it fills the observer of a finding and tells an ask for another module. + Module string + // State is the observer's own state `findings`. + State FindingState + // JS is the bus connection's JetStream; Conn, when given, lets the writer hear the connection made again. + JS jetstream.JetStream + Conn *nats.Conn + // Subjects are the four events' subjects. + Subjects WriterSubjects + // Observe answers the findings the observer observes now and its pending keys, or an error. It is called under the + // writer's one lock, so it must not call Write or Clear. + Observe func(ctx context.Context) (Observed, error) + // Send puts a set larger than one bus message in the observer's sent store and answers what to publish instead + // (stateobjects.Sender.Send); nil sends the body as it is. + Send func(label string, body []byte) ([]byte, error) + // Logf is the observer's journal. + Logf func(string, ...any) + // Now is the clock, Wait pauses for d and answers false when ctx ended; nil are the real ones. + Now func() time.Time + Wait func(ctx context.Context, d time.Duration) bool + // Publish sends one message and answers when the bus acknowledged it, and Listen makes a consumer that delivers only + // the asks published after it is made; nil are JS's. + Publish func(ctx context.Context, subject string, data []byte, id string) error + Listen func(ctx context.Context, subject string, handle func(data []byte)) (stop func(), err error) +} + +// Writer keeps an observer's findings and says them (design 47 G §461, §485). Its acts, Write, Clear and a statement, +// run one at a time in the order they are asked. +type Writer struct { + cfg WriterConfig + mu sync.Mutex // the one lock of the acts + seq uint64 + ctl sync.Mutex // guards the consumer, the asks and the loop's context + stop func() + asks []string + want bool + wake chan struct{} + run context.Context + end context.CancelFunc +} + +// NewWriter is the writer for cfg. +func NewWriter(cfg WriterConfig) (*Writer, error) { + switch { + case cfg.Module == "": + return nil, errors.New("a finding writer needs its module's name") + case cfg.State == nil: + return nil, errors.New("a finding writer needs the state `findings`") + case cfg.Observe == nil: + return nil, errors.New("a finding writer needs the observe function") + case cfg.Subjects.Written == "" || cfg.Subjects.Cleared == "" || cfg.Subjects.Stated == "" || cfg.Subjects.Asked == "": + return nil, errors.New("a finding writer needs the subjects of finding-written, finding-cleared, findings-stated and findings-asked") + case cfg.JS == nil && (cfg.Publish == nil || cfg.Listen == nil): + return nil, errors.New("a finding writer needs the bus connection's JetStream") + } + if cfg.Now == nil { + cfg.Now = func() time.Time { return time.Now().UTC() } + } + if cfg.Wait == nil { + cfg.Wait = func(ctx context.Context, d time.Duration) bool { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-ctx.Done(): + return false + case <-t.C: + return true + } + } + } + if cfg.Logf == nil { + cfg.Logf = func(string, ...any) {} + } + if cfg.Publish == nil { + cfg.Publish = jetStreamPublish(cfg.JS) + } + if cfg.Listen == nil { + cfg.Listen = jetStreamListen(cfg.JS) + } + return &Writer{cfg: cfg, wake: make(chan struct{}, 1)}, nil +} + +func jetStreamPublish(js jetstream.JetStream) func(context.Context, string, []byte, string) error { + return func(ctx context.Context, subject string, data []byte, id string) error { + pub, cancel := context.WithTimeout(ctx, 10*time.Second) + defer cancel() + _, err := js.PublishMsg(pub, &nats.Msg{Subject: subject, Data: data}, jetstream.WithMsgID(id)) + return err + } +} + +// jetStreamListen makes an ephemeral consumer of its own on the stream that holds subject, from the next message on. +func jetStreamListen(js jetstream.JetStream) func(context.Context, string, func([]byte)) (func(), error) { + return func(ctx context.Context, subject string, handle func([]byte)) (func(), error) { + stream, err := js.StreamNameBySubject(ctx, subject) + if err != nil { + return nil, fmt.Errorf("the stream of %s: %w", subject, err) + } + cons, err := js.CreateConsumer(ctx, stream, jetstream.ConsumerConfig{ + FilterSubject: subject, DeliverPolicy: jetstream.DeliverNewPolicy, AckPolicy: jetstream.AckNonePolicy, + InactiveThreshold: 10 * time.Minute, + }) + if err != nil { + return nil, fmt.Errorf("a consumer of %s: %w", subject, err) + } + cc, err := cons.Consume(func(m jetstream.Msg) { handle(m.Data()) }) + if err != nil { + return nil, fmt.Errorf("consuming %s: %w", subject, err) + } + return cc.Stop, nil + } +} + +// Start makes the consumer of `findings-asked` and then states the findings now, and again on each ask and each time the +// connection is made again (design 47 G §384, §485). It returns once the consumer exists; the statements run in the +// writer's own goroutine until ctx ends or Close. +func (w *Writer) Start(ctx context.Context) error { + w.ctl.Lock() + defer w.ctl.Unlock() + if w.run != nil { + return errors.New("the finding writer is started") + } + w.run, w.end = context.WithCancel(ctx) + if err := w.listen(); err != nil { + w.end() + w.run = nil + return err + } + w.asks, w.want = append(w.asks, ""), true + if w.cfg.Conn != nil { + prev := w.cfg.Conn.Opts.ReconnectedCB + w.cfg.Conn.SetReconnectHandler(func(nc *nats.Conn) { + if prev != nil { + prev(nc) + } + w.Reconnected() + }) + } + go w.loop(w.run) + w.signal() + return nil +} + +// Close ends the writer's consumer and its statements. +func (w *Writer) Close() { + w.ctl.Lock() + defer w.ctl.Unlock() + if w.stop != nil { + w.stop() + w.stop = nil + } + if w.end != nil { + w.end() + } +} + +func (w *Writer) listen() error { + stop, err := w.cfg.Listen(w.run, w.cfg.Subjects.Asked, w.heard) + if err != nil { + return err + } + if w.stop != nil { + w.stop() + } + w.stop = stop + return nil +} + +// Reconnected is told that the bus connection was made again: the writer makes its consumer again, from the newest ask, +// and states its findings again (design 47 G §485). +func (w *Writer) Reconnected() { + w.ctl.Lock() + defer w.ctl.Unlock() + if w.run == nil || w.run.Err() != nil { + return + } + if err := w.listen(); err != nil { + w.cfg.Logf("the finding writer could not make its consumer of %s again: %v", w.cfg.Subjects.Asked, err) + } + w.asks, w.want = append(w.asks, ""), true + w.signal() +} + +func (w *Writer) signal() { + select { + case w.wake <- struct{}{}: + default: + } +} + +// heard is an ask: answered when it names no module or this one. +func (w *Writer) heard(data []byte) { + var a FindingsAsked + if err := json.Unmarshal(data, &a); err != nil { + w.cfg.Logf("an ask for findings could not be read: %v", err) + return + } + if a.Module != "" && a.Module != w.cfg.Module { + return + } + w.ctl.Lock() + w.asks, w.want = append(w.asks, a.Ask), true + w.ctl.Unlock() + w.signal() +} + +func (w *Writer) loop(ctx context.Context) { + for { + select { + case <-ctx.Done(): + return + case <-w.wake: + } + w.ctl.Lock() + asks, want := w.asks, w.want + w.asks, w.want = nil, false + w.ctl.Unlock() + if !want { + continue + } + asked := "" + if len(asks) > 0 { + asked = asks[len(asks)-1] + } + for { + err := w.State(ctx, asked) + if err == nil { + break + } + w.cfg.Logf("the findings could not be stated, no set said, trying again in %s: %v", ObserveRetryEvery, err) + if !w.cfg.Wait(ctx, ObserveRetryEvery) { + return + } + } + } +} + +func (w *Writer) now() time.Time { return w.cfg.Now().UTC() } + +// unsaid is a finding as compared to say whether what it says changed: not the time it was last seen, nor what saw it. +func unsaid(f FindingBody) string { + f.At, f.Run = time.Time{}, "" + raw, _ := json.Marshal(f) + return string(raw) +} + +func (w *Writer) read(key string) (*FindingBody, error) { + e, err := w.cfg.State.Get(key) + if err != nil { + return nil, err + } + if e == nil { + return nil, nil + } + var old FindingBody + if err := json.Unmarshal(e.Value, &old); err != nil { + return nil, fmt.Errorf("the finding %s in the state findings: %w", key, err) + } + return &old, nil +} + +// complete fills what the observer leaves to the writer and keeps the key's `since` where the key exists (design 47 G §384). +func (w *Writer) complete(f FindingBody) (FindingBody, *FindingBody, error) { + if f.Kind == "" || f.Scope == "" || f.ID == "" { + return f, nil, fmt.Errorf("a finding needs its kind, scope and id: %q.%q.%q", f.Scope, f.ID, f.Kind) + } + old, err := w.read(f.Key()) + if err != nil { + return f, nil, err + } + if f.Observer == "" { + f.Observer = w.cfg.Module + } + if f.At.IsZero() { + f.At = w.now() + } + f.At = f.At.UTC() + switch { + case old != nil && !old.Since.IsZero(): + f.Since = old.Since + case f.Since.IsZero(): + f.Since = f.At + } + f.Since = f.Since.UTC() + return f, old, nil +} + +// Write keeps the finding in its key and says `finding-written` only when what the finding says changed (design 47 G +// §383, §461). The key's `since` stays where the key exists. +func (w *Writer) Write(ctx context.Context, f FindingBody) error { + w.mu.Lock() + defer w.mu.Unlock() + f, old, err := w.complete(f) + if err != nil { + return err + } + if _, err := w.cfg.State.Put(f.Key(), f); err != nil { + return fmt.Errorf("writing the finding %s: %w", f.Key(), err) + } + if old != nil && unsaid(*old) == unsaid(f) { + return nil + } + raw, _ := json.Marshal(f) + w.say(ctx, w.cfg.Subjects.Written, "finding-written", f.Key(), raw) + return nil +} + +// Clear deletes the finding's key and says `finding-cleared`; a key that does not exist is deleted of nothing and is said +// all the same, so a keeper holding the finding clears it (design 47 G §485). +func (w *Writer) Clear(ctx context.Context, scope, id, kind, why string) error { + w.mu.Lock() + defer w.mu.Unlock() + key := FindingKey(scope, id, kind) + if err := w.cfg.State.Delete(key); err != nil { + return fmt.Errorf("clearing the finding %s: %w", key, err) + } + raw, _ := json.Marshal(FindingCleared{Kind: kind, Scope: scope, ID: id, At: w.now(), Why: why}) + w.say(ctx, w.cfg.Subjects.Cleared, "finding-cleared", key, raw) + return nil +} + +// State judges every finding anew with observe, writes each stated finding's key, deletes each key neither stated nor +// pending, and says `findings-stated` (design 47 G §384, §461); asked is the ask it answers, none at a start. An error of +// observe says no set. +func (w *Writer) State(ctx context.Context, asked string) error { + w.mu.Lock() + defer w.mu.Unlock() + obs, err := w.cfg.Observe(ctx) + if err != nil { + return fmt.Errorf("observe: %w", err) + } + keep := map[string]bool{} + body := FindingsStated{At: w.now(), Asked: asked, Findings: []FindingBody{}, Pending: []PendingKey{}} + for _, f := range obs.Findings { + f, _, err := w.complete(f) + if err != nil { + return err + } + if _, err := w.cfg.State.Put(f.Key(), f); err != nil { + return fmt.Errorf("writing the finding %s: %w", f.Key(), err) + } + keep[f.Key()] = true + body.Findings = append(body.Findings, f) + } + for _, p := range obs.Pending { + keep[p.Key] = true + body.Pending = append(body.Pending, p) + } + keys, err := w.cfg.State.Keys() + if err != nil { + return fmt.Errorf("the keys of the state findings: %w", err) + } + slices.Sort(keys) + for _, k := range keys { + if !keep[k] { + if err := w.cfg.State.Delete(k); err != nil { + return fmt.Errorf("deleting the finding %s: %w", k, err) + } + } + } + raw, _ := json.Marshal(body) + if w.cfg.Send != nil { + if raw, err = w.cfg.Send("findings-stated."+w.cfg.Module, raw); err != nil { + w.cfg.Logf("the findings-stated of %s could not be sent, the next statement covers it: %v", w.cfg.Module, err) + return nil + } + } + w.say(ctx, w.cfg.Subjects.Stated, "findings-stated", w.cfg.Module, raw) + return nil +} + +// say publishes one event awaiting the bus's acknowledgement, sent again for PublishRetryFor when unacknowledged, then +// written in the journal as an error naming what it said (design 47 G §383); the next statement covers it. +func (w *Writer) say(ctx context.Context, subject, event, what string, data []byte) { + w.seq++ + id := fmt.Sprintf("%s-%s-%d-%d", w.cfg.Module, event, w.now().UnixNano(), w.seq) + start := w.now() + for delay := time.Second; ; delay = min(delay*2, 10*time.Second) { + err := w.cfg.Publish(ctx, subject, data, id) + if err == nil { + return + } + if w.now().Sub(start) >= PublishRetryFor || ctx.Err() != nil || !w.cfg.Wait(ctx, delay) { + w.cfg.Logf("error: %s for %s was not acknowledged by the bus, the next statement covers it: %v", event, what, err) + return + } + } +} diff --git a/go/views/writer_integration_test.go b/go/views/writer_integration_test.go new file mode 100644 index 00000000..41c743bb --- /dev/null +++ b/go/views/writer_integration_test.go @@ -0,0 +1,92 @@ +//go:build integration + +package views_test + +import ( + "context" + "encoding/json" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/testbus" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// Against a real bus: the writer's events are acknowledged by their stream, a start states what the observer observes, an +// ask published before the writer's consumer existed is never answered, one published after is, and after the connection +// is made again (Reconnected) the consumer is made again from the newest ask (design 47 G §383, §390, §485). +func TestWriterSaysItsEventsAndHearsAsksOnTheBus(t *testing.T) { + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, err := jetstream.New(nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + for name, subj := range map[string]string{"EVENTS": "obs.event.>", "ASKS": "obs.ask"} { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: []string{subj}}); err != nil { + t.Fatal(err) + } + } + if _, err := js.Publish(ctx, "obs.ask", []byte(`{"ask":"old","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil { + t.Fatal(err) + } + listen, err := js.CreateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{AckPolicy: jetstream.AckNonePolicy}) + if err != nil { + t.Fatal(err) + } + state := &fakeState{m: map[string][]byte{}} + w, err := views.NewWriter(views.WriterConfig{Module: "m", State: state, JS: js, Conn: nc, + Subjects: views.WriterSubjects{Written: "obs.event.written", Cleared: "obs.event.cleared", Stated: "obs.event.stated", Asked: "obs.ask"}, + Observe: func(context.Context) (views.Observed, error) { + return views.Observed{Findings: []views.FindingBody{finding("a", "now")}}, nil + }, Logf: t.Logf}) + if err != nil { + t.Fatal(err) + } + if err := w.Start(ctx); err != nil { + t.Fatal(err) + } + defer w.Close() + next := func(want string) views.FindingsStated { + t.Helper() + m, err := listen.Next(jetstream.FetchMaxWait(10 * time.Second)) + if err != nil { + t.Fatal(err) + } + if m.Subject() != want { + t.Fatalf("said %s, want %s", m.Subject(), want) + } + var st views.FindingsStated + if want == "obs.event.stated" { + if err := json.Unmarshal(m.Data(), &st); err != nil { + t.Fatal(err) + } + } + return st + } + if st := next("obs.event.stated"); st.Asked != "" || len(st.Findings) != 1 { + t.Fatalf("start statement: %+v", st) + } + if err := w.Write(ctx, finding("b", "x")); err != nil { + t.Fatal(err) + } + next("obs.event.written") + for _, ask := range []string{"new", "again"} { + w.Reconnected() + next("obs.event.stated") // the statement after the connection was made again + if _, err := js.Publish(ctx, "obs.ask", []byte(`{"ask":"`+ask+`","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil { + t.Fatal(err) + } + if st := next("obs.event.stated"); st.Asked != ask { + t.Fatalf("answered %q, want %q", st.Asked, ask) + } + } +} diff --git a/go/views/writer_test.go b/go/views/writer_test.go new file mode 100644 index 00000000..85fe34c9 --- /dev/null +++ b/go/views/writer_test.go @@ -0,0 +1,315 @@ +package views_test + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "runtime" + "slices" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/views" +) + +// fakeState is an observer's state `findings` in memory. +type fakeState struct { + mu sync.Mutex + m map[string][]byte +} + +func (s *fakeState) Get(key string) (*stdio.StateEntry, error) { + s.mu.Lock() + defer s.mu.Unlock() + if v, ok := s.m[key]; ok { + return &stdio.StateEntry{Key: key, Value: v}, nil + } + return nil, nil +} +func (s *fakeState) Put(key string, value any) (uint64, error) { + raw, _ := json.Marshal(value) + s.mu.Lock() + defer s.mu.Unlock() + s.m[key] = raw + return 1, nil +} +func (s *fakeState) Delete(key string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.m, key) + return nil +} +func (s *fakeState) Keys() ([]string, error) { + s.mu.Lock() + defer s.mu.Unlock() + var k []string + for key := range s.m { + k = append(k, key) + } + return k, nil +} + +type said struct { + subject string + data []byte +} + +type rig struct { + t *testing.T + state *fakeState + now time.Time + mu sync.Mutex + sent []said + fail int // publishes refused before one is taken + waits []time.Duration + logs []string + observe func(context.Context) (views.Observed, error) + handle func([]byte) + listens int + stopped int + w *views.Writer + ch chan struct{} +} + +var subjects = views.WriterSubjects{ + Written: "mesh.stream.m.event.finding-written.n1", Cleared: "mesh.stream.m.event.finding-cleared.n1", + Stated: "mesh.stream.m.event.findings-stated.n1", Asked: "mesh.seat.s.event.findings-asked", +} + +func newRig(t *testing.T) *rig { + r := &rig{t: t, state: &fakeState{m: map[string][]byte{}}, now: time.Date(2026, 11, 11, 12, 0, 0, 0, time.UTC), ch: make(chan struct{}, 100)} + r.observe = func(context.Context) (views.Observed, error) { return views.Observed{}, nil } + w, err := views.NewWriter(views.WriterConfig{ + Module: "m", State: r.state, Subjects: subjects, + Observe: func(ctx context.Context) (views.Observed, error) { return r.observe(ctx) }, + Logf: func(f string, a ...any) { + r.mu.Lock() + r.logs = append(r.logs, fmt.Sprintf(f, a...)) + r.mu.Unlock() + }, + Now: func() time.Time { r.mu.Lock(); defer r.mu.Unlock(); return r.now }, + Wait: func(ctx context.Context, d time.Duration) bool { + r.mu.Lock() + defer r.mu.Unlock() + r.waits = append(r.waits, d) + r.now = r.now.Add(d) + return ctx.Err() == nil + }, + Publish: func(_ context.Context, subject string, data []byte, _ string) error { + r.mu.Lock() + defer r.mu.Unlock() + if r.fail > 0 { + r.fail-- + return errors.New("no ack") + } + r.sent = append(r.sent, said{subject, data}) + r.ch <- struct{}{} + return nil + }, + Listen: func(_ context.Context, subject string, handle func([]byte)) (func(), error) { + r.mu.Lock() + defer r.mu.Unlock() + if subject != subjects.Asked { + t.Errorf("listening on %s", subject) + } + r.listens++ + r.handle = handle + return func() { r.mu.Lock(); r.stopped++; r.mu.Unlock() }, nil + }, + }) + if err != nil { + t.Fatal(err) + } + r.w = w + return r +} + +func (r *rig) events() []said { + r.mu.Lock() + defer r.mu.Unlock() + return slices.Clone(r.sent) +} + +func finding(id, summary string) views.FindingBody { + return views.FindingBody{Kind: "k", Scope: "module", ID: id, Severity: "warning", Summary: summary, Source: "probe", + Subject: views.FindingSubject{Scope: "module", ID: id}} +} + +// A finding written says its event with the layout whole after the key, keeps the key's since, and says again only when +// what it says changed; a finding cleared says its event, also of a key that does not exist (design 47 G §383, §485). +func TestWriterWritesAndClearsFindingsAsEvents(t *testing.T) { + r := newRig(t) + ctx := context.Background() + first := r.now + if err := r.w.Write(ctx, finding("a", "one")); err != nil { + t.Fatal(err) + } + r.now = r.now.Add(time.Hour) + f := finding("a", "one") + f.Run = "other run" + if err := r.w.Write(ctx, f); err != nil { // nothing said changed + t.Fatal(err) + } + if err := r.w.Write(ctx, finding("a", "two")); err != nil { + t.Fatal(err) + } + ev := r.events() + if len(ev) != 2 || ev[0].subject != subjects.Written || ev[1].subject != subjects.Written { + t.Fatalf("events: %+v", ev) + } + var got views.FindingBody + if err := json.Unmarshal(ev[1].data, &got); err != nil { + t.Fatal(err) + } + if got.Summary != "two" || !got.Since.Equal(first) || got.Observer != "m" || got.Key() != "module.a.k" || !got.At.Equal(r.now) { + t.Fatalf("body: %+v", got) + } + if e, _ := r.state.Get("module.a.k"); e == nil { + t.Fatal("the key was not written") + } + if err := r.w.Clear(ctx, "module", "a", "k", "gone"); err != nil { + t.Fatal(err) + } + if err := r.w.Clear(ctx, "module", "never", "k", "held by the keeper"); err != nil { + t.Fatal(err) + } + ev = r.events() + var cl views.FindingCleared + if len(ev) != 4 || ev[3].subject != subjects.Cleared || json.Unmarshal(ev[3].data, &cl) != nil || cl.ID != "never" || cl.Why != "held by the keeper" { + t.Fatalf("clear events: %+v", ev) + } + if e, _ := r.state.Get("module.a.k"); e != nil { + t.Fatal("the key was not deleted") + } +} + +// An event the bus does not acknowledge is sent again for one minute, then written in the journal as an error naming the +// finding (design 47 G §383). +func TestWriterSendsAgainForAMinuteThenLogs(t *testing.T) { + r := newRig(t) + r.fail = 2 + if err := r.w.Write(context.Background(), finding("a", "one")); err != nil || len(r.events()) != 1 { + t.Fatalf("resent: %v %+v", err, r.events()) + } + r.fail = 1000 + start := r.now + if err := r.w.Write(context.Background(), finding("b", "one")); err != nil { + t.Fatal(err) + } + if r.now.Sub(start) < views.PublishRetryFor || r.now.Sub(start) > views.PublishRetryFor+10*time.Second { + t.Fatalf("resent for %s", r.now.Sub(start)) + } + if len(r.logs) != 1 || !strings.Contains(r.logs[0], "module.b.k") || !strings.Contains(r.logs[0], "error") { + t.Fatalf("journal: %v", r.logs) + } +} + +// A start states the findings observed now, writes their keys, deletes the keys neither stated nor pending, and pends the +// rest; an ask naming another module is not answered, one naming none or this one is, with its id; an observe that fails +// says no set and is called again once a minute until a set is said; a connection made again makes the consumer again and +// states again (design 47 G §384, §461, §485). +func TestWriterStatesAtStartAndOnAsksAndRecovers(t *testing.T) { + r := newRig(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + r.state.Put("module.old.k", finding("old", "x")) + r.state.Put("module.wait.k", finding("wait", "x")) + first := r.now + r.state.Put("module.a.k", func() views.FindingBody { f := finding("a", "before"); f.Since = first.Add(-time.Hour); return f }()) + calls := 0 + r.observe = func(context.Context) (views.Observed, error) { + calls++ + if calls <= 2 { + return views.Observed{}, errors.New("not yet") + } + return views.Observed{Findings: []views.FindingBody{finding("a", "now")}, + Pending: []views.PendingKey{{Key: "module.wait.k", Since: first, Bound: "5m"}}}, nil + } + if err := r.w.Start(ctx); err != nil { + t.Fatal(err) + } + defer r.w.Close() + stated := waitEvent(t, r, 1) + if calls != 3 || len(r.waits) != 2 || r.waits[0] != views.ObserveRetryEvery { + t.Fatalf("observe called %d times, waits %v", calls, r.waits) + } + var st views.FindingsStated + if err := json.Unmarshal(stated.data, &st); err != nil { + t.Fatal(err) + } + if stated.subject != subjects.Stated || st.Asked != "" || len(st.Findings) != 1 || st.Findings[0].Summary != "now" || + !st.Findings[0].Since.Equal(first.Add(-time.Hour)) || len(st.Pending) != 1 || st.Pending[0].Key != "module.wait.k" { + t.Fatalf("statement: %s", stated.data) + } + keys, _ := r.state.Keys() + slices.Sort(keys) + if !slices.Equal(keys, []string{"module.a.k", "module.wait.k"}) { + t.Fatalf("keys: %v", keys) + } + r.handle([]byte(`{"ask":"a1","at":"2026-11-11T12:00:00Z","by":"keeper","module":"other"}`)) + r.handle([]byte(`{"ask":"a2","at":"2026-11-11T12:00:00Z","by":"keeper","module":"m"}`)) + st = FindingsStatedOf(t, waitEvent(t, r, 2)) + if st.Asked != "a2" || len(r.events()) != 2 { + t.Fatalf("ask answered: %+v %d", st, len(r.events())) + } + r.w.Reconnected() + waitEvent(t, r, 3) + if r.listens != 2 || r.stopped != 1 { + t.Fatalf("consumers made %d, stopped %d", r.listens, r.stopped) + } +} + +func FindingsStatedOf(t *testing.T, e said) views.FindingsStated { + var st views.FindingsStated + if err := json.Unmarshal(e.data, &st); err != nil { + t.Fatal(err) + } + return st +} + +// waitEvent is the n-th event said, woken by each event said (nothing sleeps for time). +func waitEvent(t *testing.T, r *rig, n int) said { + t.Helper() + for len(r.events()) < n { + select { + case <-r.ch: + case <-time.After(10 * time.Second): + t.Fatalf("event %d never said", n) + } + } + return r.events()[n-1] +} + +// A write asked during a statement waits for it, so the statement never deletes the key the write made (design 47 G §485). +func TestWriterRunsOneActAtATime(t *testing.T) { + r := newRig(t) + inside, release := make(chan struct{}), make(chan struct{}) + r.observe = func(context.Context) (views.Observed, error) { + close(inside) + <-release + return views.Observed{}, nil + } + done := make(chan error, 1) + go func() { done <- r.w.State(context.Background(), "") }() + <-inside + wrote, started := make(chan error, 1), make(chan struct{}) + go func() { close(started); wrote <- r.w.Write(context.Background(), finding("a", "one")) }() + <-started + for range 1000 { + runtime.Gosched() + } + close(release) + if err := <-done; err != nil { + t.Fatal(err) + } + if err := <-wrote; err != nil { + t.Fatal(err) + } + if e, _ := r.state.Get("module.a.k"); e == nil { + t.Fatal("the statement deleted the key the write made") + } +}