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

@@ -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) {

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})

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"}}})