diff --git a/go/process/state.go b/go/process/state.go index 43221fa..e6c0655 100644 --- a/go/process/state.go +++ b/go/process/state.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "sort" "time" "github.com/nats-io/nats.go" @@ -128,3 +129,41 @@ func (s State) Update(key string, value any, revision uint64) (uint64, error) { } return rev, nil } + +// Put writes a key whole, whatever it held, and returns the revision written. +func (s State) Put(key string, value any) (uint64, error) { + 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.Put(ctx, key, data) + if err != nil { + return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err) + } + return rev, nil +} + +// Keys is every key with a current value, sorted; none is an empty list. +func (s State) Keys() ([]string, error) { + ctx, cancel := context.WithTimeout(context.Background(), stateTimeout) + defer cancel() + kv, err := s.store(ctx) + if err != nil { + return nil, err + } + keys, err := kv.Keys(ctx) + if errors.Is(err, jetstream.ErrNoKeysFound) { + return []string{}, nil + } + if err != nil { + return nil, fmt.Errorf("%s's state %s, keys: %w", s.Module, s.Name, err) + } + sort.Strings(keys) + return keys, nil +} diff --git a/go/process/state_integration_test.go b/go/process/state_integration_test.go index cc01c15..01085b3 100644 --- a/go/process/state_integration_test.go +++ b/go/process/state_integration_test.go @@ -94,3 +94,38 @@ func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) { t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error()) } } + +// 1840/S1: given a process connection granted its bucket, when it puts keys whole and lists them, then Put +// returns a new revision each time and Keys lists them sorted (none for a bucket with none). +func TestStatePutAndKeysOnAGrantedBucket(t *testing.T) { + url, account := grantedSendProcess(t) + _, adminJetStream := adminJS(t, url) + if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil { + t.Fatalf("the admin could not make the bucket: %v", err) + } + conn, err := Connect(account) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + s := conn.State("sends") + + if keys, err := s.Keys(); err != nil || len(keys) != 0 { + t.Fatalf("Keys of an empty bucket: %v, %v", keys, err) + } + rev, err := s.Put("b", map[string]any{"v": 1}) + if err != nil || rev == 0 { + t.Fatalf("Put on a granted bucket: %d, %v", rev, err) + } + rev2, err := s.Put("b", map[string]any{"v": 2}) + if err != nil || rev2 <= rev { + t.Fatalf("Put over a key: %d, %v", rev2, err) + } + if _, err := s.Put("a", 1); err != nil { + t.Fatal(err) + } + keys, err := s.Keys() + if err != nil || strings.Join(keys, ",") != "a,b" { + t.Fatalf("Keys: %v, %v, want a,b", keys, err) + } +} diff --git a/go/process/state_test.go b/go/process/state_test.go index 542153b..ba02ad9 100644 --- a/go/process/state_test.go +++ b/go/process/state_test.go @@ -3,6 +3,8 @@ package process import ( "context" "errors" + "sort" + "strings" "sync" "testing" @@ -68,6 +70,28 @@ func (f *fakeKV) Update(_ context.Context, key string, value []byte, revision ui return f.next, nil } +func (f *fakeKV) Put(_ context.Context, key string, value []byte) (uint64, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.next++ + f.values[key], f.revs[key] = value, f.next + return f.next, nil +} + +func (f *fakeKV) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) { + f.mu.Lock() + defer f.mu.Unlock() + if len(f.values) == 0 { + return nil, jetstream.ErrNoKeysFound + } + keys := make([]string, 0, len(f.values)) + for k := range f.values { + keys = append(keys, k) + } + sort.Sort(sort.Reverse(sort.StringSlice(keys))) // as a bucket lists them: in no order the caller may rely on + return keys, nil +} + func stateOn(kv jetstream.KeyValue) State { return State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { return kv, nil }} } @@ -103,3 +127,38 @@ func TestStateSatisfiesAskerStore(t *testing.T) { t.Fatal("Update did not advance the revision") } } + +// 1840/S1: given a state, when a key is put twice, then each Put returns a new revision and the key reads +// back the last value, whether or not it had one. +func TestStatePutWritesAKeyWhole(t *testing.T) { + s := stateOn(newFakeKV()) + rev, err := s.Put("k", map[string]any{"a": 1}) + if err != nil || rev == 0 { + t.Fatalf("Put of a new key: %d, %v", rev, err) + } + rev2, err := s.Put("k", map[string]any{"a": 2}) + if err != nil || rev2 <= rev { + t.Fatalf("Put over a key: %d, %v, want a revision after %d", rev2, err, rev) + } + e, err := s.Get("k") + if err != nil || e == nil || e.Revision != rev2 || string(e.Value) != `{"a":2}` { + t.Fatalf("Get after Put: %+v, %v", e, err) + } +} + +// 1840/S1: given a state, when its keys are listed, then they come sorted, and none is an empty list, not an error. +func TestStateKeysAreSortedAndEmptyWhenNone(t *testing.T) { + s := stateOn(newFakeKV()) + if keys, err := s.Keys(); err != nil || len(keys) != 0 { + t.Fatalf("Keys of an empty state: %v, %v", keys, err) + } + for _, k := range []string{"b", "c", "a"} { + if _, err := s.Put(k, 1); err != nil { + t.Fatal(err) + } + } + keys, err := s.Keys() + if err != nil || strings.Join(keys, ",") != "a,b,c" { + t.Fatalf("Keys: %v, %v, want a,b,c", keys, err) + } +}