diff --git a/internal/broker/jetstream.go b/internal/broker/jetstream.go index f709ea5f..71fcb474 100644 --- a/internal/broker/jetstream.go +++ b/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) diff --git a/internal/broker/state.go b/internal/broker/state.go index 32a169f9..a77af471 100644 --- a/internal/broker/state.go +++ b/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 } diff --git a/internal/broker/state_integration_test.go b/internal/broker/state_integration_test.go index fa92d0cb..50a858f7 100644 --- a/internal/broker/state_integration_test.go +++ b/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) + } +} diff --git a/internal/broker/state_keyttl_test.go b/internal/broker/state_keyttl_test.go new file mode 100644 index 00000000..98b85378 --- /dev/null +++ b/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) + } +} diff --git a/internal/catalogue/state.go b/internal/catalogue/state.go index 1d08e229..4c3dcb8d 100644 --- a/internal/catalogue/state.go +++ b/internal/catalogue/state.go @@ -36,6 +36,9 @@ type StateDeclaration struct { // PerMachine says the state is keyed by machine: each machine's instance writes the key named for // its machine and no other, and the bus grants it that key alone (novox/hq ADR 0260). PerMachine bool `json:"per-machine,omitempty"` + // KeyTTL says each key of the bucket lives the time its writer gives it, and the server writes a delete marker + // when it ends (novox/hq ADR 0433 §1): the bucket is made with the server's time to live per message. + KeyTTL bool `json:"key-ttl,omitempty"` } // UnmarshalJSON reads a bucket as its bare name, or as {name, history, ttl-seconds}. @@ -49,7 +52,7 @@ func (s *StateDeclaration) UnmarshalJSON(raw []byte) error { dec := json.NewDecoder(bytes.NewReader(trimmed)) dec.DisallowUnknownFields() if err := dec.Decode(&full); err != nil { - return fmt.Errorf("a state is either a name or {name, history, ttl-seconds, per-machine}: %w", typedUnknown(err)) + return fmt.Errorf("a state is either a name or {name, history, ttl-seconds, per-machine, key-ttl}: %w", typedUnknown(err)) } *s = StateDeclaration(full) return nil @@ -57,7 +60,7 @@ func (s *StateDeclaration) UnmarshalJSON(raw []byte) error { // MarshalJSON writes back the short form when there is nothing else to say. func (s StateDeclaration) MarshalJSON() ([]byte, error) { - if s.History == 0 && s.TTLSeconds == 0 && !s.PerMachine { + if s.History == 0 && s.TTLSeconds == 0 && !s.PerMachine && !s.KeyTTL { return json.Marshal(s.Name) } type plain StateDeclaration diff --git a/internal/catalogue/state_test.go b/internal/catalogue/state_test.go index e82f9711..d4cf687c 100644 --- a/internal/catalogue/state_test.go +++ b/internal/catalogue/state_test.go @@ -39,7 +39,7 @@ func TestAStateNameIsLocalAndOneToken(t *testing.T) { {`{"module":"a.b","version":"1","state":["s"]}`, `no dot`}, {`{"module":"a","version":"1","reads":["bindings"]}`, `a read is .`}, {`{"module":"a","version":"1","reads":["a.s"]}`, `which is its own state`}, - {`{"module":"a","version":"1","state":[{"name":"s","shared":true}]}`, `{name, history, ttl-seconds, per-machine}`}, + {`{"module":"a","version":"1","state":[{"name":"s","shared":true}]}`, `{name, history, ttl-seconds, per-machine, key-ttl}`}, } { _, err := ParseManifest([]byte(c.manifest)) if err == nil { @@ -107,3 +107,19 @@ func TestTheCataloguePassNamesAReadItsOwnerDoesNotKeep(t *testing.T) { t.Fatalf("the catalogue pass said %v", problems) } } + +// 1154/AC1: a state entry may say key-ttl; it is read, written back, and a bare name or an entry without it reads as before. +func TestAStateMayHaveKeysThatExpireOneByOne(t *testing.T) { + m, err := ParseManifest([]byte(`{"module":"mesh-delivery-watchman","version":"1",` + + `"state":[{"name":"pulse","key-ttl":true},"stuck",{"name":"seen","history":5}]}`)) + if err != nil { + t.Fatal(err) + } + if !m.State[0].KeyTTL || m.State[1].KeyTTL || m.State[2].KeyTTL { + t.Fatalf("key-ttl not read as said: %+v", m.State) + } + out, _ := json.Marshal(m.State) + if string(out) != `[{"name":"pulse","key-ttl":true},"stuck",{"name":"seen","history":5}]` { + t.Fatalf("written back as %s", out) + } +} diff --git a/internal/inventory/busrecords.go b/internal/inventory/busrecords.go index 15ec696c..e8b78364 100644 --- a/internal/inventory/busrecords.go +++ b/internal/inventory/busrecords.go @@ -269,7 +269,7 @@ func bucketsOf(m catalogue.Manifest) []broker.Bucket { var out []broker.Bucket for _, s := range m.State { out = append(out, broker.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds, - PerMachine: s.PerMachine}) + PerMachine: s.PerMachine, KeyTTL: s.KeyTTL}) } return out } diff --git a/internal/inventory/state_keyttl_test.go b/internal/inventory/state_keyttl_test.go new file mode 100644 index 00000000..a15dfeb3 --- /dev/null +++ b/internal/inventory/state_keyttl_test.go @@ -0,0 +1,16 @@ +package inventory + +import ( + "testing" + + "github.com/novox/mesh-controller/internal/catalogue" +) + +// 1154/AC2: a module's state with key-ttl is composed into a bucket that carries KeyTTL; without it, none. +func TestAKeyTTLStateIsComposedIntoABucketThatSaysSo(t *testing.T) { // 1154/AC2 + got := bucketsOf(catalogue.Manifest{Module: "mesh-delivery-watchman", + State: []catalogue.StateDeclaration{{Name: "pulse", KeyTTL: true}, {Name: "stuck"}}}) + if len(got) != 2 || !got[0].KeyTTL || got[1].KeyTTL { + t.Fatalf("buckets composed as %+v", got) + } +}