A state entry may say key-ttl and its bucket allows a time to live per key (issue 1154)

internal/broker/jetstream.go

@@ -436,30 +436,50 @@ func retentionOf(r Retention) nats.RetentionPolicy {
 // discards what the bucket holds, and what a module's state holds is data. The mesh's caps are
 // asserted with the owner's options, so a bucket made by hand converges to them.
 func (j *JetStream) EnsureBucket(b Bucket) error {
-	history := b.History
-	if history == 0 {
-		history = 1
-	}
 	js, err := jetstream.New(j.conn)
 	if err != nil {
 		return err
 	}
 	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
-	if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
-		Bucket:       b.Bucket(),
-		Description:  b.Why(),
-		History:      uint8(history),
-		TTL:          time.Duration(b.TTLSeconds) * time.Second,
-		MaxValueSize: StateMaxValueBytes,
-		MaxBytes:     StateMaxBytes,
-		Storage:      jetstream.FileStorage,
-	}); err != nil {
+	// A stream that allows a time to live per message cannot be made to disallow it (the server refuses the update),
+	// so a bucket that has it keeps it whatever the owner now declares (novox/hq ADR 0433 §1).
+	var held time.Duration
+	if kv, err := js.KeyValue(ctx, b.Bucket()); err == nil {
+		if st, err := kv.Status(ctx); err == nil {
+			held = st.LimitMarkerTTL()
+		}
+	}
+	if _, err := js.CreateOrUpdateKeyValue(ctx, BucketConfig(b, held)); err != nil {
 		return fmt.Errorf("asserting bucket %s: %w", b.Bucket(), err)
 	}
 	return nil
 }
 
+// BucketConfig is the options a module's bucket is asserted with. held is the marker age the bucket already has, zero
+// when it is absent or has none: **a bucket that allows a time to live per message keeps it** (the server refuses to
+// turn it off), so without key-ttl it is asserted with the age it holds, and with key-ttl with KeyTTLMarkerTTL.
+func BucketConfig(b Bucket, held time.Duration) jetstream.KeyValueConfig {
+	history := b.History
+	if history == 0 {
+		history = 1
+	}
+	markers := held
+	if b.KeyTTL {
+		markers = KeyTTLMarkerTTL
+	}
+	return jetstream.KeyValueConfig{
+		Bucket:         b.Bucket(),
+		Description:    b.Why(),
+		History:        uint8(history),
+		TTL:            time.Duration(b.TTLSeconds) * time.Second,
+		MaxValueSize:   StateMaxValueBytes,
+		MaxBytes:       StateMaxBytes,
+		Storage:        jetstream.FileStorage,
+		LimitMarkerTTL: markers,
+	}
+}
+
 // BucketNames is every key-value bucket on the server, the mesh's and anybody else's.
 func (j *JetStream) BucketNames() ([]string, error) {
 	js, err := jetstream.New(j.conn)

internal/broker/state.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"sort"
 	"strings"
+	"time"
 )
 
 // A module's state on the bus (novox/hq ADR 0201, design 32 §4, design 25 §3).
@@ -35,8 +36,14 @@ type Bucket struct {
 	// PerMachine says each machine's instance writes only the key named for its machine (novox/hq ADR
 	// 0260), and is granted that key and no other.
 	PerMachine bool
+	// KeyTTL says the bucket allows a time to live per key (novox/hq ADR 0433 §1): its owner writes each key with its
+	// own, and the server writes a delete marker, kept KeyTTLMarkerTTL, when it ends.
+	KeyTTL bool
 }
 
+// KeyTTLMarkerTTL is how long the marker an expired key leaves is kept: long enough for a watcher to hear it.
+const KeyTTLMarkerTTL = time.Minute
+
 // BucketName is the bucket a module's state lives in: the module and the local name joined by an
 // underscore, which neither may contain, so two modules can never derive one bucket.
 func BucketName(module, name string) string { return module + "_" + name }

internal/broker/state_integration_test.go

