Serve a key's kept values so the issue tracker can carry them into its log (ADR 0307, issue 502)
mesh/delivery-group group adr-0307-state-history delivered: every member is delivered
mesh/delivery delivered
mesh/merge-gate pass: builds mesh-tools, node-tools → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without (4 of …
mesh/repo-check pass: THE CHANGE ALTERS ITS OWN CHECK (merge-check.sh): main's version judged it; the change's judges the pull requests after it merges; it…
mesh/delivery-group group adr-0307-state-history delivered: every member is delivered
mesh/delivery delivered
mesh/merge-gate pass: builds mesh-tools, node-tools → ace, g14, novox, shanks; no bus step; every machine composes with the change as it did without (4 of …
mesh/repo-check pass: THE CHANGE ALTERS ITS OWN CHECK (merge-check.sh): main's version judged it; the change's judges the pull requests after it merges; it…
mesh/state.history answers every value a bucket still keeps for a key, oldest first, to whoever may get it, by the same watch the runtime already uses, so no new grant is needed.
This commit is contained in:
+2
-2
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user