106 lines
3.4 KiB
Go
106 lines
3.4 KiB
Go
//go:build integration
|
|
|
|
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) {
|
|
js := aLiveBus(t)
|
|
b := Bucket{Module: "statetest", Name: "servers"}
|
|
if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
|
|
t.Fatalf("a real server refused a module's bucket: %v", err)
|
|
}
|
|
kv, err := js.Context().KeyValue(b.Bucket())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := kv.Put("all.one", []byte(`{"kept":true}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.History = 3
|
|
if _, err := RaiseBuckets(js, []Bucket{b}); err != nil {
|
|
t.Fatalf("asserting the bucket again failed, so a restart would: %v", err)
|
|
}
|
|
got, err := kv.Get("all.one")
|
|
if err != nil || string(got.Value()) != `{"kept":true}` {
|
|
t.Fatalf("asserting again lost what the bucket held: %v %v", got, err)
|
|
}
|
|
status, err := kv.Status()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if status.History() != 3 {
|
|
t.Fatalf("history is %d, the owner declared 3", status.History())
|
|
}
|
|
if s, ok := status.(*nats.KeyValueBucketStatus); ok {
|
|
if c := s.StreamInfo().Config; c.MaxMsgSize != StateMaxValueBytes || c.MaxBytes != StateMaxBytes {
|
|
t.Fatalf("the mesh's caps are not on the bucket: value %d, bucket %d", c.MaxMsgSize, c.MaxBytes)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestABucketIsAssertedOverAHeldConnection(t *testing.T) {
|
|
js := aLiveBus(t)
|
|
held := OnConn(js.Conn())
|
|
if _, err := RaiseBuckets(held, []Bucket{{Module: "statetest", Name: "held"}}); err != nil {
|
|
t.Fatalf("asserting over a held connection failed: %v", err)
|
|
}
|
|
if _, err := js.Context().KeyValue("statetest_held"); err != nil {
|
|
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)
|
|
}
|
|
}
|