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)
This commit is contained in:
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user