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.

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"`

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
+}