diff --git a/modules/distribution/cmd/distribution-tools/main.go b/modules/distribution/cmd/distribution-tools/main.go index 86b562851..166ab10c3 100644 --- a/modules/distribution/cmd/distribution-tools/main.go +++ b/modules/distribution/cmd/distribution-tools/main.go @@ -125,6 +125,51 @@ func bootstrapCloser(windows kept[holder.Window]) func() { }) } +// learned are the proved callers whose build this process recorded as publishing through the seats. +var learned sync.Map + +// learnPublishing records the fingerprint the proved caller's node runs of its build module as one that publishes +// through the seats (holder.PublishingKey), once per caller and process: the windows' driver in the program counts a +// node's have running it as publishing, closing the direct-write window when every node's build modules do. +func learnPublishing(windows kept[holder.Window], caller string) { + node, module, ok := strings.Cut(caller, ".") + if !ok || strings.HasPrefix(caller, "person.") { + return + } + if _, done := learned.Load(caller); done { + return + } + raw, err := stdio.Ask("mesh-delivery-view.node_haves", map[string]any{"node": node, "module": module}) + if err != nil { + logf("the build of %s that published cannot be read (%v): the direct-write window counts it at its next publish", caller, err) + return + } + var a struct { + Keys map[string]struct { + FP string `json:"fp"` + } `json:"keys"` + } + if json.Unmarshal(raw, &a) != nil || a.Keys[caller].FP == "" { + logf("node_haves names no build of %s: the direct-write window counts it at its next publish", caller) + return + } + pub := kept[holder.Publishing]{state: "windows", key: func(p holder.Publishing) string { return holder.PublishingKey }} + cur, _, err := pub.Get(holder.PublishingKey) + if err != nil { + logf("the publishing builds cannot be read: %v", err) + return + } + cur.Name = holder.PublishingKey + if cur.Learn(module, a.Keys[caller].FP) { + if err := pub.Put(cur); err != nil { + logf("the publishing build of %s is not kept: %v", caller, err) + return + } + logf("%s's build %s publishes through the seats", module, a.Keys[caller].FP) + } + learned.Store(caller, true) +} + func arg[T any](args map[string]any) (T, error) { var v T clean := map[string]any{} @@ -159,6 +204,9 @@ func tools(v *holder.Verbs, windows kept[holder.Window]) []stdio.Tool { if err == nil && closeBootstrap != nil { closeBootstrap() } + if err == nil { + learnPublishing(windows, stdio.CallerOf(args)) + } return a, err }}, stdio.Tool{Name: seat + "." + holder.VerbPublishPart, Description: "Send one chunk (at most 1 MiB) of one part of an open transfer, from the caller that opened it.", diff --git a/modules/distribution/cmd/distribution/main.go b/modules/distribution/cmd/distribution/main.go index 9c61d8061..cf856c6e2 100644 --- a/modules/distribution/cmd/distribution/main.go +++ b/modules/distribution/cmd/distribution/main.go @@ -1,8 +1,8 @@ // Command distribution is distribution's program (novox/hq design 51): it serves the artifact store's port in front of // the registry container, which listens on its node's loopback only. Reads (GET, HEAD) go through to every node's // container runtime; a write is refused but while the direct-write window is open, never over a tag the seat holds -// published, or with the bootstrap's one-off credential (holder.Front). The window is distribution's state windows, -// written by its tools bundle; the program reads it on each write and holds no timer. +// published, or with the bootstrap's one-off credential (holder.Front). The windows are distribution's state windows: +// the program moves them (windows.go) and follows the direct-write window's record on each change; it holds no timer. package main import ( @@ -59,6 +59,18 @@ func run(ctx context.Context, getenv func(string) string) error { if windows != nil { go follow(ctx, windows, &open, logger) go followBootstrap(ctx, windows, logger) + calls, err := process.Connect(cred) + if err != nil { + return err + } + defer calls.Close() + callers, err := holder.ParseCallers(getenv("MESH_CONFIG_CALLERS")) + if err != nil { + return err + } + findings := process.OptionalBucket(ctx, js, "distribution_findings", "a window held for seven days is said in the log only", logger.Printf) + go drive(ctx, js, cred, &holder.Driver{Delivery: busDelivery{calls}, State: kvWindows{windows}, Callers: callers, + Closed: sayClosed(js, cred), Finding: findingsKV(findings, logger.Printf)}, logger) } front := &holder.Front{Store: store, Window: &open, Logf: logger.Printf, Published: func(name, tag string) (bool, error) { diff --git a/modules/distribution/cmd/distribution/windows.go b/modules/distribution/cmd/distribution/windows.go new file mode 100644 index 000000000..1bf2fd989 --- /dev/null +++ b/modules/distribution/cmd/distribution/windows.go @@ -0,0 +1,239 @@ +package main + +// The windows' driver (design 51 *The move*): at the program's start, on each `delivered` of the status view and on each +// node-engine's `manifest-reads-changed` (ADR 0618 §4) it reads every node through node_haves and the node-engine +// seat's manifest-reads, and steps the old-name and the direct-write windows in distribution's state windows. The +// program does it, not the tools bundle: it holds distribution's own account, its durable consumer of the events +// stream and its own stream for the closing events. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log" + "strings" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/health" + "git.novox.be/novox/mesh-sdk/go/process" + + "distribution/internal/holder" +) + +// The addresses the driver asks (manifest invokes). +const ( + havesAsk = "mesh-delivery-view.node_haves" + readsAsk = "seat:node-engine.manifest-reads" +) + +// caller is a process's call of a tool or a seat verb. +type caller interface { + Call(key string, body any) (json.RawMessage, error) + CallOn(node, key string, body any) (json.RawMessage, error) +} + +// busDelivery answers holder.Delivery through the bus. +type busDelivery struct{ c caller } + +func (d busDelivery) Haves() ([]holder.NodeHave, error) { + raw, err := d.c.Call(havesAsk, map[string]any{}) + if err != nil { + return nil, err + } + var a struct { + Keys map[string]struct { + FP string `json:"fp"` + } `json:"keys"` + Cut bool `json:"cut"` + } + if err := json.Unmarshal(unwrap(raw), &a); err != nil { + return nil, fmt.Errorf("node_haves answered what is not its listing: %w", err) + } + if a.Cut { + return nil, errors.New("node_haves cut its answer, so not every node is read") + } + out := make([]holder.NodeHave, 0, len(a.Keys)) + for key, h := range a.Keys { + node, module, ok := strings.Cut(key, ".") + if ok { + out = append(out, holder.NodeHave{Node: node, Module: module, FP: h.FP}) + } + } + return out, nil +} + +func (d busDelivery) Reads(node string) (bool, []string, error) { + raw, err := d.c.CallOn(node, readsAsk, map[string]any{}) + if err != nil { + if notServed(err) { + return false, nil, nil // a node-engine not serving manifest-reads states nothing (design 51) + } + return false, nil, err + } + var a struct { + Reads []string `json:"manifest-reads"` + } + if err := json.Unmarshal(unwrap(raw), &a); err != nil { + return true, nil, fmt.Errorf("%s's manifest-reads answered what is not its statement: %w", node, err) + } + return true, a.Reads, nil +} + +// notServed is whether a call's error says nothing serves the address (an away node, an older node-engine). +func notServed(err error) bool { + if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrTimeout) || errors.Is(err, context.DeadlineExceeded) { + return true + } + m := strings.ToLower(err.Error()) + for _, w := range []string{"no responders", "serves no verb", "not served", "unknown tool", "timeout"} { + if strings.Contains(m, w) { + return true + } + } + return false +} + +// unwrap takes a tool's answer out of the MCP content form the tool runner may answer in. +func unwrap(raw json.RawMessage) json.RawMessage { + var s string + if json.Unmarshal(raw, &s) == nil { + return unwrap(json.RawMessage(s)) + } + var c struct { + Content []struct { + Text string `json:"text"` + } `json:"content"` + } + if json.Unmarshal(raw, &c) == nil && len(c.Content) > 0 { + return json.RawMessage(c.Content[0].Text) + } + return raw +} + +// kvWindows answers holder.WindowState over the state windows. +type kvWindows struct{ kv jetstream.KeyValue } + +func (k kvWindows) get(key string, v any) error { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + e, err := k.kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil + } + if err != nil { + return err + } + return json.Unmarshal(e.Value(), v) +} + +func (k kvWindows) Window(name string) (holder.Window, error) { + var w holder.Window + return w, k.get(name, &w) +} + +func (k kvWindows) PutWindow(w holder.Window) error { + b, _ := json.Marshal(w) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err := k.kv.Put(ctx, w.Name, b) + return err +} + +func (k kvWindows) Publishing() (holder.Publishing, error) { + var p holder.Publishing + return p, k.get(holder.PublishingKey, &p) +} + +// findingsKV writes the driver's findings into distribution's state findings, as the tools bundle writes its own. +func findingsKV(kv jetstream.KeyValue, logf func(string, ...any)) func(kind, subject, words string, raised bool) { + return func(kind, subject, words string, raised bool) { + if kv == nil { + return + } + key := health.FindingKey(health.ScopeSeat, holder.Self, kind+"."+strings.NewReplacer(".", "_", "@", "_", "/", "_").Replace(subject)) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if !raised { + _ = kv.Delete(ctx, key) + return + } + f := health.Finding{V: health.FindingVersion, Key: key, Kind: kind, Ref: key[strings.LastIndex(key, ".")+1:], + Subject: health.Subject{Scope: health.ScopeSeat, ID: holder.Self}, Severity: health.Warning, + Summary: words, Headline: words, Explanation: words + "; registry_old_name or registry_direct_writes drops a retired node by hand", + Resolved: "cleared when the window closes", Source: kind, Observer: holder.Self, At: time.Now().UTC()} + b, _ := json.Marshal(f) + if _, err := kv.Put(ctx, key, b); err != nil { + logf("the finding %s could not be written: %v", key, err) + } + } +} + +// sayClosed says a window's closing on distribution's own family distribution.events, once (its id is the window's). +func sayClosed(js jetstream.JetStream, cred process.Account) func(holder.Closed) error { + return func(c holder.Closed) error { + event := holder.OldNameClosedEvent + if c.Window == holder.DirectWritesWindow { + event = holder.DirectWritesClosedEvent + } + b, _ := json.Marshal(c) + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err := js.Publish(ctx, cred.StreamSubject(event), b, jetstream.WithMsgID(holder.Self+"."+event)) + return err + } +} + +// wakes says which events wake the driver, by subject: the status view's delivered, a node-engine's +// manifest-reads-changed (its node the subject's last token). +func wakes(subject string) (string, bool) { + switch { + case strings.HasSuffix(subject, ".mesh-delivery-view.event.delivered"): + return "delivered", true + case strings.Contains(subject, ".node-engine.event.manifest-reads-changed."): + return "manifest-reads-changed on " + subject[strings.LastIndex(subject, ".")+1:], true + } + return "", false +} + +// drive wakes the driver at start and on each waking event of the program's durable consumer of the events stream. +// A wake that could not judge is said and its event asked again a minute later. +func drive(ctx context.Context, js jetstream.JetStream, cred process.Account, d *holder.Driver, logger *log.Logger) { + wake := func(why string) error { + said, err := d.Wake(why) + for _, s := range said { + logger.Printf("%s", s) + } + if err != nil { + logger.Printf("the windows were not judged (%s): %v", why, err) + } + return err + } + _ = wake("distribution started") + cons, err := cred.EventConsumer(ctx, js, process.EventsStream) + if err != nil { + logger.Printf("the events stream cannot be heard (%v): the windows move only at the next start", err) + return + } + cc, err := process.HearEvents(cons, func(m jetstream.Msg) { + why, ok := wakes(m.Subject()) + if !ok { + _ = m.Ack() + return + } + if wake(why) != nil { + _ = m.NakWithDelay(time.Minute) + return + } + _ = m.Ack() + }, logger.Printf) + if err != nil { + logger.Printf("the events stream cannot be heard (%v): the windows move only at the next start", err) + return + } + <-ctx.Done() + cc.Stop() +} diff --git a/modules/distribution/internal/holder/driver.go b/modules/distribution/internal/holder/driver.go new file mode 100644 index 000000000..9c29d0455 --- /dev/null +++ b/modules/distribution/internal/holder/driver.go @@ -0,0 +1,190 @@ +package holder + +import ( + "errors" + "fmt" + "slices" + "sort" + "strings" + "time" +) + +// PublishingKey is the key of distribution's state windows that records, per build module, the fingerprints of the +// builds heard publishing through the seats: the tools bundle adds the have's fingerprint of the proved caller of each +// accepted publish. A node's have of a build module running one of them counts as publishing through the seats, so a +// node whose build module runs the same build but builds nothing (a paused build agent) does not hold the window. +const PublishingKey = "publishing" + +// Publishing is that record: module -> fingerprints. +type Publishing struct { + Name string `json:"name"` + Modules map[string][]string `json:"modules"` +} + +// Learn adds fp for module; it answers whether it was new. +func (p *Publishing) Learn(module, fp string) bool { + if module == "" || fp == "" { + return false + } + if p.Modules == nil { + p.Modules = map[string][]string{} + } + if slices.Contains(p.Modules[module], fp) { + return false + } + p.Modules[module] = append(p.Modules[module], fp) + sort.Strings(p.Modules[module]) + return true +} + +// NodeHave is one key of the status view's node_haves: a node, a module, the fingerprint it runs. +type NodeHave struct { + Node, Module, FP string +} + +// Delivery is what the windows' driver asks of the delivery parts and the node-engine seat, through their tools and +// verbs on the bus (ADR 0596): every node's haves, away ones included, and one node-engine's stated reader +// capabilities. Reads answers served false when the node-engine does not serve `manifest-reads` (it states nothing). +type Delivery interface { + Haves() ([]NodeHave, error) + Reads(node string) (served bool, stated []string, err error) +} + +// WindowState is distribution's state windows. +type WindowState interface { + Window(name string) (Window, error) + PutWindow(w Window) error + Publishing() (Publishing, error) +} + +// Driver moves the old-name and the direct-write windows: at distribution's start, on each `delivered` and on each +// node-engine's `manifest-reads-changed` (ADR 0618 §4), never on a timer. Each wake reads every node, since a window +// closes only when no node holds it. +type Driver struct { + Delivery Delivery + State WindowState + Callers Callers + // Closed says a window's closing on distribution.events. + Closed func(c Closed) error + Finding func(kind, subject, words string, raised bool) + Now func() time.Time +} + +// FindingWindowHeld is the finding of a node holding a window open for seven days. +const FindingWindowHeld = "window-held" + +// Read reads every node: its haves, its node-engine's stated capabilities, and whether each build module's have runs a +// build heard publishing through the seats. +func (d *Driver) Read() ([]NodeRead, error) { + haves, err := d.Delivery.Haves() + if err != nil { + return nil, fmt.Errorf("node_haves cannot be read, so no window moves: %w", err) + } + pub, err := d.State.Publishing() + if err != nil { + return nil, fmt.Errorf("the publishing builds cannot be read, so no window moves: %w", err) + } + builds := d.Callers[VerbPublish] + byNode := map[string]*NodeRead{} + var order []string + for _, h := range haves { + if h.Module == "node" || h.Node == "" { + continue + } + n := byNode[h.Node] + if n == nil { + n = &NodeRead{Node: h.Node, NamesArtifact: true} + byNode[h.Node] = n + order = append(order, h.Node) + } + if slices.Contains(builds, h.Module) { + if n.BuildModules == nil { + n.BuildModules = map[string]bool{} + } + n.BuildModules[h.Module] = slices.Contains(pub.Modules[h.Module], h.FP) + } + } + sort.Strings(order) + out := make([]NodeRead, 0, len(order)) + for _, node := range order { + n := byNode[node] + served, stated, err := d.Delivery.Reads(node) + if err != nil && served { + return nil, fmt.Errorf("%s's manifest-reads cannot be read, so no window moves: %w", node, err) + } + n.Serves, n.Stated = served, stated + out = append(out, *n) + } + return out, nil +} + +// Wake reads every node and steps both windows; it answers a line for each thing it did, and an error when nothing +// could be judged. +func (d *Driver) Wake(why string) ([]string, error) { + nodes, err := d.Read() + if err != nil { + return nil, err + } + now := d.now() + var said, errs []string + for _, c := range []struct { + name, event, seat string + holding func([]NodeRead, []DroppedNode) []string + }{ + {OldNameWindow, OldNameClosedEvent, OldSeat, HoldingOldName}, + {DirectWritesWindow, DirectWritesClosedEvent, "", HoldingDirectWrites}, + } { + cur, err := d.State.Window(c.name) + if err != nil { + errs = append(errs, err.Error()) + continue + } + if cur.Name == "" { + cur = Window{Name: c.name, Open: true} + } + next, closed, overdue := cur.Step(c.holding(nodes, cur.Dropped), now) + if !windowsEqual(cur, next) { + if err := d.State.PutWindow(next); err != nil { + errs = append(errs, fmt.Sprintf("the %s window is not kept: %v", c.name, err)) + continue + } + } + for _, n := range cur.Holding { + if !slices.Contains(overdue, n) && d.Finding != nil { + d.Finding(FindingWindowHeld, c.name+"."+n, "", false) + } + } + for _, n := range overdue { + if d.Finding != nil { + d.Finding(FindingWindowHeld, c.name+"."+n, fmt.Sprintf("%s has held distribution's %s window open since %s", n, c.name, next.Since[n].Format(time.RFC3339)), true) + } + } + if closed != nil { + closed.Seat = c.seat + if d.Closed != nil { + if err := d.Closed(*closed); err != nil { + errs = append(errs, fmt.Sprintf("%s was not said: %v", c.event, err)) + } + } + said = append(said, fmt.Sprintf("the %s window is closed (%s); it was held last by %v", c.name, why, closed.LastHolding)) + } else if next.Open { + said = append(said, fmt.Sprintf("the %s window stays open (%s): held by %v", c.name, why, next.Holding)) + } + } + if len(errs) > 0 { + return said, errors.New(strings.Join(errs, "; ")) + } + return said, nil +} + +func (d *Driver) now() time.Time { + if d.Now != nil { + return d.Now() + } + return time.Now().UTC() +} + +func windowsEqual(a, b Window) bool { + return a.Open == b.Open && slices.Equal(a.Holding, b.Holding) && a.Closed.Equal(b.Closed) && + len(a.Since) == len(b.Since) && len(a.Dropped) == len(b.Dropped) +} diff --git a/modules/distribution/internal/holder/holder_test.go b/modules/distribution/internal/holder/holder_test.go index 812a74d3c..a12da1e6b 100644 --- a/modules/distribution/internal/holder/holder_test.go +++ b/modules/distribution/internal/holder/holder_test.go @@ -441,6 +441,42 @@ func TestTheHolderOfTheTwoImageSeats(t *testing.T) { t.Error("a closed window closed again") } }) + t.Run("the windows' driver: every node read on each wake, closed once, a paused build agent's same build counted", func(t *testing.T) { + del := &fakeDelivery{haves: []NodeHave{ + {"ace", "mesh-build-go", "fp-new"}, {"novox", "mesh-build-go", "fp-new"}, {"g14", "mesh-build-go", "fp-old"}, + {"g14", "node", ""}, {"ace", "gitea", "x"}, {"novox", "gitea", "x"}, + }, stated: map[string][]string{"ace": {NewNamesCapability}, "novox": {NewNamesCapability}}} + st := &memWindows{w: map[string]Window{}} + st.pub.Learn("mesh-build-go", "fp-new") + var closed []string + found := map[string]bool{} + now := time.Unix(0, 0) + d := &Driver{Delivery: del, State: st, Callers: DefaultCallers(), Now: func() time.Time { return now }, + Closed: func(c Closed) error { closed = append(closed, c.Window+" "+c.Seat); return nil }, + Finding: func(kind, subject, _ string, raised bool) { found[kind+" "+subject] = raised }} + if _, err := d.Wake("start"); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(st.w[OldNameWindow].Holding, []string{"g14"}) || !reflect.DeepEqual(st.w[DirectWritesWindow].Holding, []string{"g14"}) { + t.Errorf("windows = %+v, want g14 holding both (away: states nothing; runs the old build)", st.w) + } + now = now.Add(WindowFindingAfter) + _, _ = d.Wake("delivered") + if !found[FindingWindowHeld+" "+OldNameWindow+".g14"] { + t.Errorf("findings %v, want g14 named after seven days", found) + } + del.haves[2].FP, del.stated["g14"] = "fp-new", []string{NewNamesCapability} + if _, err := d.Wake("manifest-reads-changed on g14"); err != nil { + t.Fatal(err) + } + _, _ = d.Wake("delivered") + if want := []string{OldNameWindow + " " + OldSeat, DirectWritesWindow + " "}; !reflect.DeepEqual(closed, want) { + t.Errorf("closed %v, want each window said once, %v", closed, want) + } + if found[FindingWindowHeld+" "+OldNameWindow+".g14"] { + t.Error("the finding stays after the window closed") + } + }) t.Run("the manifest", func(t *testing.T) { raw, err := os.ReadFile("../../module.json") if err != nil { diff --git a/modules/distribution/internal/holder/mirror_fakes_test.go b/modules/distribution/internal/holder/mirror_fakes_test.go index 5f0f5189e..c33b649da 100644 --- a/modules/distribution/internal/holder/mirror_fakes_test.go +++ b/modules/distribution/internal/holder/mirror_fakes_test.go @@ -85,3 +85,24 @@ func (f *fakeRegistry) ServeHTTP(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNotFound) } } + +// fakeDelivery answers the driver's reads: haves, and per node whether its node-engine serves manifest-reads. +type fakeDelivery struct { + haves []NodeHave + stated map[string][]string // a node absent does not serve manifest-reads +} + +func (f *fakeDelivery) Haves() ([]NodeHave, error) { return f.haves, nil } +func (f *fakeDelivery) Reads(node string) (bool, []string, error) { + s, ok := f.stated[node] + return ok, s, nil +} + +type memWindows struct { + w map[string]Window + pub Publishing +} + +func (m *memWindows) Window(n string) (Window, error) { return m.w[n], nil } +func (m *memWindows) PutWindow(w Window) error { m.w[w.Name] = w; return nil } +func (m *memWindows) Publishing() (Publishing, error) { return m.pub, nil } diff --git a/modules/distribution/module.json b/modules/distribution/module.json index 5f3fcf449..537ba8c67 100644 --- a/modules/distribution/module.json +++ b/modules/distribution/module.json @@ -54,7 +54,8 @@ "seat:node-engine.manifest-reads" ], "consumes": [ - "mesh-delivery-view.delivered" + "mesh-delivery-view.delivered", + "node-engine.manifest-reads-changed" ], "emits": [ "old-name-closed", @@ -192,7 +193,8 @@ "env": { "MESH_BROKER_FILE": "/var/lib/mesh/registry/broker", "MESH_REGISTRY_LOOPBACK": "http://127.0.0.1:5001", - "MESH_REGISTRY_LISTEN": ":5000" + "MESH_REGISTRY_LISTEN": ":5000", + "MESH_CONFIG_CALLERS": "${config:callers}" }, "user": "distribution", "health": {