Artifact [512496932c]

Artifact 512496932cd741f56bc04724143d59012240af42fbe8b649ade3545d966a4c40:

• File internal/events/events_test.go — part of check-in [14b0c96b83] at 2026-08-15 20:36:39 on branch trunk — events: work-queue retention on ZITEVT; migrate old streams

  Acked events leave the stream, so disk tracks the in-flight backlog, not store history (was 5.6GB beside sqlite's 218MB). Old-config streams are dropped with their cursors; the census republishes and idempotent folds converge. (user: dmestas, size: 2746) [annotate] [blame] [check-ins using]

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)
	}
}

// Acked events must leave the stream — its disk footprint is the
// in-flight backlog, never the store's history (ticket 151: 5.6GB of
// stream beside a 218MB store).
func TestAckedEventsAreDropped(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.wq.artifact", []byte("e")); err != nil {
			t.Fatal(err)
		}
	}
	sub, err := ev.JS().PullSubscribe("zit.evt.wq.>", "indexer-wq")
	if err != nil {
		t.Fatal(err)
	}
	msgs, err := sub.Fetch(10, nats.MaxWait(2*time.Second))
	if err != nil {
		t.Fatal(err)
	}
	for _, m := range msgs {
		if err := m.AckSync(); err != nil {
			t.Fatal(err)
		}
	}
	info, err := ev.JS().StreamInfo(StreamName)
	if err != nil {
		t.Fatal(err)
	}
	if info.State.Msgs != 0 {
		t.Fatalf("stream retains %d acked events", info.State.Msgs)
	}
}

// An existing stream with the old limits retention is dropped and
// recreated 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 pre-151 layout: recreate the stream with limits
	// retention 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.>"},
	}); 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.WorkQueuePolicy {
		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")
	}
}