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 }