Merge pull request 'node-tools: serve a module's log verbs (ADR 0297, issues 501, 502)' (#28) from adr-0297-log-verbs into main
This commit was merged in pull request #28.
This commit is contained in:
+3
-2
@@ -35,8 +35,9 @@ if ! command -v node >/dev/null 2>&1 || [ ! -d node_modules/@novox/mesh-sdk ]; t
|
||||
bundleless='^(TestASeatAndAModuleOfOneNameAreEachReached|TestSearch.*|TestASearchByCommandSaysWhatItReplaces)$'
|
||||
# And what of the runtime launches no bundle: how a bundle's seat traffic reaches the bus, what its
|
||||
# environment holds, what a runtime on a module's own account serves (novox/hq ADR 0259), and that a write
|
||||
# by compare-and-set loses no update under contention (novox/hq research 042).
|
||||
runtimeless='^(TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing|TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller|TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone)$'
|
||||
# by compare-and-set loses no update under contention (novox/hq research 042), and a module's log verbs
|
||||
# (novox/hq ADR 0297).
|
||||
runtimeless='^(TestALogAnswersEachVerbOnTheBus|TestALogPageIsBoundedByItsBytes|TestAnAppendAtAStaleExpectIsRefusedAsALostUpdate|TestALogNotIssuedIsRefusedWithTheReason|TestWhatALogCannotHoldIsRefusedWithTheReason|TestALogRefusesAValueNamingACredential|TestARuntimeOnAModulesOwnAccountServesThatModuleAlone|TestNoBusWordReachesABundle|TestAToolCallNeverReachesASeatsEventAcceptOrProof|TestABundlesSeatTrafficGoesThroughItsModulesBus|TestALostCompareAndSetCarriesItsCode|TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate|TestAStaleWriteIsRefusedAndWritesNothing|TestACallOnTheCallerNamedSubjectReachesTheModuleWithItsCaller|TestTheSubjectThatNamesNoCallerStillWorksAndNamesNone)$'
|
||||
fi
|
||||
if command -v gcc >/dev/null 2>&1; then
|
||||
CGO_ENABLED=1 go test -race -count=1 $packages
|
||||
|
||||
@@ -66,6 +66,8 @@ type Membership struct {
|
||||
Tools string `json:"tools"`
|
||||
// State is every bucket the module's code may reach, by the name it uses (novox/hq ADR 0201).
|
||||
State []StateIssued `json:"state,omitempty"`
|
||||
// Logs is every log the module keeps, by its local name (novox/hq ADR 0297): its own alone.
|
||||
Logs []LogIssued `json:"logs,omitempty"`
|
||||
// SeatTraffic is what the module's code may submit, say, take, ask, answer and read on seats that name
|
||||
// their caller or their kind (novox/hq ADR 0259): the runtime does for it only what is listed here.
|
||||
SeatTraffic *SeatTraffic `json:"seat-traffic,omitempty"`
|
||||
|
||||
@@ -0,0 +1,367 @@
|
||||
package bus
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// A module's log on the bus (novox/hq ADR 0297): a stream of operations, one subject per key, that the
|
||||
// controller creates from what the module declared under `logs` and issues to the module's own instances
|
||||
// in its membership. It mirrors a bucket (ADR 0201) in how it is reached: what a module may reach is
|
||||
// decided here, from its membership, and refused with the reason before anything is sent.
|
||||
//
|
||||
// **The runtime uses no consumer of a log.** The bus grants it, per issued log, only the publish of an
|
||||
// entry, the stream's info and the direct get: an append is a publish, a read is a batch direct get
|
||||
// (NATS 2.11), the last entry of a key is a direct get by its subject.
|
||||
|
||||
// LogIssued is one log an assignment may reach (ADR 0297): its owner's alone, so there are no reads.
|
||||
type LogIssued struct {
|
||||
Name string `json:"name"`
|
||||
Stream string `json:"stream"`
|
||||
Subject string `json:"subject"`
|
||||
Writes bool `json:"writes,omitempty"`
|
||||
}
|
||||
|
||||
// LogEntry is one operation in a log: its sequence is its identity.
|
||||
type LogEntry struct {
|
||||
Sequence uint64 `json:"sequence"`
|
||||
Key string `json:"key"`
|
||||
Value json.RawMessage `json:"value"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// LogPage is one page of a read: Next is the sequence to continue from, 0 when nothing follows; Last
|
||||
// is the stream's last sequence.
|
||||
type LogPage struct {
|
||||
Entries []LogEntry `json:"entries"`
|
||||
Next uint64 `json:"next"`
|
||||
Last uint64 `json:"last"`
|
||||
}
|
||||
|
||||
// LogInfo is how full a log is.
|
||||
type LogInfo struct {
|
||||
Entries uint64 `json:"entries"`
|
||||
Bytes uint64 `json:"bytes"`
|
||||
MaxBytes uint64 `json:"max_bytes"`
|
||||
First uint64 `json:"first"`
|
||||
Last uint64 `json:"last"`
|
||||
}
|
||||
|
||||
const (
|
||||
// LogValueMax is the most one entry's value may hold (ADR 0297 §3); the stream's MaxMsgSize is 4 KiB
|
||||
// more, for the headers an append carries.
|
||||
LogValueMax = 256 * 1024
|
||||
// LogKeyMax is the longest key.
|
||||
LogKeyMax = 255
|
||||
// LogReadDefault and LogReadMax are a page's entries when none is asked, and at most.
|
||||
LogReadDefault = 100
|
||||
LogReadMax = 1000
|
||||
// LogPageBytes bounds the bytes of one page.
|
||||
LogPageBytes = 4 * 1024 * 1024
|
||||
)
|
||||
|
||||
// logToken is one name of a key: letters, digits, `_` and `-`.
|
||||
var logToken = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
|
||||
|
||||
func checkLogKey(key string) error {
|
||||
if key == "" || len(key) > LogKeyMax {
|
||||
return fmt.Errorf("%q is not a log key: one or more names of letters, digits, _ and -, joined by "+
|
||||
"dots, at most %d bytes (novox/hq ADR 0297)", key, LogKeyMax)
|
||||
}
|
||||
for _, t := range strings.Split(key, ".") {
|
||||
if !logToken.MatchString(t) {
|
||||
return fmt.Errorf("%q is not a log key: one or more names of letters, digits, _ and -, joined by "+
|
||||
"dots, at most %d bytes (novox/hq ADR 0297)", key, LogKeyMax)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// logFilter is a read's key pattern as a subject filter under the log's subject: `*` for one name, `>`
|
||||
// for the rest; empty is every key.
|
||||
func logFilter(l LogIssued, pattern string) (string, error) {
|
||||
if pattern == "" {
|
||||
return l.Subject + ".>", nil
|
||||
}
|
||||
parts := strings.Split(pattern, ".")
|
||||
for i, p := range parts {
|
||||
switch {
|
||||
case p == ">" && i == len(parts)-1, p == "*", logToken.MatchString(p):
|
||||
default:
|
||||
return "", fmt.Errorf("%q is not a key pattern: names, `*` for one and `>` for the rest, last "+
|
||||
"(novox/hq ADR 0297)", pattern)
|
||||
}
|
||||
}
|
||||
if len(pattern) > LogKeyMax {
|
||||
return "", fmt.Errorf("%q is longer than a key can be (%d bytes)", pattern, LogKeyMax)
|
||||
}
|
||||
return l.Subject + "." + pattern, nil
|
||||
}
|
||||
|
||||
// issuedLog is the log a module may reach by a name, or the reason it may not.
|
||||
func (c *Conn) issuedLog(module, name string) (LogIssued, error) {
|
||||
m := c.Membership(module)
|
||||
if m == nil {
|
||||
return LogIssued{}, fmt.Errorf("%s has no membership issued on %s yet, so no log of it is reachable "+
|
||||
"until the mesh issues one (novox/hq ADR 0297)", module, c.node)
|
||||
}
|
||||
var names []string
|
||||
for _, l := range m.Logs {
|
||||
if l.Name == name {
|
||||
return l, nil
|
||||
}
|
||||
names = append(names, l.Name)
|
||||
}
|
||||
sort.Strings(names)
|
||||
issued := "none"
|
||||
if len(names) > 0 {
|
||||
issued = strings.Join(names, ", ")
|
||||
}
|
||||
return LogIssued{}, fmt.Errorf("%s keeps no log called %q: a module declares the logs it keeps under `logs`, "+
|
||||
"and a log is its owner's alone; it was issued: %s (novox/hq ADR 0297)", module, name, issued)
|
||||
}
|
||||
|
||||
// notOnTheBus says a log is issued and its stream is not there yet.
|
||||
func notOnTheBus(l LogIssued) error {
|
||||
return fmt.Errorf("the log %q is issued and its stream %s is not on the bus yet: the controller creates "+
|
||||
"it from the catalogue on its next raise (novox/hq ADR 0297)", l.Name, l.Stream)
|
||||
}
|
||||
|
||||
// LogAppend writes one entry for a key, as the module, and answers its sequence. With expect, it is
|
||||
// written only when the key's last entry is still at that sequence (0: the key has none); otherwise it
|
||||
// is refused as a lost compare-and-set, the same error a lost state update answers (ADR 0259).
|
||||
func (c *Conn) LogAppend(module, name, key string, value json.RawMessage, expect *uint64) (uint64, error) {
|
||||
l, err := c.issuedLog(module, name)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if !l.Writes {
|
||||
return 0, fmt.Errorf("%s was issued the log %s without writing it (novox/hq ADR 0297)", module, name)
|
||||
}
|
||||
if err := checkLogKey(key); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if len(value) == 0 || !json.Valid(value) {
|
||||
return 0, fmt.Errorf("a log entry's value is JSON (novox/hq ADR 0297)")
|
||||
}
|
||||
if len(value) > LogValueMax {
|
||||
return 0, fmt.Errorf("a log entry's value is at most %d bytes (256 KiB), and this one is %d (novox/hq ADR 0297)",
|
||||
LogValueMax, len(value))
|
||||
}
|
||||
if field := credentialField(value); field != "" {
|
||||
return 0, fmt.Errorf("%s's %s.%s carries a field %q, which names a credential: no secret is kept in a log, "+
|
||||
"which keeps it for good. Name the secret and fetch it on request/reply (novox/hq ADR 0201 §13, ADR 0297)",
|
||||
module, name, key, field)
|
||||
}
|
||||
msg := nats.NewMsg(l.Subject + "." + key)
|
||||
msg.Data = value
|
||||
if expect != nil {
|
||||
msg.Header.Set(jetstream.ExpectedLastSubjSeqHeader, strconv.FormatUint(*expect, 10))
|
||||
}
|
||||
reply, err := c.nc.RequestMsg(msg, StateTimeout)
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
return 0, notOnTheBus(l)
|
||||
}
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
var ack struct {
|
||||
Seq uint64 `json:"seq"`
|
||||
Error *struct {
|
||||
Code int `json:"code"`
|
||||
ErrCode uint16 `json:"err_code"`
|
||||
Description string `json:"description"`
|
||||
} `json:"error"`
|
||||
}
|
||||
if err := json.Unmarshal(reply.Data, &ack); err != nil {
|
||||
return 0, fmt.Errorf("the bus answered the append to %s with what is not an acknowledgement: %q", name, reply.Data)
|
||||
}
|
||||
if ack.Error != nil {
|
||||
// The key's last entry is no longer the one expected: a lost compare-and-set, answered as a lost
|
||||
// state update is (10164 is the same refusal from a replicated stream).
|
||||
if code := jetstream.ErrorCode(ack.Error.ErrCode); code == jetstream.JSErrCodeStreamWrongLastSequence ||
|
||||
code == jetstream.JSErrCodeStreamWrongLastSequenceConstant {
|
||||
return 0, changed{name + "." + key}
|
||||
}
|
||||
return 0, fmt.Errorf("the bus refused the append to %s.%s: %s (%d)", name, key, ack.Error.Description, ack.Error.ErrCode)
|
||||
}
|
||||
if ack.Seq == 0 {
|
||||
return 0, fmt.Errorf("the bus answered the append to %s.%s with no sequence", name, key)
|
||||
}
|
||||
return ack.Seq, nil
|
||||
}
|
||||
|
||||
// LogRead is the entries from a sequence on, of the keys matching a pattern, at most limit of them and
|
||||
// about 4 MiB, by one batch direct get.
|
||||
func (c *Conn) LogRead(module, name string, from uint64, keys string, limit int) (*LogPage, error) {
|
||||
l, err := c.issuedLog(module, name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
switch {
|
||||
case limit == 0:
|
||||
limit = LogReadDefault
|
||||
case limit < 0 || limit > LogReadMax:
|
||||
return nil, fmt.Errorf("a read's limit is 1 to %d entries, %d when none is given, not %d (novox/hq ADR 0297)",
|
||||
LogReadMax, LogReadDefault, limit)
|
||||
}
|
||||
filter, err := logFilter(l, keys)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if from == 0 {
|
||||
from = 1
|
||||
}
|
||||
req, _ := json.Marshal(map[string]any{"seq": from, "next_by_subj": filter, "batch": limit, "max_bytes": LogPageBytes})
|
||||
inbox := c.nc.NewInbox()
|
||||
sub, err := c.nc.SubscribeSync(inbox)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() { _ = sub.Unsubscribe() }()
|
||||
if err := c.nc.PublishRequest("$JS.API.DIRECT.GET."+l.Stream, inbox, req); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
page := &LogPage{Entries: []LogEntry{}}
|
||||
deadline := time.Now().Add(StateTimeout)
|
||||
for done := false; !done; {
|
||||
msg, err := sub.NextMsg(time.Until(deadline))
|
||||
if err != nil {
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
return nil, notOnTheBus(l)
|
||||
}
|
||||
if errors.Is(err, nats.ErrTimeout) {
|
||||
return nil, fmt.Errorf("the read of %s did not end in %s", name, StateTimeout)
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
switch status := msg.Header.Get("Status"); status {
|
||||
case "":
|
||||
e, err := logEntryOf(l, msg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
page.Entries = append(page.Entries, *e)
|
||||
case "204": // the end of the batch: what is still pending after the last entry sent
|
||||
pending, _ := strconv.ParseUint(msg.Header.Get("Nats-Num-Pending"), 10, 64)
|
||||
if pending > 0 && len(page.Entries) > 0 {
|
||||
page.Next = page.Entries[len(page.Entries)-1].Sequence + 1
|
||||
}
|
||||
done = true
|
||||
case "404": // nothing from that sequence on
|
||||
done = true
|
||||
case "503":
|
||||
return nil, notOnTheBus(l)
|
||||
default:
|
||||
return nil, fmt.Errorf("the read of %s was refused by the bus: %s %s", name, status, msg.Header.Get("Description"))
|
||||
}
|
||||
}
|
||||
info, err := c.logInfo(l)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
page.Last = info.Last
|
||||
return page, nil
|
||||
}
|
||||
|
||||
// LogLast is a key's last entry, or nil when it has none.
|
||||
func (c *Conn) LogLast(module, name, key string) (*LogEntry, error) {
|
||||
l, err := c.issuedLog(module, name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := checkLogKey(key); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
reply, err := c.nc.Request("$JS.API.DIRECT.GET."+l.Stream+"."+l.Subject+"."+key, nil, StateTimeout)
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
return nil, notOnTheBus(l)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
switch status := reply.Header.Get("Status"); status {
|
||||
case "":
|
||||
return logEntryOf(l, reply)
|
||||
case "404":
|
||||
return nil, nil
|
||||
default:
|
||||
return nil, fmt.Errorf("the last entry of %s.%s could not be read: %s %s", name, key, status, reply.Header.Get("Description"))
|
||||
}
|
||||
}
|
||||
|
||||
// LogInfo is a log's entries, bytes, cap and first and last sequences.
|
||||
func (c *Conn) LogInfo(module, name string) (*LogInfo, error) {
|
||||
l, err := c.issuedLog(module, name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return c.logInfo(l)
|
||||
}
|
||||
|
||||
func (c *Conn) logInfo(l LogIssued) (*LogInfo, error) {
|
||||
reply, err := c.nc.Request("$JS.API.STREAM.INFO."+l.Stream, nil, StateTimeout)
|
||||
if errors.Is(err, nats.ErrNoResponders) {
|
||||
return nil, notOnTheBus(l)
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var info struct {
|
||||
Config struct {
|
||||
MaxBytes int64 `json:"max_bytes"`
|
||||
} `json:"config"`
|
||||
State struct {
|
||||
Msgs uint64 `json:"messages"`
|
||||
Bytes uint64 `json:"bytes"`
|
||||
FirstSeq uint64 `json:"first_seq"`
|
||||
LastSeq uint64 `json:"last_seq"`
|
||||
} `json:"state"`
|
||||
Error *struct {
|
||||
ErrCode uint16 `json:"err_code"`
|
||||
Description string `json:"description"`
|
||||
} `json:"error"`
|
||||
}
|
||||
if err := json.Unmarshal(reply.Data, &info); err != nil {
|
||||
return nil, fmt.Errorf("the bus answered the info of %s with what is not one: %q", l.Name, reply.Data)
|
||||
}
|
||||
if info.Error != nil {
|
||||
if info.Error.ErrCode == uint16(jetstream.JSErrCodeStreamNotFound) {
|
||||
return nil, notOnTheBus(l)
|
||||
}
|
||||
return nil, fmt.Errorf("the info of the log %s was refused: %s (%d)", l.Name, info.Error.Description, info.Error.ErrCode)
|
||||
}
|
||||
var maxBytes uint64
|
||||
if info.Config.MaxBytes > 0 {
|
||||
maxBytes = uint64(info.Config.MaxBytes)
|
||||
}
|
||||
return &LogInfo{Entries: info.State.Msgs, Bytes: info.State.Bytes, MaxBytes: maxBytes,
|
||||
First: info.State.FirstSeq, Last: info.State.LastSeq}, nil
|
||||
}
|
||||
|
||||
// logEntryOf is an entry a direct get answered: its key is its subject under the log's.
|
||||
func logEntryOf(l LogIssued, msg *nats.Msg) (*LogEntry, error) {
|
||||
subject := msg.Header.Get("Nats-Subject")
|
||||
key, ok := strings.CutPrefix(subject, l.Subject+".")
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("the bus answered an entry of %s on %q, which is not under %s", l.Name, subject, l.Subject)
|
||||
}
|
||||
seq, err := strconv.ParseUint(msg.Header.Get("Nats-Sequence"), 10, 64)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the bus answered an entry of %s with no sequence", l.Name)
|
||||
}
|
||||
at, err := time.Parse(time.RFC3339Nano, msg.Header.Get("Nats-Time-Stamp"))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the bus answered entry %d of %s with no time: %v", seq, l.Name, err)
|
||||
}
|
||||
return &LogEntry{Sequence: seq, Key: key, Value: valueOf(msg.Data), At: at.UTC()}, nil
|
||||
}
|
||||
@@ -37,6 +37,7 @@ type subscribing struct {
|
||||
func (b *subscribing) Publish(json.RawMessage) error { return nil }
|
||||
func (b *subscribing) Ask(json.RawMessage) (json.RawMessage, error) { return nil, nil }
|
||||
func (b *subscribing) State(string, json.RawMessage) (json.RawMessage, error) { return nil, nil }
|
||||
func (b *subscribing) Log(string, json.RawMessage) (json.RawMessage, error) { return nil, nil }
|
||||
func (b *subscribing) Watch(json.RawMessage, func(json.RawMessage) error) (func(), error) {
|
||||
return func() {}, nil
|
||||
}
|
||||
|
||||
@@ -77,6 +77,8 @@ type Bus interface {
|
||||
Subscribe(deliver func(envelope json.RawMessage) error) error
|
||||
State(verb string, params json.RawMessage) (json.RawMessage, error)
|
||||
Watch(params json.RawMessage, deliver func(change json.RawMessage) error) (stop func(), err error)
|
||||
// Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (novox/hq ADR 0297).
|
||||
Log(verb string, params json.RawMessage) (json.RawMessage, error)
|
||||
// Seat answers a bundle's seat traffic (novox/hq ADR 0259 §3): `publish`, `prove` and `record` at once;
|
||||
// `take` and `answer` bind once and hand each message or proof on through handOn, to whichever child
|
||||
// of the module asked last — a restarted child asks again as its code runs again.
|
||||
@@ -288,6 +290,11 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
|
||||
if answered, err = mesh.State(strings.TrimPrefix(m.Method, "mesh/state."), m.Params); err == nil {
|
||||
result = answered
|
||||
}
|
||||
case "mesh/log.append", "mesh/log.read", "mesh/log.last", "mesh/log.info":
|
||||
var answered json.RawMessage
|
||||
if answered, err = mesh.Log(strings.TrimPrefix(m.Method, "mesh/log."), m.Params); err == nil {
|
||||
result = answered
|
||||
}
|
||||
case "mesh/state.watch":
|
||||
// Each change is asked of this child, in order, naming the watch it is for by
|
||||
// the id the child asked it with — two watches of one state are two handlers;
|
||||
|
||||
@@ -231,3 +231,32 @@ func (m *Mesh) Bucket(t *testing.T, name string) nats.KeyValue {
|
||||
}
|
||||
return kv
|
||||
}
|
||||
|
||||
// Log makes a module's log afresh as the controller does from the catalogue (novox/hq ADR 0297): a file
|
||||
// stream named LOG_<module>_<name> on mesh.log.<module>.<name>.>, limits retention, no age, no cap per
|
||||
// key, the declared cap, 260 KiB a message (a 256 KiB value and its headers), refusing new entries when full, direct get, no delete or
|
||||
// purge. It answers the membership entry the controller issues for it.
|
||||
func (m *Mesh) Log(t *testing.T, module, name string, maxBytes int64) bus.LogIssued {
|
||||
t.Helper()
|
||||
stream := "LOG_" + module + "_" + name
|
||||
subject := "mesh.log." + module + "." + name
|
||||
_ = m.js.DeleteStream(stream)
|
||||
if _, err := m.js.AddStream(&nats.StreamConfig{Name: stream, Subjects: []string{subject + ".>"},
|
||||
Storage: nats.FileStorage, Retention: nats.LimitsPolicy, MaxAge: 0, MaxMsgsPerSubject: -1,
|
||||
MaxBytes: maxBytes, MaxMsgSize: 260 * 1024, Discard: nats.DiscardNew, AllowDirect: true,
|
||||
DenyDelete: true, DenyPurge: true, Replicas: 1,
|
||||
Description: module + "'s log " + name + ", kept as its record (novox/hq ADR 0297)"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return bus.LogIssued{Name: name, Stream: stream, Subject: subject, Writes: true}
|
||||
}
|
||||
|
||||
// Consumers is how many consumers a stream has.
|
||||
func (m *Mesh) Consumers(t *testing.T, stream string) int {
|
||||
t.Helper()
|
||||
info, err := m.js.StreamInfo(stream)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return info.State.Consumers
|
||||
}
|
||||
|
||||
@@ -0,0 +1,336 @@
|
||||
package runtime
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
||||
"github.com/novox/mesh-tools/node-tools/internal/launch"
|
||||
mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
|
||||
)
|
||||
|
||||
// logBuses is a module, issues, that keeps a log `changes` on a real bus, and a module, other, on the
|
||||
// same runtime that was issued no log — each reached as a bundle reaches it: mesh/log.<verb> with its
|
||||
// JSON, through its module's bus (novox/hq ADR 0297).
|
||||
func logBuses(t *testing.T, maxBytes int64) (issues, other launch.Bus, mesh *mt.Mesh) {
|
||||
t.Helper()
|
||||
mesh = mt.New(t)
|
||||
m := mt.MembershipOf("issues", "anchor", false, nil)
|
||||
m.Logs = []bus.LogIssued{mesh.Log(t, "issues", "changes", maxBytes)}
|
||||
mesh.Issue(t, m)
|
||||
mesh.Issue(t, mt.MembershipOf("other", "anchor", false, nil))
|
||||
conn := connect(t, "node-tools", "anchor")
|
||||
conn.Follow("issues")
|
||||
conn.Follow("other")
|
||||
mt.Until(t, func() error {
|
||||
if conn.Membership("issues") == nil || conn.Membership("other") == nil {
|
||||
return errors.New("not issued yet")
|
||||
}
|
||||
return nil
|
||||
})
|
||||
all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}}
|
||||
return all.forModule("issues"), all.forModule("other"), mesh
|
||||
}
|
||||
|
||||
// logAsk is a bundle's mesh/log.<verb>, with its params exactly as they go over the wire.
|
||||
func logAsk(b launch.Bus, verb, params string) (json.RawMessage, error) {
|
||||
return b.Log(verb, json.RawMessage(params))
|
||||
}
|
||||
|
||||
func appendOf(t *testing.T, b launch.Bus, params string) uint64 {
|
||||
t.Helper()
|
||||
raw, err := logAsk(b, "append", params)
|
||||
if err != nil {
|
||||
t.Fatalf("append %s: %v", params, err)
|
||||
}
|
||||
var r struct {
|
||||
Sequence *uint64 `json:"sequence"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &r); err != nil || r.Sequence == nil {
|
||||
t.Fatalf("append answered %s, not {sequence}", raw)
|
||||
}
|
||||
return *r.Sequence
|
||||
}
|
||||
|
||||
// wirePage is a read's answer with exactly the contract's field names.
|
||||
type wirePage struct {
|
||||
Entries []struct {
|
||||
Sequence uint64 `json:"sequence"`
|
||||
Key string `json:"key"`
|
||||
Value json.RawMessage `json:"value"`
|
||||
At string `json:"at"`
|
||||
} `json:"entries"`
|
||||
Next uint64 `json:"next"`
|
||||
Last uint64 `json:"last"`
|
||||
}
|
||||
|
||||
func readOf(t *testing.T, b launch.Bus, params string) wirePage {
|
||||
t.Helper()
|
||||
raw, err := logAsk(b, "read", params)
|
||||
if err != nil {
|
||||
t.Fatalf("read %s: %v", params, err)
|
||||
}
|
||||
var fields map[string]json.RawMessage
|
||||
_ = json.Unmarshal(raw, &fields)
|
||||
for _, f := range []string{"entries", "next", "last"} {
|
||||
if _, ok := fields[f]; !ok {
|
||||
t.Fatalf("read answered %s, without %q", raw, f)
|
||||
}
|
||||
}
|
||||
var p wirePage
|
||||
if err := json.Unmarshal(raw, &p); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
func infoOf(t *testing.T, b launch.Bus) map[string]uint64 {
|
||||
t.Helper()
|
||||
raw, err := logAsk(b, "info", `{"log":"changes"}`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var info map[string]uint64
|
||||
if err := json.Unmarshal(raw, &info); err != nil {
|
||||
t.Fatalf("info answered %s: %v", raw, err)
|
||||
}
|
||||
for _, f := range []string{"entries", "bytes", "max_bytes", "first", "last"} {
|
||||
if _, ok := info[f]; !ok {
|
||||
t.Fatalf("info answered %s, without %q", raw, f)
|
||||
}
|
||||
}
|
||||
return info
|
||||
}
|
||||
|
||||
// Each verb against a real bus: append answers the sequence, read pages from a sequence on, of a key
|
||||
// pattern, with the sequence to continue from (0 at the end) and the log's last, each entry's time in
|
||||
// RFC 3339; last is a key's last entry or null; info says entries, bytes, cap, first and last — and no
|
||||
// consumer is made for any of it.
|
||||
func TestALogAnswersEachVerbOnTheBus(t *testing.T) {
|
||||
b, _, mesh := logBuses(t, 1<<20)
|
||||
before := time.Now().Add(-time.Second)
|
||||
|
||||
var seqs []uint64
|
||||
for i, key := range []string{"12", "13", "12", "14.notes", "12"} {
|
||||
seqs = append(seqs, appendOf(t, b, fmt.Sprintf(`{"log":"changes","key":%q,"value":{"op":"move","n":%d}}`, key, i)))
|
||||
}
|
||||
for i := 1; i < len(seqs); i++ {
|
||||
if seqs[i] != seqs[i-1]+1 {
|
||||
t.Fatalf("sequences %v are not the log's order", seqs)
|
||||
}
|
||||
}
|
||||
|
||||
// Everything, no keys and no limit: the default page holds all five.
|
||||
all := readOf(t, b, `{"log":"changes","from":0}`)
|
||||
if len(all.Entries) != 5 || all.Next != 0 || all.Last != seqs[4] {
|
||||
t.Fatalf("a read of everything: %d entries, next %d, last %d", len(all.Entries), all.Next, all.Last)
|
||||
}
|
||||
for i, e := range all.Entries {
|
||||
at, err := time.Parse(time.RFC3339, e.At)
|
||||
if err != nil || at.Before(before) || at.After(time.Now().Add(time.Second)) {
|
||||
t.Errorf("entry %d's time %q is not RFC 3339 and now: %v", e.Sequence, e.At, err)
|
||||
}
|
||||
if e.Sequence != seqs[i] || string(e.Value) != fmt.Sprintf(`{"op":"move","n":%d}`, i) {
|
||||
t.Errorf("entry %d: %d %s %s", i, e.Sequence, e.Key, e.Value)
|
||||
}
|
||||
}
|
||||
if all.Entries[3].Key != "14.notes" {
|
||||
t.Errorf("a key of two names read back as %q", all.Entries[3].Key)
|
||||
}
|
||||
|
||||
// One key's entries, a page of two at a time.
|
||||
first := readOf(t, b, `{"log":"changes","from":1,"keys":"12","limit":2}`)
|
||||
if len(first.Entries) != 2 || first.Entries[0].Sequence != seqs[0] || first.Entries[1].Sequence != seqs[2] ||
|
||||
first.Next != seqs[2]+1 {
|
||||
t.Fatalf("the first page of key 12: %+v", first)
|
||||
}
|
||||
second := readOf(t, b, fmt.Sprintf(`{"log":"changes","from":%d,"keys":"12","limit":2}`, first.Next))
|
||||
if len(second.Entries) != 1 || second.Entries[0].Sequence != seqs[4] || second.Next != 0 {
|
||||
t.Fatalf("the second page of key 12: %+v", second)
|
||||
}
|
||||
// A pattern: `*` one name, `>` the rest.
|
||||
if p := readOf(t, b, `{"log":"changes","from":1,"keys":"14.*"}`); len(p.Entries) != 1 || p.Entries[0].Key != "14.notes" {
|
||||
t.Fatalf("14.*: %+v", p)
|
||||
}
|
||||
if p := readOf(t, b, `{"log":"changes","from":1,"keys":"14.>"}`); len(p.Entries) != 1 {
|
||||
t.Fatalf("14.>: %+v", p)
|
||||
}
|
||||
// Past the end: nothing, nothing follows.
|
||||
if p := readOf(t, b, fmt.Sprintf(`{"log":"changes","from":%d}`, seqs[4]+1)); len(p.Entries) != 0 || p.Next != 0 || p.Last != seqs[4] {
|
||||
t.Fatalf("a read past the end: %+v", p)
|
||||
}
|
||||
if p := readOf(t, b, `{"log":"changes","from":1,"keys":"99"}`); len(p.Entries) != 0 || p.Next != 0 {
|
||||
t.Fatalf("a key with no entries: %+v", p)
|
||||
}
|
||||
|
||||
raw, err := logAsk(b, "last", `{"log":"changes","key":"12"}`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var last map[string]json.RawMessage
|
||||
_ = json.Unmarshal(raw, &last)
|
||||
if string(last["sequence"]) != fmt.Sprint(seqs[4]) || string(last["key"]) != `"12"` ||
|
||||
string(last["value"]) != `{"op":"move","n":4}` || len(last["at"]) == 0 {
|
||||
t.Fatalf("the last entry of 12: %s", raw)
|
||||
}
|
||||
if raw, err := logAsk(b, "last", `{"log":"changes","key":"99"}`); err != nil || string(raw) != "null" {
|
||||
t.Fatalf("the last entry of a key with none: %s %v", raw, err)
|
||||
}
|
||||
|
||||
info := infoOf(t, b)
|
||||
if info["entries"] != 5 || info["first"] != seqs[0] || info["last"] != seqs[4] || info["max_bytes"] != 1<<20 ||
|
||||
info["bytes"] == 0 {
|
||||
t.Fatalf("info: %v", info)
|
||||
}
|
||||
if n := mesh.Consumers(t, "LOG_issues_changes"); n != 0 {
|
||||
t.Fatalf("the runtime made %d consumers of the log; it uses none", n)
|
||||
}
|
||||
}
|
||||
|
||||
// A page holds about 4 MiB at most, whatever its limit, and says where to continue.
|
||||
func TestALogPageIsBoundedByItsBytes(t *testing.T) {
|
||||
b, _, _ := logBuses(t, 64<<20)
|
||||
big := `"` + strings.Repeat("x", 250*1024) + `"`
|
||||
for i := 0; i < 20; i++ {
|
||||
appendOf(t, b, fmt.Sprintf(`{"log":"changes","key":"k%d","value":%s}`, i, big))
|
||||
}
|
||||
p := readOf(t, b, `{"log":"changes","from":1,"limit":1000}`)
|
||||
if len(p.Entries) == 0 || len(p.Entries) >= 20 || p.Next != p.Entries[len(p.Entries)-1].Sequence+1 {
|
||||
t.Fatalf("a page of 20 entries of 250 KiB: %d entries, next %d", len(p.Entries), p.Next)
|
||||
}
|
||||
got := len(p.Entries)
|
||||
for p.Next != 0 {
|
||||
p = readOf(t, b, fmt.Sprintf(`{"log":"changes","from":%d,"limit":1000}`, p.Next))
|
||||
got += len(p.Entries)
|
||||
}
|
||||
if got != 20 {
|
||||
t.Fatalf("paging read %d of 20 entries", got)
|
||||
}
|
||||
}
|
||||
|
||||
// An append with `expect` is written only when the key's last entry is the one expected — 0 when it
|
||||
// has none — and is otherwise refused with the error a lost mesh/state.update answers: the same code,
|
||||
// the same error, and nothing written. An append without `expect` is not checked.
|
||||
func TestAnAppendAtAStaleExpectIsRefusedAsALostUpdate(t *testing.T) {
|
||||
b, _, _ := logBuses(t, 1<<20)
|
||||
isLost := func(err error) bool {
|
||||
var coded interface{ ErrorCode() int }
|
||||
return errors.As(err, &coded) && coded.ErrorCode() == launch.CodeChanged && errors.Is(err, bus.ErrStateChanged)
|
||||
}
|
||||
|
||||
first := appendOf(t, b, `{"log":"changes","key":"7","value":{"op":"open"},"expect":0}`)
|
||||
if _, err := logAsk(b, "append", `{"log":"changes","key":"7","value":{"op":"open"},"expect":0}`); !isLost(err) {
|
||||
t.Fatalf("a second open of one key at expect 0: %v", err)
|
||||
}
|
||||
second := appendOf(t, b, fmt.Sprintf(`{"log":"changes","key":"7","value":{"op":"move"},"expect":%d}`, first))
|
||||
if _, err := logAsk(b, "append", fmt.Sprintf(`{"log":"changes","key":"7","value":{"op":"note"},"expect":%d}`, first)); !isLost(err) {
|
||||
t.Fatalf("an append at a stale expect: %v", err)
|
||||
}
|
||||
if _, err := logAsk(b, "append", fmt.Sprintf(`{"log":"changes","key":"7","value":{"op":"note"},"expect":%d}`, second+100)); !isLost(err) {
|
||||
t.Fatalf("an append at an expect not yet written: %v", err)
|
||||
}
|
||||
// Another key's entries do not make this key's expect stale.
|
||||
appendOf(t, b, `{"log":"changes","key":"8","value":{"op":"open"}}`)
|
||||
third := appendOf(t, b, fmt.Sprintf(`{"log":"changes","key":"7","value":{"op":"note"},"expect":%d}`, second))
|
||||
// No expect: no check.
|
||||
appendOf(t, b, `{"log":"changes","key":"7","value":{"op":"set"}}`)
|
||||
|
||||
if info := infoOf(t, b); info["entries"] != 5 {
|
||||
t.Fatalf("refused appends wrote: %d entries, not 5", info["entries"])
|
||||
}
|
||||
p := readOf(t, b, `{"log":"changes","from":1,"keys":"7"}`)
|
||||
if len(p.Entries) != 4 || p.Entries[2].Sequence != third {
|
||||
t.Fatalf("key 7 holds %+v", p.Entries)
|
||||
}
|
||||
|
||||
// The refusal is the very error a lost state update is: its code reaches a bundle as CodeChanged.
|
||||
_, err := logAsk(b, "append", `{"log":"changes","key":"7","value":{},"expect":0}`)
|
||||
var coded interface{ ErrorCode() int }
|
||||
if !errors.As(err, &coded) || coded.ErrorCode() != -32010 {
|
||||
t.Fatalf("the refusal carries no lost-update code: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// A log is its owner's alone: another module on the same runtime is refused it with the reason, and so
|
||||
// is a log of a name the owner was not issued — before anything is sent.
|
||||
func TestALogNotIssuedIsRefusedWithTheReason(t *testing.T) {
|
||||
issues, other, _ := logBuses(t, 1<<20)
|
||||
for verb, params := range map[string]string{
|
||||
"append": `{"log":"changes","key":"1","value":{}}`,
|
||||
"read": `{"log":"changes","from":1}`,
|
||||
"last": `{"log":"changes","key":"1"}`,
|
||||
"info": `{"log":"changes"}`,
|
||||
} {
|
||||
_, err := logAsk(other, verb, params)
|
||||
if err == nil || !strings.Contains(err.Error(), `other keeps no log called "changes"`) ||
|
||||
!strings.Contains(err.Error(), "issued: none") {
|
||||
t.Errorf("%s of a log not issued: %v", verb, err)
|
||||
}
|
||||
}
|
||||
_, err := logAsk(issues, "append", `{"log":"moves","key":"1","value":{}}`)
|
||||
if err == nil || !strings.Contains(err.Error(), `issues keeps no log called "moves"`) ||
|
||||
!strings.Contains(err.Error(), "issued: changes") {
|
||||
t.Errorf("an append to a log of a name not issued: %v", err)
|
||||
}
|
||||
if info := infoOf(t, issues); info["entries"] != 0 {
|
||||
t.Fatalf("a refused append wrote: %v", info)
|
||||
}
|
||||
}
|
||||
|
||||
// What the bus cannot hold, or the contract does not allow, is refused with the reason and not sent:
|
||||
// a value that is not JSON or over 256 KiB, a key that is not names joined by dots or over 255 bytes, a
|
||||
// key pattern with `>` not last, a limit over 1000.
|
||||
func TestWhatALogCannotHoldIsRefusedWithTheReason(t *testing.T) {
|
||||
b, _, _ := logBuses(t, 1<<20)
|
||||
over := `"` + strings.Repeat("x", 256*1024) + `"`
|
||||
for params, reason := range map[string]string{
|
||||
`{"log":"changes","key":"1","value":nope}`: "names no log",
|
||||
`{"log":"changes","key":"1"}`: "is JSON",
|
||||
`{"log":"changes","key":"1","value":` + over + `}`: "at most 262144 bytes",
|
||||
`{"log":"changes","key":"a b","value":1}`: "is not a log key",
|
||||
`{"log":"changes","key":"a..b","value":1}`: "is not a log key",
|
||||
`{"log":"changes","key":"a.*","value":1}`: "is not a log key",
|
||||
`{"log":"changes","key":"` + strings.Repeat("k", 256) + `","value":1}`: "is not a log key",
|
||||
`{"key":"1","value":1}`: "names no log",
|
||||
} {
|
||||
if _, err := logAsk(b, "append", params); err == nil || !strings.Contains(err.Error(), reason) {
|
||||
t.Errorf("append %.80s: %v, want %q", params, err, reason)
|
||||
}
|
||||
}
|
||||
for params, reason := range map[string]string{
|
||||
`{"log":"changes","from":1,"keys":"a.>.b"}`: "is not a key pattern",
|
||||
`{"log":"changes","from":1,"limit":1001}`: "1 to 1000",
|
||||
`{"log":"changes","from":1,"limit":-1}`: "1 to 1000",
|
||||
} {
|
||||
if _, err := logAsk(b, "read", params); err == nil || !strings.Contains(err.Error(), reason) {
|
||||
t.Errorf("read %s: %v, want %q", params, err, reason)
|
||||
}
|
||||
}
|
||||
if info := infoOf(t, b); info["entries"] != 0 {
|
||||
t.Fatalf("a refused append wrote: %v", info)
|
||||
}
|
||||
// The largest value allowed is taken, also with the header an expect carries.
|
||||
largest := `"` + strings.Repeat("x", 256*1024-2) + `"`
|
||||
appendOf(t, b, `{"log":"changes","key":"1","value":`+largest+`,"expect":0}`)
|
||||
appendOf(t, b, `{"log":"changes","key":"2","value":`+largest+`}`)
|
||||
}
|
||||
|
||||
// A value naming a credential anywhere in it is refused, as mesh/state.put refuses one (novox/hq ADR
|
||||
// 0201 §13): a log keeps what it is given for good. A value that only counts tokens is taken.
|
||||
func TestALogRefusesAValueNamingACredential(t *testing.T) {
|
||||
b, _, _ := logBuses(t, 1<<20)
|
||||
for _, value := range []string{`{"op":"set","api_key":"s"}`, `{"op":"note","with":{"headers":{"Authorization":"Bearer s"}}}`} {
|
||||
_, err := logAsk(b, "append", `{"log":"changes","key":"1","value":`+value+`}`)
|
||||
if err == nil || !strings.Contains(err.Error(), "names a credential") {
|
||||
t.Errorf("append of %s: %v", value, err)
|
||||
}
|
||||
}
|
||||
if info := infoOf(t, b); info["entries"] != 0 {
|
||||
t.Fatalf("a refused append wrote: %v", info)
|
||||
}
|
||||
appendOf(t, b, `{"log":"changes","key":"1","value":{"op":"note","maxTokens":4096}}`)
|
||||
}
|
||||
@@ -670,6 +670,61 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage,
|
||||
return json.Marshal(answer)
|
||||
}
|
||||
|
||||
// logAsked is what a bundle names when it reaches its log (novox/hq ADR 0297).
|
||||
type logAsked struct {
|
||||
Log string `json:"log"`
|
||||
Key string `json:"key"`
|
||||
Value json.RawMessage `json:"value"`
|
||||
Expect *uint64 `json:"expect"`
|
||||
From uint64 `json:"from"`
|
||||
Keys string `json:"keys"`
|
||||
Limit int `json:"limit"`
|
||||
}
|
||||
|
||||
// Log answers a bundle's `append`, `read`, `last` and `info` on its module's log (ADR 0297). The runtime
|
||||
// refuses, with the reason, a log the module was not issued; an append whose `expect` is stale is
|
||||
// refused with the code of a lost state update.
|
||||
func (b *moduleBus) Log(verb string, params json.RawMessage) (json.RawMessage, error) {
|
||||
var asked logAsked
|
||||
if err := json.Unmarshal(params, &asked); err != nil || asked.Log == "" {
|
||||
return nil, fmt.Errorf("mesh/log.%s names no log: {log, ...}", verb)
|
||||
}
|
||||
conn := b.all.conn
|
||||
var answer any
|
||||
switch verb {
|
||||
case "append":
|
||||
seq, err := conn.LogAppend(b.module, asked.Log, asked.Key, asked.Value, asked.Expect)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
answer = map[string]any{"sequence": seq}
|
||||
case "read":
|
||||
page, err := conn.LogRead(b.module, asked.Log, asked.From, asked.Keys, asked.Limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
answer = page
|
||||
case "last":
|
||||
entry, err := conn.LogLast(b.module, asked.Log, asked.Key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if entry == nil {
|
||||
return json.RawMessage("null"), nil
|
||||
}
|
||||
answer = entry
|
||||
case "info":
|
||||
info, err := conn.LogInfo(b.module, asked.Log)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
answer = info
|
||||
default:
|
||||
return nil, fmt.Errorf("the runtime answers no mesh/log.%s", verb)
|
||||
}
|
||||
return json.Marshal(answer)
|
||||
}
|
||||
|
||||
// Watch hands a bundle its module's state as it is and as it changes (ADR 0201): `{state, key}`, the
|
||||
// key a pattern with `*` and `**`, empty for every key.
|
||||
func (b *moduleBus) Watch(params json.RawMessage, deliver func(json.RawMessage) error) (func(), error) {
|
||||
|
||||
Reference in New Issue
Block a user