package buscompose import ( "fmt" "time" "github.com/nats-io/nats.go/jetstream" ) // 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 ) // 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__`. 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 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, } }