Artifact [c6dd047436]

Artifact c6dd0474367db79984c9e9fe90059a367d0b4614700f9096418cd56ebb9027da:

• File internal/events/events_test.go — part of check-in [89e4991bbf] at 2026-08-16 03:48:28 on branch trunk — scale-out: external NATS + multi-node fan-out (trio part 1)

  events.Connect joins an external NATS so N nodes share one plane; --nats selects it (embedded stays the single-node default). Retention switched work-queue -> limits+cap: work-queue load-balanced one copy across nodes, leaving each index partial; limits fans out to per-node durables (name now carries a hashed node id) while MaxBytes/MaxAge keep disk bounded. Lease refusals surface as 503 (ops.ErrUnavailable), not 500. Verified live: two nodes on one external NATS + shared store, each folds the full store, the leased node writes while the other 503s and still folds the write cross-node. (user: dmestas, size: 2861) [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)
	}
}

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