Artifact [d4c27b2866]

Artifact d4c27b2866a747b34c464de85fe6c5403b9d429da2ffb66bc8a3916ad35e259e:

• File internal/events/gap_test.go — part of check-in [f08add75f5] at 2026-08-24 15:50:54 on branch trunk — events: the vocabulary as code, and what the bus really does

  RFC 0005 §1 as internal/evt, named at the eight sites that raise events instead of sliced out of a TxnID — so a ticket change that closes one is ticket.closed, not ticket. Measured the window: a gap IS detectable (first_seq - ack_floor) but the next fetch succeeds silently, so every consumer must check. dagnats has no scheduler; quarry owns the clock. (user: dmestas, size: 3311) [annotate] [blame] [check-ins using]

package events

import (
	"testing"
	"time"

	"github.com/nats-io/nats.go"
)

// RFC 0005 §2 claims a consumer outside the stream's window has missed events
// permanently, and §2.1 divides the vocabulary on whether that is survivable.
// Both claims rest on what JetStream actually does when MaxAge drops a message
// a durable consumer has not read — which the documentation does not state, so
// it is measured here rather than assumed.
//
// The question that decides the design: can a consumer TELL that it missed
// something? A consumer that can detect a gap can rebuild from the store; one
// that cannot will serve a silently partial view forever.
func TestAConsumerCanDetectMessagesItMissed(t *testing.T) {
	srv, err := Start(t.TempDir())
	if err != nil {
		t.Skip("embedded NATS unavailable:", err)
	}
	defer srv.Close()
	js := srv.JS()

	const subject = "gap.test"
	if _, err := js.AddStream(&nats.StreamConfig{
		Name:      "GAPTEST",
		Subjects:  []string{subject},
		Retention: nats.LimitsPolicy, // quarry's policy
		Discard:   nats.DiscardOld,   // quarry's policy
		MaxAge:    600 * time.Millisecond,
	}); err != nil {
		t.Fatal(err)
	}
	defer js.DeleteStream("GAPTEST")

	sub, err := js.PullSubscribe(subject, "gapdurable")
	if err != nil {
		t.Fatal(err)
	}
	// Three messages the consumer reads, then three it sleeps through.
	for i := 0; i < 3; i++ {
		if _, err := js.Publish(subject, []byte("early")); err != nil {
			t.Fatal(err)
		}
	}
	msgs, err := sub.Fetch(3, nats.MaxWait(2*time.Second))
	if err != nil {
		t.Fatal(err)
	}
	for _, m := range msgs {
		m.Ack()
	}
	for i := 0; i < 3; i++ {
		if _, err := js.Publish(subject, []byte("missed")); err != nil {
			t.Fatal(err)
		}
	}
	// Outlive the window without reading them.
	time.Sleep(900 * time.Millisecond)
	if _, err := js.Publish(subject, []byte("after the gap")); err != nil {
		t.Fatal(err)
	}

	st, err := js.StreamInfo("GAPTEST")
	if err != nil {
		t.Fatal(err)
	}
	ci, err := sub.ConsumerInfo()
	if err != nil {
		t.Fatal(err)
	}

	// The finding, whichever way it goes, is what RFC 0005 §2 rests on.
	t.Logf("stream first_seq=%d last_seq=%d msgs=%d",
		st.State.FirstSeq, st.State.LastSeq, st.State.Msgs)
	t.Logf("consumer ack_floor=%d delivered=%d pending=%d",
		ci.AckFloor.Stream, ci.Delivered.Stream, ci.NumPending)

	// A gap is visible when the stream's oldest surviving message is
	// beyond the next one this consumer expected.
	gap := int(st.State.FirstSeq) - int(ci.AckFloor.Stream) - 1
	if gap <= 0 {
		t.Fatalf("no gap detected after messages aged out: first_seq=%d ack_floor=%d — "+
			"if this is right, RFC 0005 §2 is wrong about the window being lossy",
			st.State.FirstSeq, ci.AckFloor.Stream)
	}
	t.Logf("DETECTABLE: %d message(s) aged out unread, visible as first_seq - ack_floor - 1", gap)

	// And the next fetch returns the message AFTER the gap rather than
	// erroring, so a consumer that does not compare sequences carries on
	// silently with a hole in its view.
	next, err := sub.Fetch(1, nats.MaxWait(2*time.Second))
	if err != nil {
		t.Fatalf("fetch after gap: %v", err)
	}
	if got := string(next[0].Data); got != "after the gap" {
		t.Errorf("after a gap the consumer got %q", got)
	}
	next[0].Ack()
	t.Log("SILENT: the fetch after a gap succeeds and returns the newer message")
}