V-3, V-4: the finding writer in go/views (write, clear, statements, asks, reconnect)
Builds: design 47 G §383, §384, §390, §437, §439, §452, §461, §485, §503, §609 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -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 `<Scope>.<ID>.<Kind>` 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`: `<scope>.<id>.<kind>` (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
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user