Artifact [359ebae54e]

Artifact 359ebae54edd9c207125e35548f7668612db288d60304d9b7521d3e4c0f310cc:

• File internal/events/external_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: 2414) [annotate] [blame] [check-ins using]

package events

import (
	"testing"
	"time"

	natsserver "github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
)

// externalNATS starts a real listening JetStream server (the stand-in
// for an operator's external NATS) and returns its client URL.
func externalNATS(t *testing.T) string {
	t.Helper()
	srv, err := natsserver.NewServer(&natsserver.Options{
		Host:       "127.0.0.1",
		Port:       -1, // OS-assigned
		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("external NATS not ready")
	}
	t.Cleanup(srv.Shutdown)
	return srv.ClientURL()
}

// Two nodes on one external NATS share the event plane: an artifact
// published by one is delivered to the other's Watch, and both see the
// same stream — the substrate multi-node folding and leasing ride on.
func TestExternalNATSSharedPlane(t *testing.T) {
	url := externalNATS(t)

	a, err := Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer a.Close()
	b, err := Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer b.Close()

	// b watches; a publishes; the tick crosses the wire.
	ticks, cancel := b.Watch("repo1")
	defer cancel()
	if _, err := a.JS().Publish("zit.evt.repo1.artifact", []byte("x")); err != nil {
		t.Fatal(err)
	}
	select {
	case <-ticks:
	case <-time.After(5 * time.Second):
		t.Fatal("cross-node event not delivered")
	}

	// Both connected to the same stream (work-queue retention set once).
	info, err := b.JS().StreamInfo(StreamName)
	if err != nil {
		t.Fatal(err)
	}
	if info.Config.Retention != nats.LimitsPolicy {
		t.Fatalf("stream retention = %v", info.Config.Retention)
	}
}

// A large parsed artifact must cross an external NATS too — the server
// is required to allow the 8 MiB payload, matching the embedded default.
func TestExternalNATSLargePayload(t *testing.T) {
	url := externalNATS(t)
	s, err := Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer s.Close()
	big := make([]byte, 3<<20)
	if _, err := s.JS().Publish("zit.evt.big.artifact", big); err != nil {
		t.Fatalf("publish %d bytes over external NATS: %v", len(big), err)
	}
}

// Connect to an unreachable URL fails rather than hanging.
func TestExternalNATSUnreachable(t *testing.T) {
	if _, err := Connect("nats://127.0.0.1:1"); err == nil {
		t.Fatal("connect to dead port should fail")
	}
}