package events
import (
"testing"
"time"
"github.com/nats-io/nats-server/v2/server"
"github.com/nats-io/nats.go"
)
// A stream an older binary created keeps its limits unless Start
// updates it: adding a stream that exists is a no-op, which is how
// production ran on a window far tighter than this code declared. The
// update keeps every message and the cursors; only a retention change
// recreates the stream.
func TestStartRaisesAnExistingStreamsLimits(t *testing.T) {
dir := t.TempDir()
ev, err := Start(dir)
if err != nil {
t.Fatal(err)
}
info, err := ev.JS().StreamInfo(StreamName)
if err != nil {
t.Fatal(err)
}
old := info.Config
old.MaxAge, old.MaxBytes, old.MaxMsgs = 30*time.Minute, 1<<20, 50
if _, err := ev.JS().UpdateStream(&old); err != nil {
t.Fatal(err)
}
for i := 0; i < 3; i++ {
if _, err := ev.JS().Publish("zit.evt.kept.artifact", []byte("e")); err != nil {
t.Fatal(err)
}
}
kv, err := ev.Cursors()
if err != nil {
t.Fatal(err)
}
if _, err := kv.Put("kept", []byte(`{}`)); err != nil {
t.Fatal(err)
}
created := info.Created
ev.Close()
ev, err = Start(dir)
if err != nil {
t.Fatal(err)
}
defer ev.Close()
info, err = ev.JS().StreamInfo(StreamName)
if err != nil {
t.Fatal(err)
}
c := info.Config
if c.MaxAge != WindowAge || c.MaxBytes != WindowBytes || c.MaxMsgs != WindowMsgs || c.Discard != nats.DiscardOld {
t.Errorf("FAIL: limits not applied to the existing stream: age=%s bytes=%d msgs=%d discard=%v",
c.MaxAge, c.MaxBytes, c.MaxMsgs, c.Discard)
}
if info.State.Msgs != 3 || !info.Created.Equal(created) {
t.Errorf("FAIL: the update lost the stream: msgs=%d created=%s (was %s)", info.State.Msgs, info.Created, created)
}
kv, err = ev.Cursors()
if err != nil {
t.Fatal(err)
}
if _, err := kv.Get("kept"); err != nil {
t.Errorf("FAIL: a limits update dropped the cursors: %v", err)
}
}
// The window readout is what the boot log and the ops page print: the
// stream's first and last sequence and when its oldest message landed.
func TestWindowReportsTheStreamsSpan(t *testing.T) {
ev, err := Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
w, err := ev.Window()
if err != nil {
t.Fatal(err)
}
if w.Msgs != 0 || !w.Oldest.IsZero() || w.MaxAge != WindowAge {
t.Errorf("FAIL: an empty stream's window: %+v", w)
}
start := time.Now()
for i := 0; i < 3; i++ {
if _, err := ev.JS().Publish("zit.evt.span.artifact", []byte("e")); err != nil {
t.Fatal(err)
}
}
w, err = ev.Window()
if err != nil {
t.Fatal(err)
}
if w.FirstSeq != 1 || w.LastSeq != 3 || w.Msgs != 3 || w.Bytes == 0 {
t.Errorf("FAIL: window span: %+v", w)
}
if w.Oldest.Before(start.Add(-time.Second)) || w.Oldest.After(time.Now()) {
t.Errorf("FAIL: oldest message time %s outside the publish", w.Oldest)
}
}
// A stream the update cannot change (here, one kept in memory, whose
// storage JetStream will not convert) must not stop the forge: it ran on
// those limits before, and the boot log and ops page show the live ones.
func TestAStreamThatRefusesTheUpdateStillStarts(t *testing.T) {
srv, err := server.NewServer(&server.Options{
Host: "127.0.0.1", Port: -1, JetStream: true,
StoreDir: t.TempDir(), MaxPayload: 8 << 20,
})
if err != nil {
t.Fatal(err)
}
go srv.Start()
if !srv.ReadyForConnections(10 * time.Second) {
t.Fatal("nats not ready")
}
defer srv.Shutdown()
nc, err := nats.Connect(srv.ClientURL())
if err != nil {
t.Fatal(err)
}
defer nc.Close()
js, err := nc.JetStream()
if err != nil {
t.Fatal(err)
}
if _, err := js.AddStream(&nats.StreamConfig{Name: StreamName, Subjects: []string{"zit.evt.>"},
Storage: nats.MemoryStorage, MaxAge: 30 * time.Minute}); err != nil {
t.Fatal(err)
}
ev, err := Connect(srv.ClientURL())
if err != nil {
t.Fatalf("FAIL: a stream the update could not change stopped the start: %v", err)
}
defer ev.Close()
w, err := ev.Window()
if err != nil {
t.Fatal(err)
}
if w.MaxAge != 30*time.Minute {
t.Errorf("FAIL: the window does not report the live limit: %s", w.MaxAge)
}
}