Files
mesh-delivery/modules/mesh-grants/internal/buscompose/log.go
T
jschoubben 78fedabeb1 mesh-grants: own the bus's composer, so a change to who may say what rebuilds only the grants part
ADR 0593 §1: buscompose's engine moves from mesh-sdk into mesh-grants/internal/buscompose, its golden
user lists unchanged; the controller's raise and assert code is left behind, and the merged switch
test runs inside the part's whole integration test.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
2026-11-02 18:56:19 +01:00

203 lines
7.7 KiB
Go

package buscompose
import (
"fmt"
"strings"
"time"
"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. Only its owner appends to it. Another module reads it when its manifest's
// `reads` names it, `<module>.<log>` (novox/hq ADR 0571 clause 1; the lead, 2026-10-29): by direct gets, and through an
// ordered consumer of its own it creates and lets go of to read from a time or the newest per key and then follow new
// entries; never a durable consumer, never a write.
//
// 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
// MaxAgeDays is how long an entry is kept; zero is as long as the log (catalogue.LogDeclaration).
MaxAgeDays 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_<module>_<name>`. 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 `<LogSubject>.<K>`.
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 unless the log says one (ADR 0565 §10) 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: time.Duration(l.MaxAgeDays) * 24 * time.Hour,
MaxMsgs: -1,
MaxMsgsPerSubject: -1,
MaxBytes: l.MaxBytes(),
MaxMsgSize: LogMaxMessageBytes,
Discard: jetstream.DiscardNew,
AllowDirect: true,
DenyDelete: true,
DenyPurge: true,
Replicas: 1,
}
}
// 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
}
// DeclaredLogs is every log a module placed anywhere declares, as a read names it: `<module>.<log>`.
func DeclaredLogs(r Records) map[string]bool {
out := map[string]bool{}
for _, ds := range r.Assigned {
for _, d := range ds {
for _, l := range d.Logs {
out[d.Module+"."+l.Name] = true
}
}
}
return out
}
// withReadLogs is d with ReadLogs set: of its reads, those naming a log of logs.
func withReadLogs(d Declared, logs map[string]bool) Declared {
d.ReadLogs = nil
for _, r := range d.Reads {
if logs[strings.TrimSpace(r)] {
d.ReadLogs = append(d.ReadLogs, strings.TrimSpace(r))
}
}
return d
}
// readLogGrants is what a reader of other modules' logs publishes, for each `<module>.<log>` read: the stream's info,
// direct gets (an entry by sequence, the last of a key), and an ordered consumer of its own (created with any name and
// filter, pulled, read and deleted). No append, no purge, no delete of an entry: the log's stream refuses those anyway.
func readLogGrants(reads []string) []string {
var out []string
for _, r := range reads {
module, name, ok := strings.Cut(r, ".")
if !ok || !safeSubject.MatchString(module) || !safeSubject.MatchString(name) {
continue
}
stream := LogStreamName(module, name)
out = append(out, "$JS.API.STREAM.INFO."+stream, "$JS.API.DIRECT.GET."+stream, "$JS.API.DIRECT.GET."+stream+".>",
"$JS.API.CONSUMER.CREATE."+stream+".>", "$JS.API.CONSUMER.MSG.NEXT."+stream+".*",
"$JS.API.CONSUMER.INFO."+stream+".*", "$JS.API.CONSUMER.DELETE."+stream+".*")
}
return out
}
// readLogsIssuedFor is every other module's log a module's code reads, as its membership lists them: by the read's
// name, never written.
func readLogsIssuedFor(d Declared) []LogIssued {
var out []LogIssued
for _, r := range d.ReadLogs {
module, name, _ := strings.Cut(r, ".")
out = append(out, LogIssued{Name: r, Stream: LogStreamName(module, name), Subject: LogSubject(module, name)})
}
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
}