Let a bundle write a state key with its own time to live and hear its end, so wakes need no in-process timer (issue 1952)

mesh/state.create and mesh/state.put take ttl_seconds (mesh-sdk 32f5ce3's
CreateTTL/RenewTTL contract); a bucket not declared key-ttl is refused,
naming it; a watch hands an expired key as op "expire". Pins mesh-sdk 32f5ce3.

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",

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_<module>_<name> on mesh.log.<module>.<name>.>, 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

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

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