// Package firstview is the tool runner's picture of the four states it reads of the indexer and the assigner: the assigner's // assignments and seat-holders, the indexer's manifests and modules (novox/hq ADR 0615 §1, §3; ADR 0621 §2; issue 1966 TLS-5). // // **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 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 ( "context" "encoding/json" "fmt" "sort" "strings" "sync" "time" 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" ) // The owners and their first-view tools (the tools' addresses are `.`). const ( Indexer = "mesh-delivery-indexer" Assigner = "mesh-delivery-assigner" ) // The names the console reads the four states by (`.`), as the manifest's `consumes` names their views // families (no seat's client names a holder, novox/hq ADR 0625 §4; TestTheFirstViewsAreTheManifestsViews). const ( AssignmentsRead = Assigner + ".assignments" SeatHoldersRead = Assigner + ".seat-holders" ModulesRead = Indexer + ".modules" ManifestsRead = Indexer + ".manifests" ) // Spec is one first view the runner holds. type Spec struct{ Owner, Tool, State string } // Specs are the four. var Specs = []Spec{ {Assigner, "assignments", "assignments"}, {Assigner, "seat-holders", "seat-holders"}, {Indexer, "manifests", "manifests"}, {Indexer, "modules", "modules"}, } // Name is the name a state is read by, `.`: AssignmentsRead, SeatHoldersRead and the indexer's. func (s Spec) Name() string { return s.Owner + "." + s.State } // Views is the four followers. type Views struct { logf func(string, ...any) followers map[string]*views.Follower // by Spec.Name() byState map[string]*views.Follower // by the state an event names ctx context.Context cancel context.CancelFunc wg sync.WaitGroup deps Deps } // FindingsState is the runner's own state the findings are written in. const FindingsState = "findings" // Deps is what the views reach the bus through: Start makes them from the runner's connection, a test from fakes. type Deps struct { // Module is the runner's own name, the observer of its findings. Module string // Ask calls an owner's tool, `.`. 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) // 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), logf) } // 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() 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) }, } 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) v := &Views{logf: logf, ctx: ctx, cancel: cancel, followers: map[string]*views.Follower{}, byState: map[string]*views.Follower{}, deps: d} for _, s := range Specs { s := s call := func(tool string, body any) (json.RawMessage, error) { return d.Ask(s.Owner+"."+tool, body) } f := &views.Follower{Owner: s.Owner, Tool: s.Tool, State: s.State, Ask: views.AskVia(call, s.Tool), Logf: logf, Bound: d.Bound, Wait: d.Wait, Raise: func(st views.Status) { v.finding(st, true) }, Clear: func(st views.Status) { v.finding(st, false) }} v.followers[s.Name()] = f v.byState[s.State] = f 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(); stopFindings() } } // hear applies one views event; a gap asks the first view once more. func (v *Views) hear(env bus.Envelope) error { ev, err := views.DecodeEvent(env.Body) if err != nil { return nil // not a views event: nothing to hold, nothing to retry } f := v.byState[ev.State] if f == nil || f.Owner != ev.Owner { return nil } if gap, why := f.Apply(ev); gap { v.logf("[mesh-tools] a gap in %s's %s (%s): its first view %s is asked once more", f.Owner, f.State, why, f.Tool) go func() { _ = f.Again(v.ctx, "a gap: "+why) }() } return nil } // 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) { 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.Findings.Write(context.Background(), f) } else { 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) } } // State is the picture of the state spec names (`.`), every current value by key; an error says the owner has // not answered yet, so nothing is claimed of it. func (v *Views) State(name string) (map[string][]byte, error) { f := v.followers[name] if f == nil { return nil, fmt.Errorf("the tool runner holds no first view of %s", name) } select { case <-f.Ready(): default: return nil, fmt.Errorf("%s's first view %s has not been answered yet", f.Owner, f.Tool) } out := map[string][]byte{} for k, raw := range f.Values() { out[k] = raw } return out, nil } // Entry is one value of the picture, nil when the owner holds none. func (v *Views) Entry(name, key string) ([]byte, error) { all, err := v.State(name) if err != nil { return nil, err } return all[key], nil } // Status is each first view as it stands, by `.`, sorted. func (v *Views) Status() []views.Status { var out []views.Status for _, f := range v.followers { out = append(out, f.Status()) } sort.Slice(out, func(i, j int) bool { return out[i].Owner+out[i].State < out[j].Owner+out[j].State }) return out } // Again asks every first view once more, by hand, for why. func (v *Views) Again(ctx context.Context, why string) []error { var errs []error for _, f := range v.followers { if err := f.Again(ctx, why); err != nil { errs = append(errs, err) } } return errs } // 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. const scopeModule = "module"