Apply a ledger operation: check it against its type, append it with the last sequence expected, derive the state at the revision read, write the snapshot (issue 1966 T2)

2026-11-03 13:40:45 +01:00
parent d6c516ab44
commit ad22c5e9fe
5 changed files with 1042 additions and 0 deletions
@@ -0,0 +1,229 @@
+package bus
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strconv"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/membership"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"github.com/novox/mesh-tools/node-tools/internal/ledger"
+)
+
+// A ledger on the bus (novox/hq ADR 0597, design 25 § 12.4; issue 1966): its history a stream with a subject per
+// instance under the binding's subject, its state a key-value bucket, its snapshots a subject per instance under the
+// binding's snapshot subject, the latest only (the stream that holds them keeps one per subject, as the bus module
+// creates it). **Only the module's ledger connection publishes on its history and its snapshots** (ADR 0597 §9): what
+// is read, and the state, go on the runner's own.
+
+// LedgerStores are the stores of one ledger binding: reads and the state on c, the history's appends and the snapshots'
+// writes on writer, the module's ledger connection.
+func (c *Conn) LedgerStores(b membership.BusBinding, writer *Conn) (ledger.History, ledger.State, ledger.Snapshots) {
+	return &busHistory{read: c, write: writer, stream: b.History, prefix: prefixOf(b.Subject)},
+		&busState{c: c, bucket: b.State},
+		&busSnapshots{read: c, write: writer, prefix: prefixOf(b.Snapshot)}
+}
+
+// prefixOf is a binding's subject as the prefix an instance's key is added to: written with or without its `.>`.
+func prefixOf(subject string) string {
+	return strings.TrimSuffix(strings.TrimSuffix(subject, ">"), ".")
+}
+
+// publishExpecting publishes body on subject on the connection, written only when the subject's last message is still
+// at expect (0: it has none); ledger.ErrRaced when it is not.
+func publishExpecting(c *Conn, subject string, body []byte, expect uint64) (uint64, error) {
+	msg := nats.NewMsg(subject)
+	msg.Data = body
+	msg.Header.Set(jetstream.ExpectedLastSubjSeqHeader, strconv.FormatUint(expect, 10))
+	reply, err := c.nc.RequestMsg(msg, StateTimeout)
+	if errors.Is(err, nats.ErrNoResponders) {
+		return 0, fmt.Errorf("no stream on the bus keeps %s", subject)
+	}
+	if err != nil {
+		return 0, err
+	}
+	var ack struct {
+		Seq   uint64 `json:"seq"`
+		Error *struct {
+			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 write to %s with what is not an acknowledgement: %q", subject, reply.Data)
+	}
+	if ack.Error != nil {
+		if code := jetstream.ErrorCode(ack.Error.ErrCode); code == jetstream.JSErrCodeStreamWrongLastSequence ||
+			code == jetstream.JSErrCodeStreamWrongLastSequenceConstant {
+			return 0, ledger.ErrRaced
+		}
+		return 0, fmt.Errorf("the bus refused the write to %s: %s (%d)", subject, ack.Error.Description, ack.Error.ErrCode)
+	}
+	return ack.Seq, nil
+}
+
+// lastOn is the last message on a subject of the stream that keeps it: its bytes and sequence, nil and 0 for none.
+func lastOn(c *Conn, stream, subject string) ([]byte, uint64, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	js, err := c.jetStream()
+	if err != nil {
+		return nil, 0, err
+	}
+	if stream == "" {
+		if stream, err = js.StreamNameBySubject(ctx, subject); err != nil {
+			return nil, 0, fmt.Errorf("no stream on the bus keeps %s: %w", subject, err)
+		}
+	}
+	s, err := js.Stream(ctx, stream)
+	if err != nil {
+		return nil, 0, fmt.Errorf("the stream %s: %w", stream, err)
+	}
+	m, err := s.GetLastMsgForSubject(ctx, subject)
+	if errors.Is(err, jetstream.ErrMsgNotFound) {
+		return nil, 0, nil
+	}
+	if err != nil {
+		return nil, 0, err
+	}
+	return m.Data, m.Sequence, nil
+}
+
+type busHistory struct {
+	read, write    *Conn
+	stream, prefix string
+}
+
+func (h *busHistory) Last(key string) (uint64, error) {
+	_, seq, err := lastOn(h.read, h.stream, h.prefix+"."+key)
+	return seq, err
+}
+
+func (h *busHistory) Append(key string, op []byte, expect uint64) (uint64, error) {
+	if h.write == nil {
+		return 0, errors.New("the ledger's history is published only on its module's ledger connection, and none is held")
+	}
+	return publishExpecting(h.write, h.prefix+"."+key, op, expect)
+}
+
+func (h *busHistory) Read(keys string, from uint64, limit int) ([]ledger.Kept, error) {
+	var out []ledger.Kept
+	for {
+		want := StreamReadMost
+		if limit > 0 && limit-len(out) < want {
+			want = limit - len(out)
+		}
+		page, err := h.read.readStream(h.stream, h.prefix+"."+keys, from, want)
+		if err != nil {
+			return out, err
+		}
+		for _, m := range page.Messages {
+			out = append(out, ledger.Kept{Sequence: m.Sequence, Data: m.Body})
+		}
+		if page.Next == 0 || (limit > 0 && len(out) >= limit) {
+			return out, nil
+		}
+		from = page.Next
+	}
+}
+
+type busState struct {
+	c      *Conn
+	bucket string
+}
+
+func (s *busState) kv(ctx context.Context) (jetstream.KeyValue, error) {
+	js, err := s.c.jetStream()
+	if err != nil {
+		return nil, err
+	}
+	kv, err := js.KeyValue(ctx, s.bucket)
+	if errors.Is(err, jetstream.ErrBucketNotFound) {
+		return nil, fmt.Errorf("the ledger's state %s is not on the bus", s.bucket)
+	}
+	return kv, err
+}
+
+func (s *busState) Get(key string) ([]byte, uint64, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	kv, err := s.kv(ctx)
+	if err != nil {
+		return nil, 0, err
+	}
+	e, err := kv.Get(ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return nil, 0, nil
+	}
+	if err != nil {
+		return nil, 0, err
+	}
+	return e.Value(), e.Revision(), nil
+}
+
+// isFull is a write a bucket or stream refused because it holds as much as it may.
+func isFull(err error) bool {
+	msg := err.Error()
+	return strings.Contains(msg, "maximum bytes exceeded") || strings.Contains(msg, "maximum messages exceeded") ||
+		strings.Contains(msg, "insufficient resources")
+}
+
+func (s *busState) Write(key string, value []byte, revision uint64) (uint64, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	kv, err := s.kv(ctx)
+	if err != nil {
+		return 0, err
+	}
+	var rev uint64
+	if revision == 0 {
+		rev, err = kv.Create(ctx, key, value)
+	} else {
+		rev, err = kv.Update(ctx, key, value, revision)
+	}
+	switch {
+	case err == nil:
+		return rev, nil
+	case errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) || isWrongSequence(err):
+		return 0, ledger.ErrChanged
+	case isFull(err):
+		return 0, fmt.Errorf("%w: %v", ledger.ErrFull, err)
+	}
+	return 0, err
+}
+
+func (s *busState) Keys() ([]string, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	kv, err := s.kv(ctx)
+	if err != nil {
+		return nil, err
+	}
+	keys, err := kv.Keys(ctx)
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil, nil
+	}
+	return keys, err
+}
+
+type busSnapshots struct {
+	read, write *Conn
+	prefix      string
+}
+
+func (s *busSnapshots) Get(key string) ([]byte, uint64, error) {
+	return lastOn(s.read, "", s.prefix+"."+key)
+}
+
+func (s *busSnapshots) Put(key string, value []byte, expect uint64) error {
+	if s.write == nil {
+		return errors.New("a ledger's snapshot is published only on its module's ledger connection, and none is held")
+	}
+	_, err := publishExpecting(s.write, s.prefix+"."+key, value, expect)
+	return err
+}