From bc6c8a6d244ae5d782f188fc825e475ef7170cbd Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 22:14:42 +0100 Subject: [PATCH] Let a bundle write a key with its own time to live and hear its expiry, so pulses, settles and bounds expire on the bus through the runtime (issue 1952) --- go/seat.go | 6 ++++++ go/state.go | 34 +++++++++++++++++++++++++++++++++- go/state_test.go | 43 +++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 1 deletion(-) diff --git a/go/seat.go b/go/seat.go index 18f49a5..8a9b71f 100644 --- a/go/seat.go +++ b/go/seat.go @@ -177,6 +177,12 @@ func (s KeptState) Create(key string, value any) (uint64, error) { return s.revisionOf(s.ask("create", map[string]any{"key": key, "value": value})) } +// CreateTTL is Create with the key's own time to live (PutTTL): refused as ErrChanged when the key has a value, so an act +// begun once is not begun again; when ttl passes a watch hands the key over as OpExpire. +func (s KeptState) CreateTTL(key string, value any, ttl time.Duration) (uint64, error) { + return s.revisionOf(s.ask("create", map[string]any{"key": key, "value": value, "ttl_seconds": ttlSeconds(ttl)})) +} + // Update writes a key only when it is still at revision, and returns the new one; a key changed // meanwhile is refused, so of two writers that read one value only the first writes. func (s KeptState) Update(key string, value any, revision uint64) (uint64, error) { diff --git a/go/state.go b/go/state.go index 56490b6..ac3d5fd 100644 --- a/go/state.go +++ b/go/state.go @@ -20,7 +20,18 @@ type StateEntry struct { Revision uint64 `json:"revision"` } -// StateChange is one change a watch hands over: a key put or deleted. +// The operations a watch hands over (StateChange.Op). +const ( + // OpPut is a key written. + OpPut = "put" + // OpDelete is a key deleted or purged. + OpDelete = "delete" + // OpExpire is a key whose own time to live ended (CreateTTL, PutTTL): the runtime says the bus's marker of it as + // "expire", so a watcher tells an expiry, which is the event a pulse or a bound waits for, from a delete. + OpExpire = "expire" +) + +// StateChange is one change a watch hands over: a key put (OpPut), deleted (OpDelete) or expired (OpExpire). type StateChange struct { State string `json:"state"` Key string `json:"key"` @@ -98,6 +109,27 @@ func (s KeptState) Put(key string, value any) (uint64, error) { return put.Revision, json.Unmarshal(raw, &put) } +// PutTTL writes a key whole with its own time to live — the owner's instances only — and returns the revision written: +// the bus removes it when ttl passes and a watch hands that over as OpExpire. The state is declared `key-ttl`. +func (s KeptState) PutTTL(key string, value any, ttl time.Duration) (uint64, error) { + raw, err := s.ask("put", map[string]any{"key": key, "value": value, "ttl_seconds": ttlSeconds(ttl)}) + if err != nil { + return 0, err + } + var put struct { + Revision uint64 `json:"revision"` + } + return put.Revision, json.Unmarshal(raw, &put) +} + +// ttlSeconds is a time to live as the runtime takes it, whole seconds, at least one. +func ttlSeconds(ttl time.Duration) int64 { + if s := int64(ttl / time.Second); s > 1 { + return s + } + return 1 +} + // Delete removes a key — the owner's instances only. A key that was not there is not an error. func (s KeptState) Delete(key string) error { _, err := s.ask("delete", map[string]any{"key": key}) diff --git a/go/state_test.go b/go/state_test.go index 79c5350..7b5f109 100644 --- a/go/state_test.go +++ b/go/state_test.go @@ -3,6 +3,8 @@ package stdio import ( "bufio" "encoding/json" + "errors" + "fmt" "io" "strings" "testing" @@ -33,6 +35,12 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { return State("servers").Put(a["key"].(string), a["value"]) }}, {Name: "recall", Run: func(a map[string]any) (any, error) { return State("servers").Get(a["key"].(string)) }}, + {Name: "pulse", Run: func(a map[string]any) (any, error) { + if a["create"] == true { + return State("servers").CreateTTL(a["key"].(string), 1, 90*time.Second) + } + return State("servers").PutTTL(a["key"].(string), 1, 90*time.Second) + }}, }, inR, outW) }() lines := bufio.NewScanner(outR) @@ -87,6 +95,15 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { if err := <-watched; err != nil { t.Fatal(err) } + // 1952 (the runtime's per-key time to live, mesh-tools 1583fa85): an expiry is handed over as its own operation. + send(map[string]any{"jsonrpc": "2.0", "id": 49, "method": "mesh/state", "params": map[string]any{"watch": watch["id"], + "change": map[string]any{"state": "servers", "key": "all.gone", "op": OpExpire}}}) + if r := next(); r["id"] != float64(49) || r["error"] != nil { + t.Fatalf("an expiry was answered %v", r) + } + if s := <-seen; s != "expire all.gone " { + t.Fatalf("handled %q", s) + } // A change for a watch this bundle never asked for is refused. send(map[string]any{"jsonrpc": "2.0", "id": 51, "method": "mesh/state", "params": map[string]any{"watch": "bundle-999", "change": map[string]any{"key": "x", "op": "put"}}}) @@ -105,6 +122,32 @@ func TestABundleKeepsAndWatchesStateThroughTheRuntime(t *testing.T) { if r := next(); r["id"] != float64(2) || r["result"].(map[string]any)["content"].([]any)[0].(map[string]any)["text"] != "9" { t.Fatalf("the put answered %v", r) } + // PutTTL and CreateTTL ask the runtime's put and create with the key's time to live in whole seconds; a create the + // runtime refuses as changed (-32010) is ErrChanged. + for i, create := range []bool{false, true} { + send(map[string]any{"jsonrpc": "2.0", "id": 20 + i, "method": "tools/call", "params": map[string]any{"name": "pulse", + "arguments": map[string]any{"key": "pulse.S1", "create": create}}}) + asked := next() + p := asked["params"].(map[string]any) + want := map[bool]string{false: "mesh/state.put", true: "mesh/state.create"}[create] + if asked["method"] != want || p["ttl_seconds"] != float64(90) || p["key"] != "pulse.S1" { + t.Fatalf("asked %v", asked) + } + if create { + send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "error": map[string]any{"code": CodeChanged, "message": "exists"}}) + if r := next(); r["id"] != float64(21) || !strings.Contains(fmt.Sprint(r), "exists") { + t.Fatalf("a refused create answered %v", r) + } + continue + } + send(map[string]any{"jsonrpc": "2.0", "id": asked["id"], "result": map[string]any{"revision": 4}}) + if r := next(); r["id"] != float64(20) { + t.Fatalf("the put answered %v", r) + } + } + if !errors.Is(&refused{code: CodeChanged, message: "x"}, ErrChanged) { + t.Fatal("a refusal coded -32010 is not ErrChanged") + } // Get of a key with no value is nil. send(map[string]any{"jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": map[string]any{"name": "recall", "arguments": map[string]any{"key": "nothing"}}})