@@ -3,9 +3,14 @@
 package broker
 
 import (
+	"context"
+	"errors"
+	"time"
+
 	"testing"
 
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
 )
 
 func TestABucketIsAssertedInPlace(t *testing.T) {
@@ -53,3 +58,48 @@ func TestABucketIsAssertedOverAHeldConnection(t *testing.T) {
 		t.Fatalf("the bucket is not there: %v", err)
 	}
 }
+
+// 1154/AC4: on the pinned server a bucket declared with key-ttl takes a put with a time to live of a second, the key is
+// gone after it and the server has written a delete marker; and the bucket asserted again without key-ttl is not refused.
+func TestAKeyTTLBucketExpiresAKeyAndIsAssertedAgain(t *testing.T) { // 1154/AC4
+	js := aLiveBus(t)
+	b := Bucket{Module: "statetest", Name: "pulse", KeyTTL: true}
+	if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
+		t.Fatalf("a real server refused a key-ttl bucket: %v", err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+	kv, err := jetstream.New(js.Conn())
+	if err != nil {
+		t.Fatal(err)
+	}
+	bucket, err := kv.KeyValue(ctx, b.Bucket())
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := bucket.Create(ctx, "presence.alpha", []byte("x"), jetstream.KeyTTL(time.Second)); err != nil {
+		t.Fatalf("the bucket refused a key with a time to live: %v", err)
+	}
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		if _, err := bucket.Get(ctx, "presence.alpha"); errors.Is(err, jetstream.ErrKeyNotFound) {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatal("the key is still there after its time to live")
+		}
+		time.Sleep(200 * time.Millisecond)
+	}
+	hist, err := bucket.History(ctx, "presence.alpha")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if last := hist[len(hist)-1]; last.Operation() == jetstream.KeyValuePut {
+		t.Fatalf("no delete marker was left: last entry is %v", last.Operation())
+	}
+	// The owner no longer says key-ttl: the server cannot turn the age off, so the bucket keeps it and is not refused.
+	b.KeyTTL = false
+	if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
+		t.Fatalf("asserting the bucket without key-ttl was refused, so a restart would fail: %v", err)
+	}
+}

internal/broker/state_keyttl_test.go

@@ -0,0 +1,31 @@
+package broker
+
+import (
+	"testing"
+	"time"
+)
+
+// 1154/AC2: a bucket declared with key-ttl is asserted with the marker age that allows a time to live per message;
+// one without is asserted with none.
+func TestAKeyTTLBucketIsAssertedWithTheMarkerAge(t *testing.T) { // 1154/AC2
+	with := BucketConfig(Bucket{Module: "mesh-delivery-watchman", Name: "pulse", KeyTTL: true}, 0)
+	if with.LimitMarkerTTL != KeyTTLMarkerTTL || KeyTTLMarkerTTL <= 0 {
+		t.Fatalf("a key-ttl bucket is asserted with marker age %v, want %v", with.LimitMarkerTTL, KeyTTLMarkerTTL)
+	}
+	without := BucketConfig(Bucket{Module: "mesh-delivery-watchman", Name: "stuck"}, 0)
+	if without.LimitMarkerTTL != 0 {
+		t.Fatalf("a bucket without key-ttl is asserted with marker age %v, want none", without.LimitMarkerTTL)
+	}
+	if with.Bucket != "mesh-delivery-watchman_pulse" || with.History != 1 || with.MaxBytes != StateMaxBytes {
+		t.Fatalf("the rest of the options changed: %+v", with)
+	}
+}
+
+// 1154/AC3: a bucket that already allows a time to live per message keeps it when its owner no longer says key-ttl,
+// because the server refuses to turn it off; it is not asserted without it.
+func TestABucketThatAllowsAKeyTTLKeepsIt(t *testing.T) { // 1154/AC3
+	got := BucketConfig(Bucket{Module: "m", Name: "s"}, 5*time.Minute)
+	if got.LimitMarkerTTL != 5*time.Minute {
+		t.Fatalf("a bucket that held marker age 5m is asserted with %v, which the server refuses", got.LimitMarkerTTL)
+	}
+}