package events
import (
"testing"
"time"
"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)
}
}