From 575a05e49e59f96064399df65be34e2e84556d06 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 29 Oct 2026 21:36:11 +0100 Subject: [PATCH] Let a state create a key with its own time to live, so a pulse, a settle or a hold bound expires on the bus with no timer (issue 1952 S6, ADR 0564 clauses 4, 7, 13) --- go/process/state.go | 45 ++++++++++++++++++++++++++++ go/process/state_integration_test.go | 18 ++++++++++- go/process/state_test.go | 30 ++++++++++++++++++- 3 files changed, 91 insertions(+), 2 deletions(-) diff --git a/go/process/state.go b/go/process/state.go index 777a9fd..90523b4 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -113,6 +113,51 @@ func (s State) Create(key string, value any) (uint64, error) { return rev, nil } +// MinTTL is the shortest time to live a key may be given: the bus counts a key's age in whole seconds. +const MinTTL = time.Second + +// CreateTTL writes a key only when it has none, with its own time to live (novox/hq ADR 0564 clauses 4 and 7, issue +// 1952 S6): the bus removes it when ttl passes and its watchers hear the removal, so an expiry is an event and no timer +// runs. A key that has a value is refused as stdio.ErrChanged: an act begun once cannot be begun again. The bucket must +// take keys with their own time to live (a state declared `key-ttl`). +func (s State) CreateTTL(key string, value any, ttl time.Duration) (uint64, error) { + if ttl < MinTTL { + return 0, fmt.Errorf("%s's state %s, key %s: a time to live of %s is shorter than the bus keeps (%s)", s.Module, s.Name, key, ttl, MinTTL) + } + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return 0, err + } + data, err := json.Marshal(value) + if err != nil { + return 0, err + } + rev, err := kv.Create(ctx, key, data, jetstream.KeyTTL(ttl)) + if errors.Is(err, jetstream.ErrKeyExists) { + return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged) + } + if err != nil { + return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return rev, nil +} + +// RenewTTL writes a key whole with a fresh time to live, whatever it held: a pulse its signal renews (ADR 0564 clause +// 4). A time to live is set only when a key is created, so a live key is deleted first; its watchers hear the delete and +// then the new value. +func (s State) RenewTTL(key string, value any, ttl time.Duration) (uint64, error) { + rev, err := s.CreateTTL(key, value, ttl) + if !errors.Is(err, stdio.ErrChanged) { + return rev, err + } + if err := s.Delete(key); err != nil { + return 0, err + } + return s.CreateTTL(key, value, ttl) +} + // Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile // is refused as stdio.ErrChanged. func (s State) Update(key string, value any, revision uint64) (uint64, error) { diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index bd34388..5c5f07f 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -44,7 +44,7 @@ func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) { if _, err := awaited.Put(ctxFor(t), "hearing", []byte(`{"heard":true}`)); !errors.Is(err, ErrNotYet) { t.Fatalf("a second write before the bucket exists: %v, want ErrNotYet", err) } - if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil { + if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends", LimitMarkerTTL: time.Second}); err != nil { t.Fatalf("the admin could not make the bucket: %v", err) } select { @@ -94,6 +94,22 @@ func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) { if _, err := s.Create("key", map[string]any{"v": 4}); !errors.Is(err, stdio.ErrChanged) { t.Fatalf("Create of an existing key: %v, want stdio.ErrChanged", err) } + + // 1952 S6 (ADR 0564 clause 4): a key created with its own time to live is gone once it passes, on the bus itself. + if _, err := s.CreateTTL("pulse", map[string]any{"v": 5}, time.Second); err != nil { + t.Fatalf("CreateTTL on a bucket that takes keys with a time to live: %v", err) + } + deadline := time.Now().Add(5 * time.Second) + for { + e, err := s.Get("pulse") + if err == nil && e == nil { + break + } + if time.Now().After(deadline) { + t.Fatalf("the key created with a second to live still reads %+v, %v after 5s", e, err) + } + time.Sleep(100 * time.Millisecond) + } } // 1188/AC4: given an account not granted a bucket, when it writes there, then the bus's refusal comes diff --git a/go/process/state_test.go b/go/process/state_test.go index 02b084a..7cc0a93 100644 --- a/go/process/state_test.go +++ b/go/process/state_test.go @@ -38,6 +38,8 @@ type fakeKV struct { // initial is what a Watch hands first, before its nil marker; watcher is the last one made. initial []jetstream.KeyValueEntry watcher *fakeWatcher + // withTTL says the last create of a key carried an option: its time to live. + withTTL map[string]bool } func newFakeKV() *fakeKV { return &fakeKV{values: map[string][]byte{}, revs: map[string]uint64{}} } @@ -52,9 +54,13 @@ func (f *fakeKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, er return fakeEntry{key: key, value: v, rev: f.revs[key]}, nil } -func (f *fakeKV) Create(_ context.Context, key string, value []byte, _ ...jetstream.KVCreateOpt) (uint64, error) { +func (f *fakeKV) Create(_ context.Context, key string, value []byte, opts ...jetstream.KVCreateOpt) (uint64, error) { f.mu.Lock() defer f.mu.Unlock() + if f.withTTL == nil { + f.withTTL = map[string]bool{} + } + f.withTTL[key] = len(opts) > 0 if _, ok := f.values[key]; ok { return 0, jetstream.ErrKeyExists } @@ -160,6 +166,28 @@ func TestStateSatisfiesAskerStore(t *testing.T) { if rev2 == rev { t.Fatal("Update did not advance the revision") } + + // 1952 S6 (ADR 0564 clauses 4, 7): a key created with its own time to live is created once, a second create + // refused as an act begun already; a renewal writes it whole with a fresh time to live; a time to live the bus + // cannot keep is refused. + kv := newFakeKV() + s = stateOn(kv) + if _, err := s.CreateTTL("settle.H3", map[string]any{"a": 1}, time.Minute); err != nil || !kv.withTTL["settle.H3"] { + t.Fatalf("CreateTTL: %v, with a time to live %v", err, kv.withTTL["settle.H3"]) + } + if _, err := s.CreateTTL("settle.H3", map[string]any{"a": 2}, time.Minute); !errors.Is(err, stdio.ErrChanged) { + t.Fatalf("CreateTTL of a live key: %v, want stdio.ErrChanged", err) + } + kv.withTTL["settle.H3"] = false + if _, err := s.RenewTTL("settle.H3", map[string]any{"a": 3}, time.Minute); err != nil || !kv.withTTL["settle.H3"] { + t.Fatalf("RenewTTL: %v, with a time to live %v", err, kv.withTTL["settle.H3"]) + } + if e, _ := s.Get("settle.H3"); e == nil || string(e.Value) != `{"a":3}` { + t.Fatalf("after RenewTTL the key reads %+v", e) + } + if _, err := s.CreateTTL("pulse.S1", 1, 10*time.Millisecond); err == nil || !strings.Contains(err.Error(), "shorter") { + t.Fatalf("a time to live under a second: %v", err) + } } // 1840/S1: given a state, when a key is put twice, then each Put returns a new revision and the key reads