package events
import (
"bytes"
"testing"
"time"
"github.com/nats-io/nats.go"
)
// A parsed sqlite baseline manifest is well over NATS's 1MB default
// payload cap; the embedded server must accept multi-megabyte events or
// large stores fail ingest at startup.
func TestLargeArtifactPayload(t *testing.T) {
ev, err := Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
big := bytes.Repeat([]byte("F src/very/long/path.c abcdef\n"), 100_000) // ~3MB
if _, err := ev.JS().Publish("zit.evt.big.artifact", big); err != nil {
t.Fatalf("publish %d bytes: %v", len(big), err)
}
}
// Fan-out: two independent durables each read the FULL stream. This is
// why retention is limits, not work-queue — every node needs a complete
// copy to fold, so one message must reach every node's consumer (a
// shared work-queue would split it, leaving indexes partial).
func TestFanOutToEveryConsumer(t *testing.T) {
ev, err := Start(t.TempDir())
if err != nil {
t.Fatal(err)
}
defer ev.Close()
for i := 0; i < 10; i++ {
if _, err := ev.JS().Publish("zit.evt.fan.artifact", []byte("e")); err != nil {
t.Fatal(err)
}
}
for _, durable := range []string{"indexer-fan-nodeA", "indexer-fan-nodeB"} {
sub, err := ev.JS().PullSubscribe("zit.evt.fan.>", durable)
if err != nil {
t.Fatal(err)
}
got := 0
for got < 10 {
msgs, err := sub.Fetch(10, nats.MaxWait(2*time.Second))
if err != nil {
break
}
for _, m := range msgs {
m.AckSync()
got++
}
}
if got != 10 {
t.Fatalf("%s read %d/10 — fan-out broken", durable, got)
}
}
}
// A work-queue stream (the post-151, pre-multinode config) is migrated
// to limits on Start; the cursor bucket goes with it so the census
// republishes rather than silently skipping the deleted backlog.
func TestRetentionMigration(t *testing.T) {
dir := t.TempDir()
ev, err := Start(dir)
if err != nil {
t.Fatal(err)
}
// Simulate the post-151 layout: a work-queue stream and a stale cursor.
if err := ev.JS().DeleteStream(StreamName); err != nil {
t.Fatal(err)
}
if _, err := ev.JS().AddStream(&nats.StreamConfig{
Name: StreamName, Subjects: []string{"zit.evt.>"}, Retention: nats.WorkQueuePolicy,
}); err != nil {
t.Fatal(err)
}
kv, err := ev.Cursors()
if err != nil {
t.Fatal(err)
}
if _, err := kv.Put("somerepo", []byte(`{"tail":123}`)); err != nil {
t.Fatal(err)
}
ev.Close()
ev2, err := Start(dir)
if err != nil {
t.Fatal(err)
}
defer ev2.Close()
info, err := ev2.JS().StreamInfo(StreamName)
if err != nil {
t.Fatal(err)
}
if info.Config.Retention != nats.LimitsPolicy {
t.Fatalf("retention not migrated: %v", info.Config.Retention)
}
kv2, err := ev2.Cursors()
if err != nil {
t.Fatal(err)
}
if _, err := kv2.Get("somerepo"); err == nil {
t.Fatal("stale cursor survived the stream migration")
}
}
// The embedded store is quarry's budget, not a share of whatever disk
// was free at boot. It must hold a restart and a second node too: NATS
// counts a stream's ceiling again when the stream is declared once more.
// The ceilings, the largest of them twice, must leave the headroom the
// uncapped streams live in, so a new capped stream that does not grow
// the budget fails here rather than on a full-looking disk.
func TestTheStoreIsABudgetNotTheFreeDisk(t *testing.T) {
dir := t.TempDir()
for boot := 0; boot < 3; boot++ {
ev, err := Start(dir)
if err != nil {
t.Fatalf("FAIL: boot %d: %v", boot, err)
}
ev.Close()
}
ev, err := Start(dir)
if err != nil {
t.Fatal(err)
}
defer ev.Close()
if got := ev.srv.JetStreamConfig().MaxStore; got != StoreBudget {
t.Fatalf("FAIL: store budget %d, want %d", got, StoreBudget)
}
var reserved, largest int64
for info := range ev.JS().StreamsInfo() {
if b := info.Config.MaxBytes; b > 0 {
reserved += b
largest = max(largest, b)
}
}
if left := int64(StoreBudget) - reserved - largest; left < storeHeadroom {
t.Fatalf("FAIL: ceilings reserve %d (largest %d) of %d, leaving %d under the %d headroom", reserved, largest, int64(StoreBudget), left, storeHeadroom)
}
t.Logf("ok: %d reserved, %d largest, of %d", reserved, largest, int64(StoreBudget))
}