diff --git a/node-tools/go.mod b/node-tools/go.mod index 7a01da7..2b2f91b 100644 --- a/node-tools/go.mod +++ b/node-tools/go.mod @@ -8,7 +8,7 @@ require ( ) require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268 github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect github.com/google/go-tpm v0.9.8 // indirect github.com/klauspost/compress v1.20.0 // indirect diff --git a/node-tools/go.sum b/node-tools/go.sum index 2a3e9f3..4495931 100644 --- a/node-tools/go.sum +++ b/node-tools/go.sum @@ -1,5 +1,7 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268 h1:+jH77IJmxUiFDdbJljhynQRY0Cc5/5jtwmMV7UUqRI4= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029210043-32f5ce371268/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/node-tools/internal/bus/state.go b/node-tools/internal/bus/state.go index a673168..d6bd54b 100644 --- a/node-tools/internal/bus/state.go +++ b/node-tools/internal/bus/state.go @@ -37,7 +37,8 @@ type StateEntry struct { Revision uint64 `json:"revision"` } -// StateChange is one change a watch delivers: a key put or deleted. +// StateChange is one change a watch delivers: a key put, deleted, or expired (its own time to live, or the +// bucket's age, ended; issue 1952). type StateChange struct { State string `json:"state"` Key string `json:"key"` @@ -199,6 +200,61 @@ func (c *Conn) StatePut(module, name, key string, value json.RawMessage) (uint64 return kv.Put(ctx, key, value) } +// MinKeyTTL is the shortest time to live a key may be given: the bus counts a key's age in whole seconds (mesh-sdk +// process.MinTTL). +const MinKeyTTL = time.Second + +// StateCreateTTL writes one key only when it has no value, with its own time to live (novox/hq ADR 0564 clauses 4 and +// 7, issue 1952; mesh-sdk process.State.CreateTTL): the bus removes it when ttl passes and every watch hears the +// removal as an expiry, so a bundle's wake needs no timer. A key that has a value is refused as ErrStateChanged. The +// bucket must take keys with their own time to live (a state declared `key-ttl`); one that does not is refused, naming +// it, before anything is written. +func (c *Conn) StateCreateTTL(module, name, key string, value json.RawMessage, ttl time.Duration) (uint64, error) { + if ttl < MinKeyTTL { + return 0, fmt.Errorf("%s's state %s, key %s: a time to live of %s is shorter than the bus keeps (%s)", + module, name, key, ttl, MinKeyTTL) + } + return c.stateWrite(module, name, key, value, func(ctx context.Context, kv jetstream.KeyValue) (uint64, error) { + if err := takesKeyTTL(ctx, kv, module, name); err != nil { + return 0, err + } + return kv.Create(ctx, key, value, jetstream.KeyTTL(ttl)) + }) +} + +// StateRenewTTL writes one key whole with a fresh time to live, whatever it held (mesh-sdk process.State.RenewTTL): a +// time to live is set only when a key is created, so a live key is deleted first, and its watches hear the delete and +// then the new value. +func (c *Conn) StateRenewTTL(module, name, key string, value json.RawMessage, ttl time.Duration) (uint64, error) { + rev, err := c.StateCreateTTL(module, name, key, value, ttl) + if !errors.Is(err, ErrStateChanged) { + return rev, err + } + if err := c.StateDelete(module, name, key); err != nil { + return 0, err + } + return c.StateCreateTTL(module, name, key, value, ttl) +} + +// takesKeyTTL refuses a bucket that does not allow a time to live per key, or leaves no marker when one ends: there the +// server would refuse the write, or the key would vanish and no watch would hear it. +func takesKeyTTL(ctx context.Context, kv jetstream.KeyValue, module, name string) error { + status, err := kv.Status(ctx) + if err != nil { + return err + } + allows := status.LimitMarkerTTL() > 0 + if b, ok := status.(*jetstream.KeyValueBucketStatus); ok && b.StreamInfo() != nil { + allows = allows && b.StreamInfo().Config.AllowMsgTTL + } + if !allows { + return fmt.Errorf("%s's state %s does not take keys with their own time to live: its bucket %s allows none, or "+ + "leaves no marker a watch hears when one ends; %s declares the state `key-ttl` for it (novox/hq ADR 0564, "+ + "issue 1952)", module, name, status.Bucket(), module) + } + return nil +} + // StateDelete removes one key, as the module, where the module keeps the state. A key that was not // there is not an error: what is asked for is that it is gone. func (c *Conn) StateDelete(module, name, key string) error { @@ -343,6 +399,12 @@ func (c *Conn) StateWatch(module, name, pattern string, deliver func(StateChange continue } change.Op = "delete" + if e.Operation() == jetstream.KeyValuePurge { + // The marker the server leaves when a key's time to live ends (Nats-Marker-Reason MaxAge, + // read by the client as a purge): nothing else purges a module's state, since neither the + // runtime nor mesh-sdk's State offers a purge (issue 1952). + change.Op = "expire" + } } if err := deliver(change); err != nil { c.Logf("[mesh-tools] %s did not take %s %s of %s: %v; its next change, or its next start, reads it again", diff --git a/node-tools/internal/meshtest/meshtest.go b/node-tools/internal/meshtest/meshtest.go index 9edec5e..09c1bb5 100644 --- a/node-tools/internal/meshtest/meshtest.go +++ b/node-tools/internal/meshtest/meshtest.go @@ -11,6 +11,7 @@ package meshtest import ( + "context" "encoding/json" "os" "path/filepath" @@ -22,6 +23,7 @@ import ( "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-tools/node-tools/internal/bus" ) @@ -243,6 +245,23 @@ func (m *Mesh) Bucket(t *testing.T, name string) nats.KeyValue { return kv } +// BucketKeyTTL makes a module's state afresh as the controller does for one declared `key-ttl` (mesh-sdk buscompose: +// a time to live per key allowed, its end leaving a marker kept a minute; issue 1952). +func (m *Mesh) BucketKeyTTL(t *testing.T, name string) { + t.Helper() + _ = m.js.DeleteKeyValue(name) + js, err := jetstream.New(m.nc) + if err != nil { + t.Fatal(err) + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 1, MaxValueSize: 256 * 1024, + LimitMarkerTTL: time.Minute}); err != nil { + t.Fatal(err) + } +} + // Log makes a module's log afresh as the controller does from the catalogue (novox/hq ADR 0297): a file // stream named LOG__ on mesh.log...>, limits retention, no age, no cap per // key, the declared cap, 260 KiB a message (a 256 KiB value and its headers), refusing new entries when full, direct get, no delete or diff --git a/node-tools/internal/runtime/runtime.go b/node-tools/internal/runtime/runtime.go index 5cd47f9..c81773b 100644 --- a/node-tools/internal/runtime/runtime.go +++ b/node-tools/internal/runtime/runtime.go @@ -868,10 +868,14 @@ type stateAsked struct { Key string `json:"key"` Value json.RawMessage `json:"value"` Revision uint64 `json:"revision"` + // TTLSeconds gives a created or put key its own time to live (issue 1952): create makes it only when it has no + // value, put renews it whole. Zero is none. + TTLSeconds int `json:"ttl_seconds"` } // 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. +// module's state (ADR 0201, ADR 0259, ADR 0307); `create` and `put` take `ttl_seconds`, a key's own time to live, +// whose end every watch hears as an `expire` (issue 1952). 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 == "" { @@ -897,7 +901,13 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, } answer = kept case "put": - revision, err := conn.StatePut(b.module, asked.State, asked.Key, asked.Value) + put := conn.StatePut + if asked.TTLSeconds != 0 { + put = func(module, name, key string, value json.RawMessage) (uint64, error) { + return conn.StateRenewTTL(module, name, key, value, time.Duration(asked.TTLSeconds)*time.Second) + } + } + revision, err := put(b.module, asked.State, asked.Key, asked.Value) if err != nil { return nil, err } @@ -915,13 +925,23 @@ func (b *moduleBus) State(verb string, params json.RawMessage) (json.RawMessage, answer = keys case "create": // Only when the key has no value: of two writers making it, one wins (novox/hq ADR 0259). - revision, err := conn.StateCreate(b.module, asked.State, asked.Key, asked.Value) + create := conn.StateCreate + if asked.TTLSeconds != 0 { + create = func(module, name, key string, value json.RawMessage) (uint64, error) { + return conn.StateCreateTTL(module, name, key, value, time.Duration(asked.TTLSeconds)*time.Second) + } + } + revision, err := create(b.module, asked.State, asked.Key, asked.Value) if err != nil { return nil, err } answer = map[string]any{"revision": revision} case "update": // Only when the key is still at the revision read: a second answer to one ask loses (ADR 0259). + if asked.TTLSeconds != 0 { + return nil, fmt.Errorf("mesh/state.update gives no time to live: the bus sets one only when a key is " + + "created; put the key with ttl_seconds to renew it whole (issue 1952)") + } revision, err := conn.StateUpdate(b.module, asked.State, asked.Key, asked.Value, asked.Revision) if err != nil { return nil, err diff --git a/node-tools/internal/runtime/state_counter_test.go b/node-tools/internal/runtime/state_counter_test.go index 0a47db0..a6de41e 100644 --- a/node-tools/internal/runtime/state_counter_test.go +++ b/node-tools/internal/runtime/state_counter_test.go @@ -4,8 +4,10 @@ import ( "encoding/json" "errors" "fmt" + "strings" "sync" "testing" + "time" "github.com/novox/mesh-tools/node-tools/internal/bus" "github.com/novox/mesh-tools/node-tools/internal/launch" @@ -18,8 +20,10 @@ func counterBus(t *testing.T) launch.Bus { t.Helper() mesh := mt.New(t) mesh.Bucket(t, "issues_numbers") + mesh.BucketKeyTTL(t, "issues_wakes") m := mt.MembershipOf("issues", "anchor", false, nil) - m.State = []bus.StateIssued{{Name: "numbers", Bucket: "issues_numbers", Writes: true}} + m.State = []bus.StateIssued{{Name: "numbers", Bucket: "issues_numbers", Writes: true}, + {Name: "wakes", Bucket: "issues_wakes", Writes: true}} mesh.Issue(t, m) conn := connect(t, "node-tools", "anchor") conn.Follow("issues") @@ -131,6 +135,9 @@ func TestTwoWritersIncrementingOneCounterOnTheBusLoseNoUpdate(t *testing.T) { // A write at a revision that is no longer the key's is refused with the lost-write code and writes // nothing; create is refused for a key with a value and taken for one deleted; an update names the // revision it read, never zero. +// A key's own time to live (issue 1952, mesh-sdk process.State.CreateTTL and RenewTTL): created once, renewed whole by a +// put, its end heard by a watch as an expiry; refused, writing nothing, on a bucket that takes none, under a second, and +// on an update. func TestAStaleWriteIsRefusedAndWritesNothing(t *testing.T) { b := counterBus(t) state := func(verb string, extra map[string]any) (uint64, error) { @@ -200,4 +207,57 @@ func TestAStaleWriteIsRefusedAndWritesNothing(t *testing.T) { if v := value(); v != "1" { t.Fatalf("after delete and create: %s", v) } + + // A time to live per key: refused where it cannot be kept, writing nothing. + for _, c := range []struct { + verb, state, why string + ttl int + }{ + {"create", "numbers", "does not take keys with their own time to live", 60}, + {"put", "numbers", "does not take keys with their own time to live", 60}, + {"create", "wakes", "shorter than the bus keeps", -1}, + {"update", "wakes", "gives no time to live", 60}, + } { + _, err := ask(b, c.verb, map[string]any{"state": c.state, "key": "late", "value": 1, "revision": 1, "ttl_seconds": c.ttl}) + if err == nil || !strings.Contains(err.Error(), c.why) { + t.Errorf("%s with ttl %d on %s: want refused for %q, got %v", c.verb, c.ttl, c.state, c.why, err) + } + if raw, _ := ask(b, "get", map[string]any{"state": c.state, "key": "late"}); string(raw) != "null" { + t.Errorf("%s with ttl %d on %s wrote %s", c.verb, c.ttl, c.state, raw) + } + } + + // Created once, renewed whole, and its end heard: a wake with no timer. + heard := make(chan bus.StateChange, 8) + stop, err := b.Watch(json.RawMessage(`{"state":"wakes"}`), func(raw json.RawMessage) error { + var c bus.StateChange + _ = json.Unmarshal(raw, &c) + heard <- c + return nil + }) + if err != nil { + t.Fatal(err) + } + defer stop() + if _, err := ask(b, "create", map[string]any{"state": "wakes", "key": "settle", "value": 1, "ttl_seconds": 60}); err != nil { + t.Fatalf("create with a time to live: %v", err) + } + if _, err := ask(b, "create", map[string]any{"state": "wakes", "key": "settle", "value": 2, "ttl_seconds": 60}); !isLost(err) { + t.Fatalf("a second create with a time to live: %v", err) + } + if _, err := ask(b, "put", map[string]any{"state": "wakes", "key": "settle", "value": 3, "ttl_seconds": 1}); err != nil { + t.Fatalf("a put renewing the time to live: %v", err) + } + var ops []string + for want := "put delete put expire"; strings.Join(ops, " ") != want; { + select { + case c := <-heard: + ops = append(ops, c.Op) + if c.Op == "put" && len(ops) == 3 && string(c.Value) != "3" { + t.Fatalf("the renewed key holds %s", c.Value) + } + case <-time.After(5 * time.Second): + t.Fatalf("the watch heard %q, want %q", strings.Join(ops, " "), want) + } + } }