From 7bd61332cb51b2fcae79d2f3a0a4a0d45ff906dd Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 11 Oct 2026 19:22:43 +0200 Subject: [PATCH] Let a module declare a log the controller creates and never removes (ADR 0297) The issue tracker's history lasts seven days on the events stream and its bucket fills toward a cap with no warning (issues 501, 502). A module may now declare `logs`; the controller creates each as a stream on the bus (create-or-update, never deleted or recreated, reported when undeclared), issues it to the owner's assignments in their membership, grants the runtime exactly the append and direct-read subjects, and raises a condition when any module's bucket or log reaches 75% of its cap, cleared below 70%. --- cmd/mesh-controller/check.go | 8 + cmd/mesh-controller/doctor.go | 5 + cmd/mesh-controller/plain_words.go | 7 + cmd/mesh-controller/push.go | 26 ++- cmd/mesh-controller/store_fill.go | 165 ++++++++++++++ cmd/mesh-controller/store_fill_test.go | 76 +++++++ internal/broker/jetstream.go | 38 ++++ internal/broker/log.go | 195 ++++++++++++++++ internal/broker/log_test.go | 294 +++++++++++++++++++++++++ internal/broker/membership.go | 5 + internal/broker/nats.go | 10 + internal/broker/users.go | 3 + internal/catalogue/logs.go | 105 +++++++++ internal/catalogue/logs_test.go | 68 ++++++ internal/catalogue/manifest.go | 9 +- internal/inventory/busrecords.go | 25 +++ 16 files changed, 1037 insertions(+), 2 deletions(-) create mode 100644 cmd/mesh-controller/store_fill.go create mode 100644 cmd/mesh-controller/store_fill_test.go create mode 100644 internal/broker/log.go create mode 100644 internal/broker/log_test.go create mode 100644 internal/catalogue/logs.go create mode 100644 internal/catalogue/logs_test.go diff --git a/cmd/mesh-controller/check.go b/cmd/mesh-controller/check.go index f3c5b3a4..4ccf7b5f 100644 --- a/cmd/mesh-controller/check.go +++ b/cmd/mesh-controller/check.go @@ -179,6 +179,14 @@ func moduleCheckFor(paths []string, longestMachine int, out io.Writer) error { } fmt.Fprintf(out, ", keeps state %s", strings.Join(kept, ", ")) } + // And the logs it keeps, with their caps (novox/hq ADR 0297). + if len(m.Logs) > 0 { + kept := make([]string, 0, len(m.Logs)) + for _, l := range m.Logs { + kept = append(kept, fmt.Sprintf("%s (%d MiB)", l.Name, l.Cap())) + } + fmt.Fprintf(out, ", keeps log %s", strings.Join(kept, ", ")) + } if len(m.Reads) > 0 { fmt.Fprintf(out, ", reads %s", strings.Join(m.Reads, ", ")) } diff --git a/cmd/mesh-controller/doctor.go b/cmd/mesh-controller/doctor.go index 4986d514..3604a647 100644 --- a/cmd/mesh-controller/doctor.go +++ b/cmd/mesh-controller/doctor.go @@ -136,6 +136,11 @@ var probeRegistry = []probe{ {ID: "D-root", Asserts: "no agent can become root without a person on a machine where the router or a channel " + "proving its sender runs: not by its own account, and not through a tool that runs its command as an account " + "that can", From: "ADR 0259 §8", Kind: kindRootNotFree, Phase: 2, run: probeAgentRoot}, + // A module's bucket or log filling toward its cap (novox/hq ADR 0297 §6): said at three quarters, cleared + // below seven tenths, so a store at its threshold is not raised and cleared on every run. + {ID: probeStoreFillID, Asserts: "every module's bucket and log holds less than three quarters of its cap; one " + + "said filling is cleared once it holds less than seven tenths", From: "ADR 0297, issue 501", + Kind: kindStoreFilling, Phase: 1, run: probeStoreFill}, {ID: "DW", Asserts: "the watchdogs of the signals table ran within three of their intervals", From: "ADR 0227 rule 6: the watchers are watched", Kind: "watchdogs-silent", Phase: 1, run: probeWatchdogs}, // The core's health definitions (novox/hq to-be 45 §8, ADR 0236): what a core component's new build is diff --git a/cmd/mesh-controller/plain_words.go b/cmd/mesh-controller/plain_words.go index 29e5da85..a41ec619 100644 --- a/cmd/mesh-controller/plain_words.go +++ b/cmd/mesh-controller/plain_words.go @@ -606,6 +606,13 @@ var plainWordings = map[string]func(conditions.Observation) words{ Explanation: "The bus refused messages from a part of the mesh, so what they carried did not happen.", Resolved: "Resolved: the bus takes the messages again"} }), + kindStoreFilling: worded(func(o conditions.Observation) words { + return words{Headline: "A module's store on the bus is filling up", + Needs: "decide whether to raise its cap or have the module keep less.", + Explanation: "A bucket or log a module keeps on the bus holds three quarters of its cap or more. When it " + + "is full, the bus refuses what the module writes there.", + Resolved: "The store has room again"} + }), "stream-wrong": worded(func(o conditions.Observation) words { return words{Headline: "Part of the bus's storage is wrong", Needs: "check the machine the bus runs on; the details say what is missing.", diff --git a/cmd/mesh-controller/push.go b/cmd/mesh-controller/push.go index 9f3c1434..89e1caa1 100644 --- a/cmd/mesh-controller/push.go +++ b/cmd/mesh-controller/push.go @@ -1300,6 +1300,15 @@ func issueMemberships(ctx context.Context, open *stores, server *link.Server, se if _, err := broker.RaiseBuckets(broker.OnConn(bus.Conn), buckets); err != nil { return fmt.Errorf("the modules' state could not be asserted on the bus: %w", err) } + // **And every declared log, for the same reason** (novox/hq ADR 0297 §2): a membership names its + // logs, and a module whose log does not exist fails its first append. + logs, err := open.inventory.DeclaredLogs(ctx) + if err != nil { + return fmt.Errorf("the modules' logs could not be read, so no log was asserted: %w", err) + } + if _, err := broker.RaiseLogs(broker.OnConn(bus.Conn), logs); err != nil { + return fmt.Errorf("the modules' logs could not be asserted on the bus: %w", err) + } // Every membership is tried, and the first failure named once. issued := 0 refused := map[string]error{} @@ -1483,13 +1492,28 @@ func raiseTheBus(ctx context.Context, inv *inventory.Inventory, address string) fmt.Printf("the bus holds state nothing declares any more, kept because it is data: %s — "+ "removing it is a person's act\n", strings.Join(undeclared, ", ")) } + // Every module's log (novox/hq ADR 0297), from the catalogue, as its buckets: one that nothing + // declares any more is said and kept — a log is a module's record, and no path of the mesh removes it. + logs, err := inv.DeclaredLogs(ctx) + if err != nil { + return err + } + unlogged, err := broker.RaiseLogs(js, logs) + if err != nil { + return err + } + if len(unlogged) > 0 { + fmt.Printf("the bus holds logs nothing declares any more, kept because they are data: %s — "+ + "removing one is a person's act\n", strings.Join(unlogged, ", ")) + } // And how every module hears what it consumes: asserted with the rest above, counted here. hearing, err := moduleConsumerCount(ctx, inv) if err != nil { return err } fmt.Printf("the bus at %s has its streams, %d machine(s) can hear a declaration, %d module(s) "+ - "can hear what they consume, and %d bucket(s) of state\n", broker.BareAddress(address), len(names), hearing, len(buckets)) + "can hear what they consume, %d bucket(s) of state and %d log(s)\n", broker.BareAddress(address), len(names), hearing, + len(buckets), len(logs)) return nil } diff --git a/cmd/mesh-controller/store_fill.go b/cmd/mesh-controller/store_fill.go new file mode 100644 index 00000000..39dcaa63 --- /dev/null +++ b/cmd/mesh-controller/store_fill.go @@ -0,0 +1,165 @@ +package main + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// **A filling bucket or log is said before it is full** (novox/hq ADR 0297 §6, issue 501). +// +// A module's bucket and its log each have a cap on the bus, and when either is full the bus refuses +// the next write: the module stops doing what it writes for, and nothing said so before. On +// 2026-10-11 the issue tracker's bucket held a sixth of its cap after two days, growing toward a stop +// nobody would have heard coming. So the self-check reads every module's bucket and log, and one at +// fillRaiseAt of its cap or more raises a condition naming the module, the store and how full it is. +// +// **Cleared by observation below fillClearBelow, not below fillRaiseAt**: a store hovering at its +// threshold would otherwise be raised and cleared on every run. Between the two, an open condition is +// kept and none is raised. + +// The fill thresholds, in percent of a store's cap. +const ( + fillRaiseAt = 75 + fillClearBelow = 70 +) + +// kindStoreFilling is the condition a filling bucket or log raises. +const kindStoreFilling = "store-filling" + +// probeStoreFillID is the probe's id in the registry. +const probeStoreFillID = "D-fill" + +// moduleStore is one module's bucket or log as the fill probe reads it. +type moduleStore struct { + Module string + // Kind is `bucket` or `log`; Name its local name; Stream the stream it is on the bus. + Kind, Name, Stream string +} + +// filled is how full one store is, read from the bus. +type filled struct { + Bytes, Max uint64 +} + +// percent is how full, in whole percent, rounded down. +func (f filled) percent() uint64 { + if f.Max == 0 { + return 0 + } + return f.Bytes * 100 / f.Max +} + +// fillKey is where a store's fill condition is kept. +func fillKey(s moduleStore) string { return conditions.Key(conditions.ScopeBus, s.Stream, "filling") } + +// fillObservation is what one store's fill says: a finding at fillRaiseAt or above, and, while its +// condition is open, at fillClearBelow or above too; nothing otherwise, which is what clears it. +func fillObservation(s moduleStore, f filled, open bool) (conditions.Observation, bool) { + if f.Max == 0 { + return conditions.Observation{}, false // a store with no cap cannot fill + } + // Compared in bytes, not rounded percent: 74.9% is not 75%. + atRaise := f.Bytes*100 >= f.Max*fillRaiseAt + aboveClear := f.Bytes*100 >= f.Max*fillClearBelow + if !atRaise && !(open && aboveClear) { + return conditions.Observation{}, false + } + pct := f.percent() + fill := fmt.Sprintf("%s of %s, %d%%", mibWords(f.Bytes), mibWords(f.Max), pct) + return conditions.Observation{ + Scope: conditions.ScopeBus, ID: s.Stream, Token: "filling", Kind: kindStoreFilling, + Severity: conditions.Warning, + Summary: fmt.Sprintf("%s's %s %s holds %s: at its cap the bus refuses the module's next write", + s.Module, s.Kind, s.Name, fill), + Headline: fillHeadline(s, pct), + Explanation: fmt.Sprintf("The %s %s of %s holds %s. When it is full, the bus refuses what %s writes "+ + "there, and %s stops doing what it writes for.", s.Kind, s.Name, s.Module, fill, s.Module, s.Module), + Needs: fmt.Sprintf("decide whether to raise the cap of this %s or have %s keep less.", s.Kind, s.Module), + Resolved: fillResolved(s), + Said: fmt.Sprintf("module %s, %s %s (stream %s): %d of %d bytes, %d%%", s.Module, s.Kind, s.Name, + s.Stream, f.Bytes, f.Max, pct), + }, true +} + +// fillHeadline names the store and how full it is, in at most a headline's length. +func fillHeadline(s moduleStore, pct uint64) string { + for _, h := range []string{ + fmt.Sprintf("%s's %s %s is %d%% full", s.Module, s.Kind, s.Name, pct), + fmt.Sprintf("A %s of %s is %d%% full", s.Kind, s.Module, pct), + } { + if len(h) <= conditions.HeadlineMax { + return h + } + } + return fmt.Sprintf("A module's %s is %d%% full", s.Kind, pct) +} + +// fillResolved is the line said when the store has room again, in at most a resolved line's length. +func fillResolved(s moduleStore) string { + if h := fmt.Sprintf("%s's %s %s has room again", s.Module, s.Kind, s.Name); len(h) <= conditions.HeadlineMax { + return h + } + return fmt.Sprintf("A %s of %s has room again", s.Kind, s.Module) +} + +// mibWords is a size as a person reads it, in MiB with one decimal. +func mibWords(b uint64) string { + return fmt.Sprintf("%.1f MiB", float64(b)/(1024*1024)) +} + +// declaredStores is every module's bucket and log the catalogue declares, by the stream each is. +func declaredStores(ctx context.Context, d *doctor) ([]moduleStore, error) { + buckets, err := d.open.inventory.DeclaredBuckets(ctx) + if err != nil { + return nil, err + } + logs, err := d.open.inventory.DeclaredLogs(ctx) + if err != nil { + return nil, err + } + var out []moduleStore + for _, b := range buckets { + out = append(out, moduleStore{Module: b.Module, Kind: "bucket", Name: b.Name, Stream: "KV_" + b.Bucket()}) + } + for _, l := range logs { + out = append(out, moduleStore{Module: l.Module, Kind: "log", Name: l.Name, Stream: l.Stream()}) + } + return out, nil +} + +// probeStoreFill reads every module's bucket and log on the bus and says each one filling toward its +// cap. One not on the bus says nothing here: the controller creates it on its next raise, and what is +// not there cannot fill. +func probeStoreFill(ctx context.Context, d *doctor) ([]conditions.Observation, error) { + stores, err := declaredStores(ctx, d) + if err != nil { + return nil, err + } + js := d.js.Context() + var out []conditions.Observation + for _, s := range stores { + if ctx.Err() != nil { + return nil, ctx.Err() + } + info, err := js.StreamInfo(s.Stream, nats.Context(ctx)) + switch { + case errors.Is(err, nats.ErrStreamNotFound): + continue + case err != nil: + return nil, fmt.Errorf("%s's %s %s cannot be read: %w", s.Module, s.Kind, s.Name, err) + } + if info.Config.MaxBytes <= 0 { + continue + } + f := filled{Bytes: info.State.Bytes, Max: uint64(info.Config.MaxBytes)} + if o, found := fillObservation(s, f, isOpen(ctx, d.keeper, fillKey(s))); found { + out = append(out, o) + } + } + return sortedFound(out), nil +} diff --git a/cmd/mesh-controller/store_fill_test.go b/cmd/mesh-controller/store_fill_test.go new file mode 100644 index 00000000..7ede9fe7 --- /dev/null +++ b/cmd/mesh-controller/store_fill_test.go @@ -0,0 +1,76 @@ +package main + +import ( + "strings" + "testing" + + "github.com/novox/mesh-controller/internal/conditions" +) + +// **A filling log or bucket is said at three quarters of its cap and cleared below seven tenths** +// (novox/hq ADR 0297 §6): raised at 75% naming the module, the store and its fill; not at 74.9%; kept +// between 70% and 75% while it is open, and not raised there when it is not; and cleared — said no +// more — once it reads below 70%, open or not. +func TestAFillingStoreIsRaisedAtThreeQuartersAndClearedBelowSevenTenths(t *testing.T) { + const mib = 1024 * 1024 + log := moduleStore{Module: "mesh-issues", Kind: "log", Name: "changes", Stream: "LOG_mesh-issues_changes"} + bucket := moduleStore{Module: "mesh-issues", Kind: "bucket", Name: "issues", Stream: "KV_mesh-issues_issues"} + for _, c := range []struct { + name string + store moduleStore + bytes uint64 + max uint64 + open bool + said bool + }{ + {"a log at exactly 75%", log, 768 * mib, 1024 * mib, false, true}, + {"a bucket at exactly 75%", bucket, 48 * mib, 64 * mib, false, true}, + {"a bucket full", bucket, 64 * mib, 64 * mib, false, true}, + {"a log just under 75%", log, 768*mib - 1, 1024 * mib, false, false}, + {"a bucket at 72%, not open", bucket, 64 * mib * 72 / 100, 64 * mib, false, false}, + {"a bucket at 72%, open: kept", bucket, 64 * mib * 72 / 100, 64 * mib, true, true}, + {"a log at exactly 70%, open: kept", log, 700 * mib, 1000 * mib, true, true}, + {"a log just under 70%, open: cleared", log, 700*mib - 1, 1000 * mib, true, false}, + {"a bucket at 10%, open: cleared", bucket, 64 * mib / 10, 64 * mib, true, false}, + {"a store with no cap", bucket, 100, 0, true, false}, + } { + o, said := fillObservation(c.store, filled{Bytes: c.bytes, Max: c.max}, c.open) + if said != c.said { + t.Errorf("%s: said %v, want %v", c.name, said, c.said) + continue + } + if !said { + continue + } + if o.Key() != fillKey(c.store) || o.Kind != kindStoreFilling || o.Severity != conditions.Warning { + t.Errorf("%s: raised as %s (%s, %s)", c.name, o.Key(), o.Kind, o.Severity) + } + for _, part := range []string{c.store.Module, c.store.Kind + " " + c.store.Name, "%"} { + if !strings.Contains(o.Summary, part) || !strings.Contains(o.Headline, part) { + t.Errorf("%s: does not name %q: %q / %q", c.name, part, o.Summary, o.Headline) + } + } + if !strings.Contains(o.Summary, " of ") || !strings.Contains(o.Said, "bytes") { + t.Errorf("%s: does not say the fill as bytes of the cap: %q / %q", c.name, o.Summary, o.Said) + } + } + o, _ := fillObservation(log, filled{Bytes: 800 * mib, Max: 1024 * mib}, false) + if !strings.Contains(o.Summary, "800.0 MiB of 1024.0 MiB, 78%") { + t.Errorf("the fill is said as %q", o.Summary) + } +} + +// The words the operator reads are plain, for every store a module may name. +func TestAFillingStoreIsSaidInPlainWords(t *testing.T) { + for _, s := range []moduleStore{ + {Module: "mesh-issues", Kind: "log", Name: "changes", Stream: "LOG_mesh-issues_changes"}, + {Module: "a-module-with-a-very-long-name-indeed", Kind: "bucket", Name: "a-bucket-with-a-long-name", + Stream: "KV_a-module-with-a-very-long-name-indeed_a-bucket-with-a-long-name"}, + } { + o, _ := fillObservation(s, filled{Bytes: 90, Max: 100}, false) + if why, ok := conditions.PlainWords(conditions.Words{Headline: o.Headline, Explanation: o.Explanation, + Resolved: o.Resolved, Needs: o.Needs}); !ok { + t.Errorf("%s: %s", s.Stream, why) + } + } +} diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index 70ae35b8..8821e286 100644 --- a/internal/broker/jetstream.go +++ b/internal/broker/jetstream.go @@ -448,3 +448,41 @@ func (j *JetStream) BucketNames() ([]string, error) { } return out, lister.Error() } + +// EnsureLog creates a module's log if it is absent and brings its configuration to match if it is +// present (novox/hq ADR 0297). +// +// **An update, never a delete and recreate**, for a bucket's reason: recreating discards what the log +// holds, and a log is a module's record. A configuration the server will not change in place (its +// storage, say, on a stream made by hand) is said as this assertion's error and the stream is left as +// it is — never removed to be made again. +func (j *JetStream) EnsureLog(l Log) error { + js, err := jetstream.New(j.conn) + if err != nil { + return err + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if _, err := js.CreateOrUpdateStream(ctx, l.Config()); err != nil { + return fmt.Errorf("asserting log %s: %w", l.Stream(), err) + } + return nil +} + +// LogStreams is every log stream on the server: every stream whose name starts with LogStreamPrefix. +func (j *JetStream) LogStreams() ([]string, error) { + js, err := jetstream.New(j.conn) + if err != nil { + return nil, err + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + lister := js.StreamNames(ctx) + var out []string + for name := range lister.Name() { + if strings.HasPrefix(name, LogStreamPrefix) { + out = append(out, name) + } + } + return out, lister.Err() +} diff --git a/internal/broker/log.go b/internal/broker/log.go new file mode 100644 index 00000000..4d2b68a5 --- /dev/null +++ b/internal/broker/log.go @@ -0,0 +1,195 @@ +package broker + +import ( + "fmt" + "sort" + "strings" + + "github.com/nats-io/nats.go/jetstream" +) + +// A module's log on the bus (novox/hq ADR 0297): its record of operations, each entry appended under +// a key and kept as long as the log is. +// +// A log follows a bucket in every respect (ADR 0201): the module names it locally, the mesh derives +// its stream and subjects, the controller creates it from the catalogue on every raise and from +// registration, never a module, and **no path of the mesh removes one** — a log whose declaration +// is gone is reported, as a bucket is. Unlike a bucket, a log is its owner's alone: no other module +// reads it, so there is no read of a log to grant or issue. +// +// Pure, but for the stream's configuration, which is the server's own type so that what is tested +// is exactly what is sent; jetstream.go is the part that asks a server. + +// The mesh's caps on a log: what an entry's value may weigh, what a message on the log's stream may +// weigh — the value and its headers, which the server counts in a message's size, so a value of the +// full size still fits with the expected-last-sequence header an append carries — and what a log holds +// when its module says nothing and at most. +const ( + LogMaxEntryBytes = 256 * 1024 + LogMaxMessageBytes = LogMaxEntryBytes + 4*1024 + LogDefaultMiB = 1024 + LogMostMiB = 8192 +) + +// A Log is one module's declared log as the bus holds it. +type Log struct { + Module string + Name string + // MaxMiB is its cap in MiB; zero is LogDefaultMiB. + MaxMiB int +} + +// LogStreamPrefix starts every log's stream name, which no other stream of the mesh's starts with. +const LogStreamPrefix = "LOG_" + +// LogStreamName is the stream a module's log lives in: `LOG__`. The module and the +// local name are each one token with no underscore, so two modules can never derive one stream. +func LogStreamName(module, name string) string { return LogStreamPrefix + module + "_" + name } + +// LogSubject is the subject a log's entries are published under, without the key: an entry for key K +// is on `.`. +func LogSubject(module, name string) string { return "mesh.log." + module + "." + name } + +// Stream is this log's stream name. +func (l Log) Stream() string { return LogStreamName(l.Module, l.Name) } + +// Subject is this log's subject, without the key. +func (l Log) Subject() string { return LogSubject(l.Module, l.Name) } + +// MaxBytes is this log's cap in bytes. +func (l Log) MaxBytes() int64 { + mib := l.MaxMiB + if mib <= 0 { + mib = LogDefaultMiB + } + return int64(mib) * 1024 * 1024 +} + +// Why is carried into the server's description of the stream, so somebody reading the server's own +// state finds whose it is and why it is kept. +func (l Log) Why() string { + return fmt.Sprintf("%s's log %q (novox/hq ADR 0297): its record of operations, one entry per operation "+ + "under its key, appended by %s alone and kept as long as the log; never removed by the mesh, because "+ + "it is data", l.Module, l.Name, l.Module) +} + +// Config is the stream a log is, field by field as novox/hq ADR 0297 fixes it: a file stream kept by +// limits, with no maximum age and no cap per key, whose message is an entry's value and 4 KiB of +// headers at most, that refuses a new entry when full rather than +// drop an old one, and that refuses deleting an entry or purging it. Direct gets are allowed, which +// is how the runtime reads it. +func (l Log) Config() jetstream.StreamConfig { + return jetstream.StreamConfig{ + Name: l.Stream(), + Description: l.Why(), + Subjects: []string{l.Subject() + ".>"}, + Storage: jetstream.FileStorage, + Retention: jetstream.LimitsPolicy, + MaxAge: 0, + MaxMsgs: -1, + MaxMsgsPerSubject: -1, + MaxBytes: l.MaxBytes(), + MaxMsgSize: LogMaxMessageBytes, + Discard: jetstream.DiscardNew, + AllowDirect: true, + DenyDelete: true, + DenyPurge: true, + Replicas: 1, + } +} + +// LogIssued is one log an assignment may reach, by the name its module uses for it (novox/hq ADR +// 0297): its stream, the subject its entries go under, and whether it may append. Only the owner's +// instances are issued a log, so Writes is always true today; it is said so the runtime need not +// assume it. +type LogIssued struct { + Name string `json:"name"` + Stream string `json:"stream"` + Subject string `json:"subject"` + Writes bool `json:"writes"` +} + +// logsIssuedFor is every log a module's code may reach, as its membership lists them: its own. +func logsIssuedFor(d Declared) []LogIssued { + var out []LogIssued + for _, l := range d.Logs { + if !safeSubject.MatchString(d.Module) || !safeSubject.MatchString(l.Name) { + continue + } + out = append(out, LogIssued{Name: l.Name, Stream: LogStreamName(d.Module, l.Name), + Subject: LogSubject(d.Module, l.Name), Writes: true}) + } + return out +} + +// logGrants is what a principal publishes to reach the logs its module keeps: for each, appending +// under the log's subjects, binding to its stream, and reading it directly. Nothing more — the +// runtime reads a log by direct gets alone and makes no consumer on it — and nothing of any other +// module's log, because a log is its owner's alone. Replies come on the principal's inbox, as for a +// bucket. +func logGrants(module string, names []string) []string { + if !safeSubject.MatchString(module) { + return nil + } + var out []string + for _, name := range names { + if !safeSubject.MatchString(name) { + continue + } + stream := LogStreamName(module, name) + out = append(out, + LogSubject(module, name)+".>", + "$JS.API.STREAM.INFO."+stream, + "$JS.API.DIRECT.GET."+stream, + "$JS.API.DIRECT.GET."+stream+".>") + } + return out +} + +// logNames is the local names of a module's logs. +func logNames(logs []Log) []string { + out := make([]string, 0, len(logs)) + for _, l := range logs { + out = append(out, l.Name) + } + return out +} + +// A LogAsserter is the part of a JetStream connection log assertion needs. It has no way to remove a +// log, by design: nothing the mesh runs asks for one. +type LogAsserter interface { + // EnsureLog creates the log's stream if absent and brings its configuration to match if present, + // never deleting or recreating it. + EnsureLog(l Log) error + // LogStreams is every log stream on the server: every stream named with LogStreamPrefix. + LogStreams() ([]string, error) +} + +// RaiseLogs asserts every declared log and answers the log streams on the server that nothing +// declares any more. +// +// **Those are reported, never removed** (novox/hq ADR 0297 §2, ADR 0201 §15–16): a log is a module's +// record, and a manifest edited, a module renamed or unassigned is an ordinary day's work that must +// not take a record with it. Removing one is a person's act, outside the mesh. +func RaiseLogs(a LogAsserter, logs []Log) (undeclared []string, err error) { + sorted := append([]Log(nil), logs...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].Stream() < sorted[j].Stream() }) + declared := map[string]bool{} + for _, l := range sorted { + if err := a.EnsureLog(l); err != nil { + return nil, fmt.Errorf("asserting %s's log %q: %w", l.Module, l.Name, err) + } + declared[l.Stream()] = true + } + names, err := a.LogStreams() + if err != nil { + return nil, fmt.Errorf("listing the bus's logs: %w", err) + } + for _, n := range names { + if strings.HasPrefix(n, LogStreamPrefix) && !declared[n] { + undeclared = append(undeclared, n) + } + } + sort.Strings(undeclared) + return undeclared, nil +} diff --git a/internal/broker/log_test.go b/internal/broker/log_test.go new file mode 100644 index 00000000..3119fb03 --- /dev/null +++ b/internal/broker/log_test.go @@ -0,0 +1,294 @@ +package broker + +import ( + "context" + "encoding/json" + "go/ast" + "go/parser" + "go/token" + "io/fs" + "path/filepath" + "slices" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go/jetstream" +) + +// **The stream a log is, field by field** (novox/hq ADR 0297 §2, the shared contract): a file stream +// kept by limits, no maximum age, no cap per key, the declared cap, a message of at most 260 KiB (a +// value of 256 KiB and 4 KiB of headers), +// refusing what comes next when full, read directly, refusing a delete or a purge, one replica, and +// a description that says whose it is and why. +func TestALogsStreamIsAsTheDecisionFixesIt(t *testing.T) { + got := Log{Module: "mesh-issues", Name: "changes"}.Config() + want := jetstream.StreamConfig{ + Name: "LOG_mesh-issues_changes", + Description: got.Description, + Subjects: []string{"mesh.log.mesh-issues.changes.>"}, + Storage: jetstream.FileStorage, + Retention: jetstream.LimitsPolicy, + MaxAge: 0, + MaxMsgs: -1, + MaxMsgsPerSubject: -1, + MaxBytes: 1024 * 1024 * 1024, + MaxMsgSize: 260 * 1024, + Discard: jetstream.DiscardNew, + AllowDirect: true, + DenyDelete: true, + DenyPurge: true, + Replicas: 1, + } + a, _ := json.Marshal(got) + b, _ := json.Marshal(want) + if string(a) != string(b) { + t.Fatalf("the log's stream is\n %s\nwant\n %s", a, b) + } + if !strings.Contains(got.Description, "mesh-issues") || !strings.Contains(got.Description, "ADR 0297") { + t.Fatalf("the description does not say whose the log is and why: %q", got.Description) + } + if c := (Log{Module: "m", Name: "n", MaxMiB: 8192}).Config(); c.MaxBytes != 8192*1024*1024 { + t.Fatalf("a cap of 8192 MiB is %d bytes", c.MaxBytes) + } +} + +// **The membership names each of the owner's logs** with exactly the field names the runtime reads: +// `logs`, and in each `name`, `stream`, `subject`, `writes`. A module with no log is issued none, and +// the field is absent. +func TestAMembershipListsItsModulesLogs(t *testing.T) { + m := MembershipFor("one", Declared{Module: "mesh-issues", + Logs: []Log{{Module: "mesh-issues", Name: "changes"}}}, Placements{}) + raw, err := json.Marshal(m) + if err != nil { + t.Fatal(err) + } + var back map[string]json.RawMessage + if err := json.Unmarshal(raw, &back); err != nil { + t.Fatal(err) + } + if got := string(back["logs"]); got != + `[{"name":"changes","stream":"LOG_mesh-issues_changes","subject":"mesh.log.mesh-issues.changes","writes":true}]` { + t.Fatalf("the membership's logs are %s", got) + } + none, _ := json.Marshal(MembershipFor("one", Declared{Module: "audit"}, Placements{})) + if strings.Contains(string(none), `"logs"`) { + t.Fatalf("a module with no log was issued logs: %s", none) + } +} + +// logGrantsOf is the grants of a principal that are about logs. +func logGrantsOf(publish []string) []string { + var out []string + for _, s := range publish { + if strings.HasPrefix(s, "mesh.log.") || strings.Contains(s, ".LOG_") { + out = append(out, s) + } + } + slices.Sort(out) + return out +} + +// **The runtime is granted exactly the contract's subjects for each log it carries, and nothing else +// of any log**: appending under the log's subjects, binding to its stream, reading it directly. No +// consumer, no delete, no purge, and nothing of a log its modules do not keep. +func TestTheRuntimeIsGrantedItsModulesLogsAndNoMore(t *testing.T) { + perms, err := PermissionsFor(Principal{Kind: KindNodeTools, Node: "one", Module: RuntimeModule, + Carries: []Declared{ + {Module: "mesh-issues", Logs: []Log{{Module: "mesh-issues", Name: "changes"}}}, + {Module: "audit"}, + }, PasswordHash: "x"}) + if err != nil { + t.Fatal(err) + } + want := []string{ + "$JS.API.DIRECT.GET.LOG_mesh-issues_changes", + "$JS.API.DIRECT.GET.LOG_mesh-issues_changes.>", + "$JS.API.STREAM.INFO.LOG_mesh-issues_changes", + "mesh.log.mesh-issues.changes.>", + } + if got := logGrantsOf(perms.Publish); !slices.Equal(got, want) { + t.Fatalf("the runtime is granted\n %q\nwant\n %q", got, want) + } + for _, s := range perms.Subscribe { + if strings.HasPrefix(s, "mesh.log.") || strings.Contains(s, "LOG_") { + t.Fatalf("the runtime subscribes a log's subjects: %q", s) + } + } + // A module running on its own account is granted the same for its own logs. + own, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "mesh-issues", + Logs: []string{"changes"}, PasswordHash: "x"}) + if err != nil { + t.Fatal(err) + } + if got := logGrantsOf(own.Publish); !slices.Equal(got, want) { + t.Fatalf("the module's own account is granted\n %q\nwant\n %q", got, want) + } + // And a module with no log, nothing of any. + none, err := PermissionsFor(Principal{Kind: KindModule, Node: "one", Module: "audit", PasswordHash: "x"}) + if err != nil { + t.Fatal(err) + } + if got := logGrantsOf(none.Publish); len(got) != 0 { + t.Fatalf("a module with no log is granted %q", got) + } +} + +// A log name or module that is not one plain token is issued and granted nothing rather than a +// pattern that happens to parse. +func TestALogThatNamesNoStreamGrantsNothing(t *testing.T) { + if got := logGrants("a", []string{"x.y", "x>", "*", ""}); len(got) != 0 { + t.Fatalf("granted %v for logs that name no stream", got) + } + if got := logGrants("a.b", []string{"c"}); len(got) != 0 { + t.Fatalf("granted %v for a module that is not one token", got) + } +} + +type logs struct { + ensured []string + on []string +} + +func (l *logs) EnsureLog(x Log) error { l.ensured = append(l.ensured, x.Stream()); return nil } +func (l *logs) LogStreams() ([]string, error) { return l.on, nil } + +// Every declared log is asserted; one on the server that nothing declares is said, not removed — and +// the asserter has no way to remove one. +func TestRaisingLogsReportsWhatNothingDeclares(t *testing.T) { + l := &logs{on: []string{"LOG_mesh-issues_changes", "LOG_gone_old", "KV_not_a_log"}} + undeclared, err := RaiseLogs(l, []Log{{Module: "mesh-issues", Name: "changes"}, {Module: "a", Name: "b"}}) + if err != nil { + t.Fatal(err) + } + if !slices.Equal(l.ensured, []string{"LOG_a_b", "LOG_mesh-issues_changes"}) { + t.Fatalf("asserted %v", l.ensured) + } + if !slices.Equal(undeclared, []string{"LOG_gone_old"}) { + t.Fatalf("reported %v", undeclared) + } +} + +// **No path of the mesh deletes or purges a log or a bucket, or recreates one** (novox/hq ADR 0297 §2, +// ADR 0201 §15–16): no code of this repository outside its tests calls the client's stream or bucket +// delete, or purges a stream, but for the controller lease's own stream, which purges its own history +// below a sequence (internal/lease). A new call is a decision, not a refactor. +func TestNoPathDeletesALogOrABucket(t *testing.T) { + removers := map[string]bool{"DeleteStream": true, "DeleteKeyValue": true, "PurgeStream": true, + "DeleteObjectStore": true} + root := filepath.Join("..", "..") + var found []string + for _, dir := range []string{"cmd", "internal"} { + err := filepath.WalkDir(filepath.Join(root, dir), func(path string, e fs.DirEntry, err error) error { + if err != nil || e.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") { + return err + } + f, err := parser.ParseFile(token.NewFileSet(), path, nil, 0) + if err != nil { + return err + } + ast.Inspect(f, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok { + return true + } + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok { + return true + } + name := sel.Sel.Name + if removers[name] || (name == "Purge" && !strings.Contains(filepath.ToSlash(path), "internal/lease/")) { + found = append(found, path+": "+name) + } + return true + }) + for _, lit := range stringsIn(f) { + if strings.Contains(lit, "$JS.API.STREAM.DELETE") || strings.Contains(lit, "$JS.API.STREAM.PURGE") { + // Only the writers table names it, as a subject no one but the controller may publish. + if !strings.HasSuffix(filepath.ToSlash(path), "internal/broker/writers.go") { + found = append(found, path+": "+lit) + } + } + } + return nil + }) + if err != nil { + t.Fatal(err) + } + } + if len(found) > 0 { + t.Fatalf("a path of the mesh removes a stream, a bucket or what one holds:\n %s", strings.Join(found, "\n ")) + } +} + +// stringsIn is every string literal of a file. +func stringsIn(f *ast.File) []string { + var out []string + ast.Inspect(f, func(n ast.Node) bool { + if lit, ok := n.(*ast.BasicLit); ok && lit.Kind == token.STRING { + out = append(out, lit.Value) + } + return true + }) + return out +} + +// Against a real server: a log is created as its configuration says, asserting it again keeps what +// it holds, a changed cap is brought to match in place, an entry cannot be deleted from it, and one +// nothing declares any more is reported and stays. +func TestALogIsAssertedInPlaceAndNeverRemoved(t *testing.T) { + bus := aLiveBus(t) + l := Log{Module: "logtest", Name: "changes", MaxMiB: 1} + if _, err := RaiseLogs(bus, []Log{l}); err != nil { + t.Fatalf("a real server refused a module's log: %v", err) + } + js, err := jetstream.New(bus.Conn()) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + ack, err := js.Publish(ctx, l.Subject()+".7", []byte(`{"op":"opened"}`)) + if err != nil { + t.Fatalf("an append to the log was refused: %v", err) + } + stream, err := js.Stream(ctx, l.Stream()) + if err != nil { + t.Fatal(err) + } + have := stream.CachedInfo().Config + want := l.Config() + if have.Storage != want.Storage || have.Retention != want.Retention || have.MaxAge != 0 || + have.MaxMsgsPerSubject != -1 || have.MaxBytes != want.MaxBytes || have.MaxMsgSize != want.MaxMsgSize || + have.Discard != jetstream.DiscardNew || !have.AllowDirect || !have.DenyDelete || !have.DenyPurge || + have.Replicas != 1 || !slices.Equal(have.Subjects, want.Subjects) { + t.Fatalf("the server holds the log as %+v", have) + } + if err := stream.DeleteMsg(ctx, ack.Sequence); err == nil { + t.Fatal("an entry was deleted from a log") + } + l.MaxMiB = 2 + if _, err := RaiseLogs(bus, []Log{l}); err != nil { + t.Fatalf("asserting the log again failed, so a restart would: %v", err) + } + info, err := stream.Info(ctx) + if err != nil { + t.Fatal(err) + } + if info.Config.MaxBytes != 2*1024*1024 { + t.Fatalf("the changed cap was not brought to match: %d", info.Config.MaxBytes) + } + if info.State.Msgs < 1 { + t.Fatal("asserting the log again lost what it held") + } + undeclared, err := RaiseLogs(bus, nil) + if err != nil { + t.Fatal(err) + } + if !slices.Contains(undeclared, l.Stream()) { + t.Fatalf("a log nothing declares was not reported: %v", undeclared) + } + if _, err := js.Stream(ctx, l.Stream()); err != nil { + t.Fatalf("a log nothing declares is gone: %v", err) + } +} diff --git a/internal/broker/membership.go b/internal/broker/membership.go index b44aa6af..1f662589 100644 --- a/internal/broker/membership.go +++ b/internal/broker/membership.go @@ -51,6 +51,10 @@ type Membership struct { // and refuses, with the reason, what is not on it — the bus enforces only the union over every // module on the machine. State []StateIssued `json:"state,omitempty"` + // Logs is every log this module's code may reach, by the name it uses for each (novox/hq ADR 0297): + // its own and no other's, since a log is its owner's alone. The runtime answers a bundle's log verbs + // from this list and refuses, with the reason, a log not on it. + Logs []LogIssued `json:"logs,omitempty"` // SeatTraffic is what this module's code may submit, say, hear, take, ask, answer and read on seats // that name their caller or their kind (novox/hq ADR 0259 §3). The runtime carrying the module // publishes, takes and answers for it only what is listed here: the bus enforces only the union @@ -121,6 +125,7 @@ func MembershipFor(node string, d Declared, where Placements) Membership { } } m.State = stateIssuedFor(d, node) + m.Logs = logsIssuedFor(d) t := SeatTrafficOf(d.Module, d.Holds, d.Uses, d.Watches) for _, s := range append(append([]Seat{}, d.Uses...), d.Watches...) { if !s.Kinded { diff --git a/internal/broker/nats.go b/internal/broker/nats.go index 659a9cc8..3cdf473a 100644 --- a/internal/broker/nats.go +++ b/internal/broker/nats.go @@ -132,6 +132,8 @@ type Principal struct { // no other; KeyedReads the keys of others' state it reads one key at a time (novox/hq ADR 0260). PerMachine []string KeyedReads []KeyedRead + // Logs is the local names of the logs this principal's module keeps (novox/hq ADR 0297). + Logs []string // SnapshotsTheBus is the bus's own module, the one holding mesh-broker (novox/hq ADR 0235). Its // whole authority is BusSnapshotGrants: it copies the streams for the night's backup and nothing @@ -746,6 +748,8 @@ func PermissionsFor(p Principal) (Permissions, error) { // watched, its own written too. pub = append(pub, stateGrants(stateAccess{Module: p.Module, Node: p.Node, Keeps: p.State, PerMachine: p.PerMachine, Reads: p.Reads, KeyedReads: p.KeyedReads})...) + // And its logs (novox/hq ADR 0297): appended to and read directly, its own alone. + pub = append(pub, logGrants(p.Module, p.Logs)...) // 6. Its traffic on seats that name their caller or their kind, ask proofs or keep records // (novox/hq ADR 0259 §3). @@ -833,6 +837,12 @@ func PermissionsFor(p Principal) (Permissions, error) { pub = append(pub, stateGrants(stateAccess{Module: d.Module, Node: p.Node, Keeps: stateNames(d.State), PerMachine: perMachineNames(d.State), Reads: d.Reads, KeyedReads: d.KeyedReads})...) } + // **And it keeps the logs of the modules it carries** (novox/hq ADR 0297): each module's own, the + // union over them. That one module's code does not append to another's log through it is the + // runtime's to keep, from the logs each membership lists. + for _, d := range p.Carries { + pub = append(pub, logGrants(d.Module, logNames(d.Logs))...) + } // **Never the traffic of a trusted holder** (novox/hq ADR 0259 §8): the machine's runtime runs as the // operator's account, which every agent runs as, so a module saying warrants or speaking for a kind // that proves its sender is never composed into it — refused here, naming it, whatever registration diff --git a/internal/broker/users.go b/internal/broker/users.go index 86ff3b4e..08cac990 100644 --- a/internal/broker/users.go +++ b/internal/broker/users.go @@ -35,6 +35,8 @@ type Declared struct { Invokes []string // State is the state it keeps, each a bucket its instances write (novox/hq ADR 0201). State []Bucket + // Logs are the logs it keeps, each a stream its instances append to (novox/hq ADR 0297). + Logs []Log // Reads are other modules' state it reads, each `.` (novox/hq ADR 0201). Reads []string // KeyedReads are keys of other modules' state it reads, each one key alone: what a seat's holder is @@ -124,6 +126,7 @@ func Users(r Records) ([]Principal, error) { Holds: d.Holds, Uses: d.Uses, Watches: d.Watches, Invokes: d.Invokes, State: stateNames(d.State), Reads: d.Reads, SnapshotsTheBus: d.SnapshotsTheBus, PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads, + Logs: logNames(d.Logs), }) } if runtimeHere { diff --git a/internal/catalogue/logs.go b/internal/catalogue/logs.go new file mode 100644 index 00000000..7856b8ce --- /dev/null +++ b/internal/catalogue/logs.go @@ -0,0 +1,105 @@ +package catalogue + +import ( + "bytes" + "encoding/json" + "fmt" +) + +// What a module may call its logs (novox/hq ADR 0297). +// +// A log is a module's record of operations: entries appended under a key, each kept as long as the +// log is, in the order they came. A module names each log it owns **locally** — `changes`, never a +// stream or a subject (ADR 0201 §4) — and the mesh derives the stream from the module and the local +// name, as it derives a bucket. So the rule for a log's name is a state name's: one plain token, and +// a module that keeps a log has a name that is one plain token too. + +// The mesh's caps on a log, in MiB: what a module may ask, and what it gets when it asks nothing. +const ( + LogLeastMiB = 1 + LogMostMiB = 8192 + LogDefaultMiB = 1024 +) + +// LogDeclaration is one log a module owns: its local name, and how large it may grow. +type LogDeclaration struct { + Name string `json:"name"` + // MaxMiB is the log's cap in MiB; zero is LogDefaultMiB. When full, the log refuses new entries + // and never drops old ones. + MaxMiB int `json:"max-mib,omitempty"` +} + +// Cap is the log's cap in MiB, the default where none is said. +func (l LogDeclaration) Cap() int { + if l.MaxMiB == 0 { + return LogDefaultMiB + } + return l.MaxMiB +} + +// UnmarshalJSON reads a log as its bare name, or as {name, max-mib}. +func (l *LogDeclaration) UnmarshalJSON(raw []byte) error { + trimmed := bytes.TrimSpace(raw) + if len(trimmed) > 0 && trimmed[0] == '"' { + return json.Unmarshal(trimmed, &l.Name) + } + var full struct { + Name string `json:"name"` + MaxMiB *int `json:"max-mib"` + } + dec := json.NewDecoder(bytes.NewReader(trimmed)) + dec.DisallowUnknownFields() + if err := dec.Decode(&full); err != nil { + return fmt.Errorf("a log is either a name or {name, max-mib}: %w", typedUnknown(err)) + } + l.Name = full.Name + l.MaxMiB = 0 + if full.MaxMiB != nil { + // Said, and said as nothing: refused rather than read as the default, which it did not say. + if *full.MaxMiB == 0 { + return fmt.Errorf("log %q: max-mib is between %d and %d, not 0", full.Name, LogLeastMiB, LogMostMiB) + } + l.MaxMiB = *full.MaxMiB + } + return nil +} + +// MarshalJSON writes back the short form when there is nothing else to say. +func (l LogDeclaration) MarshalJSON() ([]byte, error) { + if l.MaxMiB == 0 { + return json.Marshal(l.Name) + } + type plain LogDeclaration + return json.Marshal(plain(l)) +} + +// LogProblems is what is wrong with a manifest's logs. +// +// Refused at registration, for a bucket's reason: a stream name the bus cannot hold, or a cap the +// mesh would not grant, is a module that installs, starts, and is refused on its first append. +func LogProblems(m Manifest) []string { + var problems []string + if len(m.Logs) > 0 && !stateName.MatchString(m.Module) { + problems = append(problems, fmt.Sprintf( + "%s keeps a log, and a module's name is part of its logs' names, which take one plain "+ + "name — no dot (novox/hq ADR 0297)", m.Module)) + } + seen := map[string]bool{} + for _, l := range m.Logs { + switch { + case !stateName.MatchString(l.Name): + problems = append(problems, fmt.Sprintf( + "%s keeps log %q: a log is named locally — lower-case letters, digits and hyphens, "+ + "no dot and no underscore; the mesh derives the stream (novox/hq ADR 0297)", m.Module, l.Name)) + case seen[l.Name]: + problems = append(problems, fmt.Sprintf("%s keeps log %q twice", m.Module, l.Name)) + } + seen[l.Name] = true + if l.MaxMiB != 0 && (l.MaxMiB < LogLeastMiB || l.MaxMiB > LogMostMiB) { + problems = append(problems, fmt.Sprintf( + "%s caps log %q at %d MiB; a log holds between %d and %d MiB (novox/hq ADR 0297)", + m.Module, l.Name, l.MaxMiB, LogLeastMiB, LogMostMiB)) + } + } + return problems +} diff --git a/internal/catalogue/logs_test.go b/internal/catalogue/logs_test.go new file mode 100644 index 00000000..76e7ee0d --- /dev/null +++ b/internal/catalogue/logs_test.go @@ -0,0 +1,68 @@ +package catalogue + +import ( + "encoding/json" + "strings" + "testing" +) + +// A module declares the logs it keeps (novox/hq ADR 0297 §1): a log by its bare name, or with its cap +// in MiB; the default cap where none is said. +func TestAManifestMaySayWhatLogsItKeeps(t *testing.T) { + m, err := ParseManifest([]byte(`{"module":"mesh-issues","version":"1",` + + `"logs":["changes",{"name":"moves","max-mib":2048}]}`)) + if err != nil { + t.Fatal(err) + } + if len(m.Logs) != 2 || m.Logs[0].Name != "changes" || m.Logs[1].Name != "moves" { + t.Fatalf("logs not read: %+v", m.Logs) + } + if m.Logs[0].Cap() != LogDefaultMiB || m.Logs[0].Cap() != 1024 || m.Logs[1].Cap() != 2048 { + t.Fatalf("caps read as %d and %d", m.Logs[0].Cap(), m.Logs[1].Cap()) + } + out, _ := json.Marshal(m.Logs) + if string(out) != `["changes",{"name":"moves","max-mib":2048}]` { + t.Fatalf("written back as %s", out) + } + for _, edge := range []string{`{"name":"a","max-mib":1}`, `{"name":"a","max-mib":8192}`} { + if _, err := ParseManifest([]byte(`{"module":"a","version":"1","logs":[` + edge + `]}`)); err != nil { + t.Errorf("%s is inside the caps and was refused: %v", edge, err) + } + } +} + +// The catalogue check refuses a log outside the caps, with a name that is not one plain token, with a +// field the mesh does not know, or kept by a module whose own name is not one plain token. +func TestALogIsNamedLocallyAndCapped(t *testing.T) { + for _, c := range []struct{ manifest, says string }{ + {`{"module":"a","version":"1","logs":["mesh.changes"]}`, `keeps log "mesh.changes": a log is named locally`}, + {`{"module":"a","version":"1","logs":["my_changes"]}`, `keeps log "my_changes"`}, + {`{"module":"a","version":"1","logs":["Changes"]}`, `keeps log "Changes"`}, + {`{"module":"a","version":"1","logs":["c","c"]}`, `keeps log "c" twice`}, + {`{"module":"a","version":"1","logs":[{"name":"c","max-mib":8193}]}`, `between 1 and 8192 MiB`}, + {`{"module":"a","version":"1","logs":[{"name":"c","max-mib":-1}]}`, `between 1 and 8192 MiB`}, + {`{"module":"a","version":"1","logs":[{"name":"c","max-mib":0}]}`, `max-mib is between 1 and 8192, not 0`}, + {`{"module":"a","version":"1","logs":[{"name":"c","stream":"LOG_x"}]}`, `{name, max-mib}`}, + {`{"module":"a.b","version":"1","logs":["c"]}`, `no dot`}, + } { + _, err := ParseManifest([]byte(c.manifest)) + if err == nil { + t.Errorf("%s was accepted", c.manifest) + continue + } + if !strings.Contains(err.Error(), c.says) { + t.Errorf("%s refused for the wrong reason: %v", c.manifest, err) + } + } +} + +// **Across the whole catalogue**: every log is named locally and capped within the mesh's caps. +func TestEveryManifestsLogsAreLocalAndCapped(t *testing.T) { + var problems []string + for _, m := range theCatalogue(t) { + problems = append(problems, LogProblems(m)...) + } + if len(problems) > 0 { + t.Fatalf("the catalogue's logs are not what ADR 0297 says:\n %s", strings.Join(problems, "\n ")) + } +} diff --git a/internal/catalogue/manifest.go b/internal/catalogue/manifest.go index be0e5871..611e46f5 100644 --- a/internal/catalogue/manifest.go +++ b/internal/catalogue/manifest.go @@ -465,6 +465,11 @@ type Manifest struct { // (novox/hq ADR 0201). Not history — that is an event — and never a secret, sealed or not. State []StateDeclaration `json:"state,omitempty"` + // Logs are the logs of operations this module keeps on the bus, by local name: each a stream the + // controller creates and never removes, which every instance of the module appends to and reads + // (novox/hq ADR 0297). A log is its owner's alone: no other module reads it. + Logs []LogDeclaration `json:"logs,omitempty"` + // Settings are the defaults this module gives its settings (novox/hq ADR 0262): each key a file, // a contribution or a served fact asks for as `${setting:}`, its default, and why that // default. Only a preference has one — a font size, a width, a number of workers — and a value @@ -1507,7 +1512,7 @@ func ParseManifest(raw []byte) (Manifest, error) { name string n int }{{"consumes", len(m.Consumes)}, {"uses", len(m.Uses)}, {"invokes", len(m.Invokes)}, - {"state", len(m.State)}, {"reads", len(m.Reads)}} { + {"state", len(m.State)}, {"logs", len(m.Logs)}, {"reads", len(m.Reads)}} { if f.n > 0 { said = append(said, f.name) } @@ -1618,6 +1623,8 @@ func ParseManifest(raw []byte) (Manifest, error) { problems = append(problems, EventProblems(m)...) // And what it may call its state, and whose it may read (state.go, novox/hq ADR 0201). problems = append(problems, StateProblems(m)...) + // And what it may call its logs, and how large it may ask them to grow (logs.go, novox/hq ADR 0297). + problems = append(problems, LogProblems(m)...) // And the defaults it gives its settings (setting_defaults.go, novox/hq ADR 0262). problems = append(problems, SettingProblems(m)...) wellFormed := true diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index 68ca4f54..30c1cc5b 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -166,6 +166,8 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio // And the state it keeps and reads (novox/hq ADR 0201). State: bucketsOf(m), Reads: m.Reads, + // And the logs it keeps (novox/hq ADR 0297). + Logs: logsOf(m), // And the tools its health asks (novox/hq ADR 0240): the machine's node-engine is granted them. Checks: catalogue.HealthChecks(m), // And whether it runs as an account of its own (novox/hq ADR 0259 §8). @@ -222,6 +224,29 @@ func (i *Inventory) DeclaredBuckets(ctx context.Context) ([]broker.Bucket, error return out, nil } +// logsOf is the logs a module keeps, as the bus holds them. +func logsOf(m catalogue.Manifest) []broker.Log { + var out []broker.Log + for _, l := range m.Logs { + out = append(out, broker.Log{Module: m.Module, Name: l.Name, MaxMiB: l.Cap()}) + } + return out +} + +// DeclaredLogs is every log the catalogue declares, registered modules assigned or not: a log exists +// from registration, as a bucket does (novox/hq ADR 0297 §2, ADR 0201 §6). +func (i *Inventory) DeclaredLogs(ctx context.Context) ([]broker.Log, error) { + declared, err := i.Catalogue(ctx) + if err != nil { + return nil, fmt.Errorf("cannot read the catalogue: %w", err) + } + var out []broker.Log + for _, m := range declared { + out = append(out, logsOf(m)...) + } + return out, nil +} + func asSeat(s catalogue.SeatDeclaration, declarer string) broker.Seat { seat := broker.Seat{Name: s.Name, Scope: s.Scope, Accepts: s.Accepts, Emits: s.Emits, Serves: catalogue.VerbNames(s.Serves), Kinded: s.Kinded, ByCaller: s.ByCaller, Proofs: s.Proofs,