From d535dd53225808bebfcb9b0f7fe7eb6849d56428 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 11 Oct 2026 19:18:46 +0200 Subject: [PATCH] Serve a module's log verbs so its record outlives a bucket's cap and the events' week (ADR 0297, issues 501, 502) The runtime answers mesh/log.append, read, last and info for a log issued in the module's membership, by publish, batch direct get and stream info alone. A stale expect is refused as a lost state update, so a module tells it by the same code. --- merge-check.sh | 5 +- node-tools/internal/bus/bus.go | 2 + node-tools/internal/bus/log.go | 361 ++++++++++++++++++ .../internal/launch/event_answer_test.go | 1 + node-tools/internal/launch/launch.go | 7 + node-tools/internal/meshtest/meshtest.go | 29 ++ node-tools/internal/runtime/log_test.go | 318 +++++++++++++++ node-tools/internal/runtime/runtime.go | 55 +++ 8 files changed, 776 insertions(+), 2 deletions(-) create mode 100644 node-tools/internal/bus/log.go create mode 100644 node-tools/internal/runtime/log_test.go diff --git a/merge-check.sh b/merge-check.sh index 8b3100c..697ff85 100644 --- a/merge-check.sh +++ b/merge-check.sh @@ -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|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 diff --git a/node-tools/internal/bus/bus.go b/node-tools/internal/bus/bus.go index 896b37e..ee80667 100644 --- a/node-tools/internal/bus/bus.go +++ b/node-tools/internal/bus/bus.go @@ -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"` diff --git a/node-tools/internal/bus/log.go b/node-tools/internal/bus/log.go new file mode 100644 index 0000000..5cd1062 --- /dev/null +++ b/node-tools/internal/bus/log.go @@ -0,0 +1,361 @@ +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 own MaxMsgSize. + 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)) + } + 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 +} diff --git a/node-tools/internal/launch/event_answer_test.go b/node-tools/internal/launch/event_answer_test.go index 2046ec0..e731154 100644 --- a/node-tools/internal/launch/event_answer_test.go +++ b/node-tools/internal/launch/event_answer_test.go @@ -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 } diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index b07cc3c..23b7c41 100644 --- a/node-tools/internal/launch/launch.go +++ b/node-tools/internal/launch/launch.go @@ -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; diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index 67b1bc9..b69195a 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -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__ on mesh.log...>, limits retention, no age, no cap per +// key, the declared cap, 256 KiB an entry, 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: 256 * 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 +} diff --git a/node-tools/internal/runtime/log_test.go b/node-tools/internal/runtime/log_test.go new file mode 100644 index 0000000..45ec9eb --- /dev/null +++ b/node-tools/internal/runtime/log_test.go @@ -0,0 +1,318 @@ +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. 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., 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. + appendOf(t, b, `{"log":"changes","key":"1","value":"`+strings.Repeat("x", 256*1024-2)+`"}`) +} diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 2c106fa..be9a91d 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -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) {