diff --git a/node-tools/internal/bus/bindings.go b/node-tools/internal/bus/bindings.go index 8937a61..3a441bf 100644 --- a/node-tools/internal/bus/bindings.go +++ b/node-tools/internal/bus/bindings.go @@ -58,15 +58,16 @@ func (c *Conn) Bindings(module string) []membership.BusBinding { return nil } -// jetStream is a JetStream handle on this connection. -func (c *Conn) jetStream() (jetstream.JetStream, error) { return jetstream.New(c.nc) } +// JetStream is a JetStream handle on this connection: its one accessor, for a part that hands it to a finding writer (design 47 G +// A-16, A-27). +func (c *Conn) JetStream() (jetstream.JetStream, error) { return jetstream.New(c.nc) } func (c *Conn) objectStore(ctx context.Context, module, local string) (jetstream.ObjectStore, membership.BusBinding, error) { b, err := c.Binding(module, local, catalogue.StorageObjectStore) if err != nil { return nil, b, err } - js, err := c.jetStream() + js, err := c.JetStream() if err != nil { return nil, b, err } @@ -238,7 +239,7 @@ func (c *Conn) ObjectsWatch(module string, r buscalls.ObjectsWatchRequest, deliv func (c *Conn) streamFill(name string) (buscalls.StreamFilled, error) { ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() - js, err := c.jetStream() + js, err := c.JetStream() if err != nil { return buscalls.StreamFilled{}, err } @@ -301,7 +302,7 @@ func (c *Conn) readStream(name, filter string, from uint64, limit int) (buscalls } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() - js, err := c.jetStream() + js, err := c.JetStream() if err != nil { return buscalls.StreamPage{}, err } @@ -361,7 +362,7 @@ func (c *Conn) WorkQueueDead(module string, r buscalls.WorkQueueDeadRequest) (bu } ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() - js, err := c.jetStream() + js, err := c.JetStream() if err != nil { return buscalls.WorkQueueDeadLetters{}, err } diff --git a/node-tools/internal/bus/ledger.go b/node-tools/internal/bus/ledger.go index c9080eb..fb38439 100644 --- a/node-tools/internal/bus/ledger.go +++ b/node-tools/internal/bus/ledger.go @@ -82,7 +82,7 @@ func publishExpecting(c *Conn, subject string, body []byte, expect uint64) (uint func lastOn(c *Conn, stream, subject string) ([]byte, uint64, error) { ctx, cancel := context.WithTimeout(context.Background(), StateTimeout) defer cancel() - js, err := c.jetStream() + js, err := c.JetStream() if err != nil { return nil, 0, err } @@ -149,7 +149,7 @@ type busState struct { } func (s *busState) kv(ctx context.Context) (jetstream.KeyValue, error) { - js, err := s.c.jetStream() + js, err := s.c.JetStream() if err != nil { return nil, err } diff --git a/node-tools/internal/bus/sent.go b/node-tools/internal/bus/sent.go index 271252e..4eecf27 100644 --- a/node-tools/internal/bus/sent.go +++ b/node-tools/internal/bus/sent.go @@ -25,7 +25,7 @@ func (c *Conn) SentGet(r buscalls.ObjectsSentGetRequest) (*buscalls.ObjectData, return nil, fmt.Errorf("%s is no sent store (a module's is `%s%s`): only a sent object is read by its reference", r.Store, catalogue.BusNameSeparator, catalogue.SentObjectsLocal) } - js, err := c.jetStream() + js, err := c.JetStream() if err != nil { return nil, err } diff --git a/node-tools/internal/firstview/firstview.go b/node-tools/internal/firstview/firstview.go index 4b93e15..0ba4efb 100644 --- a/node-tools/internal/firstview/firstview.go +++ b/node-tools/internal/firstview/firstview.go @@ -4,7 +4,8 @@ // **It never reads the owners' state.** At start it asks each owner's first-view tool until answered (the owner answers in a // runtime of its own, never through this one: ADR 0621 §1), then follows the owner's views family, one event per changed key; // a gap in the revisions, or an owner whose state was made again, asks the first view once more, said in the log. Past two -// minutes without an answer it writes a finding in its own state `findings`, which the condition keeper raises and clears. +// minutes without an answer it writes a finding through its held finding writer (design 47 G A-27, A-28), which keeps it in its +// own state `findings` and says it; the answer clears it. package firstview import ( @@ -12,12 +13,14 @@ import ( "encoding/json" "fmt" "sort" + "strings" "sync" "time" - conditions "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshconditionsclient" condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" "git.novox.be/novox/mesh-sdk/go/views" "github.com/novox/mesh-tools/node-tools/internal/bus" @@ -74,36 +77,85 @@ type Deps struct { Ask func(address string, body any) (json.RawMessage, error) // Consume hands each event of the owners' views families to deliver, until the stop is called. Consume func(deliver func(bus.Envelope) error) (stop func(), err error) - // PutFinding and DeleteFinding keep and remove a finding in the runner's own state. - PutFinding func(key string, value json.RawMessage) error - DeleteFinding func(key string) error + // Findings is the runner's held finding writer (views.HeldWriter): a follower's unanswered first view is written and cleared + // through it with the contract's Finding (design 47 G A-28). Nil says it in the log only. + Findings FindingWriter + // StartFindings starts that writer, and its stop ends it; nil starts none. + StartFindings func(ctx context.Context) (stop func(), err error) + // Node is the machine the runner is on, named in its findings' subject. + Node string // Bound and Wait are the follower's (zero: two minutes, five seconds). Bound, Wait time.Duration } +// FindingWriter is the writer the runner's findings go through: views.HeldWriter over the contract's Finding. +type FindingWriter interface { + Write(ctx context.Context, finding condseat.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error +} + // 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) + return StartWith(parent, DepsOf(conn, logf), 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 { +// DepsOf is the views' reach through the runner's own connection: its tool calls, its consumer and its held finding writer, whose +// state adapter, JetStream and node are this connection's, in this one place (design 47 G A-16, A-27). +func DepsOf(conn *bus.Conn, logf func(string, ...any)) Deps { me := conn.Module() - return Deps{Module: me, + d := Deps{Module: me, Node: conn.Node(), Ask: func(address string, body any) (json.RawMessage, error) { got, err := conn.Ask(address, body, "") return got.Result, err }, Consume: func(deliver func(bus.Envelope) error) (func(), error) { return conn.ConsumeAs(me, deliver) }, - PutFinding: func(key string, value json.RawMessage) error { - _, err := conn.StatePut(me, FindingsState, key, value) - return err - }, - DeleteFinding: func(key string) error { return conn.StateDelete(me, FindingsState, key) }, } + js, err := conn.JetStream() + if err != nil { + logf("[mesh-tools] the runner's findings are not written (%v): the first views' findings are in the log only", err) + return d + } + w, err := views.NewHeldWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{ + Asked: condseat.SubjectFindingsAsked, Module: me, Node: conn.Node(), State: findingsState{conn: conn, module: me}, JS: js, Logf: logf, + Kinds: []string{views.UnansweredKind}}) + if err != nil { + logf("[mesh-tools] the runner's findings are not written (%v): the first views' findings are in the log only", err) + return d + } + d.Findings = w + d.StartFindings = func(ctx context.Context) (func(), error) { return w.Close, w.Start(ctx) } + return d } +// findingsState is the runner's own state `findings` (views.FindingState) over its connection. +type findingsState struct { + conn *bus.Conn + module string +} + +func (s findingsState) Get(key string) (*stdio.StateEntry, error) { + e, err := s.conn.StateGet(s.module, FindingsState, key) + if err != nil || e == nil { + return nil, err + } + return &stdio.StateEntry{Key: e.Key, Value: e.Value, Revision: e.Revision, At: e.At}, nil +} + +func (s findingsState) Put(key string, value any) (uint64, error) { + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + return s.conn.StatePut(s.module, FindingsState, key, raw) +} + +func (s findingsState) Delete(key string) error { + return s.conn.StateDelete(s.module, FindingsState, key) +} + +func (s findingsState) Keys() ([]string, error) { return s.conn.StateKeys(s.module, FindingsState) } + // StartWith is Start over deps. func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*Views, func()) { ctx, cancel := context.WithCancel(parent) @@ -120,12 +172,20 @@ func StartWith(parent context.Context, d Deps, logf func(string, ...any)) (*View v.wg.Add(1) go func() { defer v.wg.Done(); _ = f.Start(ctx) }() } + stopFindings := func() {} + if d.StartFindings != nil { + if stop, err := d.StartFindings(ctx); err != nil { + logf("[mesh-tools] the runner's finding writer does not start (%v): the first views' findings are in the log only", err) + } else { + stopFindings = stop + } + } stop, err := d.Consume(v.hear) if err != nil { logf("[mesh-tools] the owners' views are not followed (%v): the pictures are the first views alone until the runner starts again", err) stop = func() {} } - return v, func() { stop(); cancel(); v.wg.Wait() } + return v, func() { stop(); cancel(); v.wg.Wait(); stopFindings() } } // hear applies one views event; a gap asks the first view once more. @@ -145,17 +205,18 @@ func (v *Views) hear(env bus.Envelope) error { return nil } -// finding writes or removes the runner's finding of a first view unanswered past its bound, in its own state. +// finding writes or clears the runner's finding of a first view unanswered past its bound, through its writer (design 47 G A-28). func (v *Views) finding(st views.Status, raised bool) { - key, value, err := conditionOf(views.UnansweredFinding(v.deps.Module, st, time.Now())) - if err != nil { - v.logf("[mesh-tools] the finding of the first view %s is not made: %v", st.Tool, err) + f := unansweredFinding(v.deps.Module, v.deps.Node, st, time.Now()) + if v.deps.Findings == nil { + v.logf("[mesh-tools] %s", f.Summary) return } + var err error if raised { - err = v.deps.PutFinding(key, value) + err = v.deps.Findings.Write(context.Background(), f) } else { - err = v.deps.DeleteFinding(key) + err = v.deps.Findings.Clear(context.Background(), f.Scope, f.ID, f.Kind, "the owner answered") } if err != nil { v.logf("[mesh-tools] the finding of the first view %s is not kept: %v", st.Tool, err) @@ -211,19 +272,15 @@ func (v *Views) Again(ctx context.Context, why string) []error { return errs } -// conditionOf is a finding in the layout of the seat mesh-conditions' state findings (its contract, ADR 0634 §1), about a -// module, a warning. -func conditionOf(f views.Finding) (string, []byte, error) { - key := conditions.Key(scopeModule, f.ID, f.Kind) - v := condseat.FindingsByKeyValue{V: 1, Key: key, Kind: f.Kind, Observer: f.Observer, Severity: conditions.Warning, - Subject: condseat.Subject{Scope: scopeModule, ID: f.ID}, Summary: f.Summary, Headline: f.Headline, - Explanation: f.Explanation, Source: f.Observer + "." + f.Kind, At: f.At.UTC(), - Evidence: []condseat.Evidence{{At: f.At.UTC(), Said: f.Summary}}} - if err := conditions.CheckFinding(v); err != nil { - return "", nil, err - } - raw, err := json.Marshal(v) - return key, raw, err +// unansweredFinding is the finding of a first view unanswered past its bound, in the contract's form (design 47 G A-28): about the +// runner's module, a warning, its id the view's name escaped as a key's name is. +func unansweredFinding(module, node string, st views.Status, at time.Time) condseat.Finding { + return condseat.Finding{Kind: views.UnansweredKind, Scope: scopeModule, ID: keys.EscapeName(module + "." + st.Owner + "." + st.State), + Severity: condseat.FindingSeverityWarning, Source: "views", Observer: module, + Subject: condseat.Subject{Scope: scopeModule, ID: module, Machine: node}, + Summary: fmt.Sprintf("%s has no first view of %s's %s: %s has not answered since %s (%s)", module, st.Owner, st.State, st.Tool, + st.Asked.Format(time.RFC3339), strings.TrimSpace(st.Error)), + Headline: fmt.Sprintf("%s waits for %s", module, st.Owner), At: at.UTC()} } // scopeModule is the scope of a finding about a module. diff --git a/node-tools/internal/firstview/firstview_integration_test.go b/node-tools/internal/firstview/firstview_integration_test.go index baa369d..08db388 100644 --- a/node-tools/internal/firstview/firstview_integration_test.go +++ b/node-tools/internal/firstview/firstview_integration_test.go @@ -11,6 +11,7 @@ import ( "testing" "time" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/views" @@ -84,10 +85,10 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen var mu sync.Mutex findings := map[string]bool{} - deps := DepsOf(conn) + deps := DepsOf(conn, t.Logf) 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 } + deps.Findings = writes{&mu, findings} + deps.StartFindings = nil v, stop := StartWith(context.Background(), deps, func(string, ...any) {}) t.Cleanup(stop) @@ -156,3 +157,23 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen type errFindings int func (e errFindings) Error() string { return "not yet" } + +// writes is the runner's finding writer in this test: the findings written and not cleared. +type writes struct { + mu *sync.Mutex + by map[string]bool +} + +func (w writes) Write(_ context.Context, f condseat.Finding) error { + w.mu.Lock() + defer w.mu.Unlock() + w.by[views.FindingKey(f.Scope, f.ID, f.Kind)] = true + return nil +} + +func (w writes) Clear(_ context.Context, scope, id, kind, _ string) error { + w.mu.Lock() + defer w.mu.Unlock() + delete(w.by, views.FindingKey(scope, id, kind)) + return nil +} diff --git a/node-tools/internal/firstview/firstview_test.go b/node-tools/internal/firstview/firstview_test.go index 11c299d..e3f5375 100644 --- a/node-tools/internal/firstview/firstview_test.go +++ b/node-tools/internal/firstview/firstview_test.go @@ -11,6 +11,7 @@ import ( "testing" "time" + condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/views" "github.com/novox/mesh-tools/node-tools/internal/bus" @@ -33,6 +34,21 @@ func newOwners() *owners { rev: map[string]uint64{"assignments": 4, "seat-holders": 2, "manifests": 7, "modules": 3}, asked: map[string]int{}, finds: map[string]bool{}} } +// Write and Clear are the runner's finding writer in a test: the findings written and not cleared, by their key. +func (o *owners) Write(_ context.Context, f condseat.Finding) error { + o.mu.Lock() + defer o.mu.Unlock() + o.finds[views.FindingKey(f.Scope, f.ID, f.Kind)] = true + return nil +} + +func (o *owners) Clear(_ context.Context, scope, id, kind, _ string) error { + o.mu.Lock() + defer o.mu.Unlock() + delete(o.finds, views.FindingKey(scope, id, kind)) + return nil +} + func (o *owners) ask(address string, _ any) (json.RawMessage, error) { o.mu.Lock() defer o.mu.Unlock() @@ -87,19 +103,8 @@ func TestTheRunnerHoldsTheFourStatesFromTheOwnersFirstViewsAndTheirEvents(t *tes var logs []string var lmu sync.Mutex deps := Deps{Module: "node-tools", Ask: o.ask, Bound: 30 * time.Millisecond, Wait: 10 * time.Millisecond, - Consume: func(deliver func(bus.Envelope) error) (func(), error) { o.deliver = deliver; return func() {}, nil }, - PutFinding: func(key string, _ json.RawMessage) error { - o.mu.Lock() - o.finds[key] = true - o.mu.Unlock() - return nil - }, - DeleteFinding: func(key string) error { - o.mu.Lock() - delete(o.finds, key) - o.mu.Unlock() - return nil - }} + Consume: func(deliver func(bus.Envelope) error) (func(), error) { o.deliver = deliver; return func() {}, nil }, + Findings: o} v, stop := StartWith(context.Background(), deps, func(f string, a ...any) { lmu.Lock(); logs = append(logs, f); lmu.Unlock() }) defer stop()