diff --git a/merge-check.sh b/merge-check.sh index 8ce0dfe..de57f12 100644 --- a/merge-check.sh +++ b/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 diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index deb39a9..e0125bd 100644 --- a/node-tools/internal/bus/state.go +++ b/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) diff --git a/node-tools/internal/launch/launch.go b/node-tools/internal/launch/launch.go index 23b7c41..ce78b49 100644 --- a/node-tools/internal/launch/launch.go +++ b/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 diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index 6dd6718..98f251c 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/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 { diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index be9a91d..a98e425 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/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 { diff --git a/node-tools/internal/runtime/state_history_test.go b/node-tools/internal/runtime/state_history_test.go new file mode 100644 index 0000000..a9afccc --- /dev/null +++ b/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) + } +}