Merge pull request 'node-tools: serve mesh/state.history (ADR 0307, issue 502)' (#29) from adr-0307-state-history into main

merge-check.sh

@@ -36,8 +36,8 @@ if ! command -v node >/dev/null 2>&1 || [ ! -d node_modules/@novox/mesh-sdk ]; t
 	# 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), 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)$'
+	# (novox/hq ADR 0297), and a key's history (novox/hq ADR 0307).
+	runtimeless='^(TestALogAnswersEachVerbOnTheBus|TestALogPageIsBoundedByItsBytes|TestAnAppendAtAStaleExpectIsRefusedAsALostUpdate|TestALogNotIssuedIsRefusedWithTheReason|TestWhatALogCannotHoldIsRefusedWithTheReason|TestALogRefusesAValueNamingACredential|TestAKeysHistoryIsEveryValueKeptOldestFirst|TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState|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

node-tools/internal/bus/state.go

@@ -129,6 +129,50 @@ func (c *Conn) StateGet(module, name, key string) (*StateEntry, error) {
 	return &StateEntry{Key: e.Key(), Value: valueOf(e.Value()), Revision: e.Revision()}, nil
 }
 
+// StateKept is one value a bucket still keeps for a key: a put, or a delete (a purge reads as one).
+type StateKept struct {
+	Key      string          `json:"key"`
+	Value    json.RawMessage `json:"value,omitempty"`
+	Revision uint64          `json:"revision"`
+	Op       string          `json:"op"`
+	At       time.Time       `json:"at"`
+}
+
+// StateHistory is every value the bucket still keeps for one key, oldest first, or none (novox/hq ADR
+// 0307). It is a read like StateGet: served to the owner and to every reader, and writes nothing.
+func (c *Conn) StateHistory(module, name, key string) ([]StateKept, error) {
+	s, err := c.issuedState(module, name)
+	if err != nil {
+		return nil, err
+	}
+	if err := checkKey(key); err != nil {
+		return nil, err
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
+	defer cancel()
+	kv, err := c.bucket(ctx, s)
+	if err != nil {
+		return nil, err
+	}
+	entries, err := kv.History(ctx, key)
+	kept := []StateKept{}
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return kept, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+	for _, e := range entries {
+		k := StateKept{Key: e.Key(), Revision: e.Revision(), Op: "delete", At: e.Created().UTC()}
+		if e.Operation() == jetstream.KeyValuePut {
+			k.Op = "put"
+			k.Value = valueOf(e.Value())
+		}
+		kept = append(kept, k)
+	}
+	return kept, nil
+}
+
 // StatePut writes one key, as the module, where the module keeps the state. It answers the revision.
 func (c *Conn) StatePut(module, name, key string, value json.RawMessage) (uint64, error) {
 	s, err := c.writable(module, name)

node-tools/internal/launch/launch.go

@@ -285,7 +285,7 @@ func Start(module, entry string, env []string, mesh Bus, logf func(string, ...an
 								result = answered
 							}
 						case "mesh/state.get", "mesh/state.put", "mesh/state.delete", "mesh/state.keys",
-							"mesh/state.create", "mesh/state.update":
+							"mesh/state.create", "mesh/state.update", "mesh/state.history":
 							var answered json.RawMessage
 							if answered, err = mesh.State(strings.TrimPrefix(m.Method, "mesh/state."), m.Params); err == nil {
 								result = answered

node-tools/internal/meshtest/meshtest.go

@@ -220,6 +220,17 @@ func (m *Mesh) Pending(t *testing.T, node, module string) (ackPending, notDelive
 	return uint64(info.NumAckPending), info.NumPending
 }
 
+// BucketKeeping makes a module's state afresh as Bucket does, keeping up to history values a key.
+func (m *Mesh) BucketKeeping(t *testing.T, name string, history uint8) nats.KeyValue {
+	t.Helper()
+	_ = m.js.DeleteKeyValue(name)
+	kv, err := m.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: name, History: history, MaxValueSize: 256 * 1024})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return kv
+}
+
 // Bucket makes a module's state afresh as the controller does from the catalogue (novox/hq ADR 0201),
 // and answers it for writing what is there before a bundle starts.
 func (m *Mesh) Bucket(t *testing.T, name string) nats.KeyValue {

node-tools/internal/runtime/runtime.go

@@ -614,8 +614,8 @@ type stateAsked struct {
 	Revision uint64          `json:"revision"`
 }
 
-// State answers a bundle's `get`, `put`, `delete` and `keys` on its module's state (ADR 0201). The
-// runtime refuses, with the reason, a state the module was not issued and a write to one it only reads.
+// State answers a bundle's `get`, `put`, `delete`, `keys`, `create`, `update` and `history` on its
+// module's state (ADR 0201, ADR 0259, ADR 0307). The runtime refuses, with the reason, a state the module was not issued and a write to one it only reads.
 func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, error) {
 	var asked stateAsked
 	if err := json.Unmarshal(params, &asked); err != nil || asked.State == "" {
@@ -633,6 +633,13 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage,
 			return json.RawMessage("null"), nil
 		}
 		answer = entry
+	case "history":
+		// Every value the bucket still keeps for the key, oldest first (novox/hq ADR 0307).
+		kept, err := conn.StateHistory(b.module, asked.State, asked.Key)
+		if err != nil {
+			return nil, err
+		}
+		answer = kept
 	case "put":
 		revision, err := conn.StatePut(b.module, asked.State, asked.Key, asked.Value)
 		if err != nil {

node-tools/internal/runtime/state_history_test.go

@@ -0,0 +1,152 @@
+package runtime
+
+import (
+	"encoding/json"
+	"errors"
+	"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"
+)
+
+// historyBuses are three modules on one runtime, each reached as a bundle reaches it (novox/hq ADR
+// 0307): issues keeps a state `issues` in a bucket keeping 8 values a key, triage reads it as
+// `issues.issues`, and other was issued no state.
+func historyBuses(t *testing.T) (owner, reader, other launch.Bus) {
+	t.Helper()
+	mesh := mt.New(t)
+	mesh.BucketKeeping(t, "issues_issues", 8)
+	m := mt.MembershipOf("issues", "anchor", false, nil)
+	m.State = []bus.StateIssued{{Name: "issues", Bucket: "issues_issues", Writes: true}}
+	mesh.Issue(t, m)
+	r := mt.MembershipOf("triage", "anchor", false, nil)
+	r.State = []bus.StateIssued{{Name: "issues.issues", Bucket: "issues_issues"}}
+	mesh.Issue(t, r)
+	mesh.Issue(t, mt.MembershipOf("other", "anchor", false, nil))
+	conn := connect(t, "node-tools", "anchor")
+	for _, module := range []string{"issues", "triage", "other"} {
+		conn.Follow(module)
+	}
+	mt.Until(t, func() error {
+		for _, module := range []string{"issues", "triage", "other"} {
+			if conn.Membership(module) == nil {
+				return errors.New(module + " not issued yet")
+			}
+		}
+		return nil
+	})
+	all := &consumers{conn: conn, logf: t.Logf, of: map[string]*moduleEvents{}}
+	return all.forModule("issues"), all.forModule("triage"), all.forModule("other")
+}
+
+// kept is one value of a history answer, by the field names on the wire.
+type kept struct {
+	Key      string          `json:"key"`
+	Value    json.RawMessage `json:"value"`
+	Revision uint64          `json:"revision"`
+	Op       string          `json:"op"`
+	At       string          `json:"at"`
+}
+
+func historyOf(t *testing.T, b launch.Bus, state, key string) []kept {
+	t.Helper()
+	raw, err := ask(b, "history", map[string]any{"state": state, "key": key})
+	if err != nil {
+		t.Fatalf("history of %s %s: %v", state, key, err)
+	}
+	var h []kept
+	if err := json.Unmarshal(raw, &h); err != nil || h == nil {
+		t.Fatalf("history answered %s, not an array", raw)
+	}
+	return h
+}
+
+// A key's history is every value the bucket keeps, oldest first, each with its revision, op and time
+// in RFC 3339 — a delete with no value; an unknown key's is []. The owner and a reader are served it,
+// and reading it writes nothing.
+func TestAKeysHistoryIsEveryValueKeptOldestFirst(t *testing.T) {
+	owner, reader, _ := historyBuses(t)
+	before := time.Now().Add(-time.Second)
+	var revisions []uint64
+	for _, v := range []string{`{"state":"open"}`, `{"state":"fixing"}`, `{"state":"resolved"}`} {
+		raw, err := ask(owner, "put", map[string]any{"state": "issues", "key": "502", "value": json.RawMessage(v)})
+		if err != nil {
+			t.Fatal(err)
+		}
+		var r struct {
+			Revision uint64 `json:"revision"`
+		}
+		_ = json.Unmarshal(raw, &r)
+		revisions = append(revisions, r.Revision)
+	}
+	if _, err := ask(owner, "put", map[string]any{"state": "issues", "key": "503", "value": 1}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ask(owner, "delete", map[string]any{"state": "issues", "key": "502"}); err != nil {
+		t.Fatal(err)
+	}
+
+	for who, b := range map[string]launch.Bus{"the owner": owner, "a reader": reader} {
+		name := "issues"
+		if b == reader {
+			name = "issues.issues"
+		}
+		h := historyOf(t, b, name, "502")
+		if len(h) != 4 {
+			t.Fatalf("%s read %d values of 502, not 4: %+v", who, len(h), h)
+		}
+		for i, v := range []string{`{"state":"open"}`, `{"state":"fixing"}`, `{"state":"resolved"}`} {
+			if h[i].Op != "put" || string(h[i].Value) != v || h[i].Revision != revisions[i] || h[i].Key != "502" {
+				t.Errorf("%s: value %d is %+v", who, i, h[i])
+			}
+		}
+		if h[3].Op != "delete" || len(h[3].Value) != 0 && string(h[3].Value) != "null" || h[3].Revision <= revisions[2] {
+			t.Errorf("%s: the delete reads %+v", who, h[3])
+		}
+		for i, k := range h {
+			at, err := time.Parse(time.RFC3339, k.At)
+			if err != nil || at.Before(before) || at.After(time.Now().Add(time.Second)) {
+				t.Errorf("%s: value %d's time %q is not RFC 3339 and now: %v", who, i, k.At, err)
+			}
+			if i > 0 && k.Revision <= h[i-1].Revision {
+				t.Errorf("%s: not oldest first: %+v", who, h)
+			}
+		}
+		if h := historyOf(t, b, name, "999"); len(h) != 0 {
+			t.Errorf("%s: an unknown key's history is %+v, not []", who, h)
+		}
+	}
+	// The history is read, never written: the key is still deleted and keeps the same four values.
+	if raw, err := ask(owner, "get", map[string]any{"state": "issues", "key": "502"}); err != nil || string(raw) != "null" {
+		t.Fatalf("after history, 502 reads %s %v", raw, err)
+	}
+	if h := historyOf(t, owner, "issues", "502"); len(h) != 4 {
+		t.Fatalf("reading the history changed it: %+v", h)
+	}
+	// A key the bucket holds nothing for answers [] on the wire, not null.
+	if raw, _ := ask(owner, "history", map[string]any{"state": "issues", "key": "999"}); string(raw) != "[]" {
+		t.Fatalf("an unknown key's history is %s on the wire", raw)
+	}
+}
+
+// A module the bucket was not issued to is refused its history with the reason, as it is a get.
+func TestAKeysHistoryIsRefusedToAModuleNotIssuedTheState(t *testing.T) {
+	owner, _, other := historyBuses(t)
+	if _, err := ask(owner, "put", map[string]any{"state": "issues", "key": "1", "value": 1}); err != nil {
+		t.Fatal(err)
+	}
+	for _, name := range []string{"issues", "issues.issues"} {
+		_, err := ask(other, "history", map[string]any{"state": name, "key": "1"})
+		if err == nil || !strings.Contains(err.Error(), "keeps and reads no state called") ||
+			!strings.Contains(err.Error(), "issued: none") {
+			t.Errorf("history of %s to a module not issued it: %v", name, err)
+		}
+	}
+	if _, err := ask(owner, "history", map[string]any{"state": "issues", "key": "a b"}); err == nil ||
+		!strings.Contains(err.Error(), "not a key") {
+		t.Errorf("history of a key the bus cannot hold: %v", err)
+	}
+}