myxce/src/frontend/apiserver/backend/heartbeat_test.go

package backend

import (
	"context"
	"sync/atomic"
	"testing"
	"time"
)

func TestSetHeartbeatInterval_ValidValue(t *testing.T) {
	atomic.StoreInt32(&heartbeatIntervalSecs, 30)
	SetHeartbeatInterval(60)
	if got := atomic.LoadInt32(&heartbeatIntervalSecs); got != 60 {
		t.Errorf("heartbeatIntervalSecs = %d, want 60", got)
	}
}

func TestSetHeartbeatInterval_ZeroIgnored(t *testing.T) {
	atomic.StoreInt32(&heartbeatIntervalSecs, 30)
	SetHeartbeatInterval(0)
	if got := atomic.LoadInt32(&heartbeatIntervalSecs); got != 30 {
		t.Errorf("heartbeatIntervalSecs = %d, want 30 (zero should be ignored)", got)
	}
}

func TestSetHeartbeatInterval_NegativeIgnored(t *testing.T) {
	atomic.StoreInt32(&heartbeatIntervalSecs, 30)
	SetHeartbeatInterval(-1)
	if got := atomic.LoadInt32(&heartbeatIntervalSecs); got != 30 {
		t.Errorf("heartbeatIntervalSecs = %d, want 30 (negative should be ignored)", got)
	}
}

// constantReconnectGen returns a currentReconnectGen double that never
// changes value -- used by tests that only care about runHeartbeat's
// tick/cancellation mechanics or fn's own success/failure classification,
// not its reconnect-detection behavior (see the Phase 6 tests below for
// that coverage).
func constantReconnectGen(gen uint64) func() uint64 {
	return func() uint64 { return gen }
}

// noopRejoinHooks are used by tests that only care about runHeartbeat's
// tick/cancellation mechanics, not its reconnect-detected/pending-rejoin
// callback behavior (see the Phase 6 tests below for that coverage).
func noopRejoinHooks() (func(), func()) {
	return func() {}, func() {}
}

func TestRunHeartbeat_CallsFunction(t *testing.T) {
	atomic.StoreInt32(&heartbeatIntervalSecs, 1)

	var callCount int64
	fn := func(_ int) int {
		atomic.AddInt64(&callCount, 1)
		return 0
	}
	onReconnectDetected, onSuccessTick := noopRejoinHooks()

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		runHeartbeat(ctx, fn, constantReconnectGen(0), onReconnectDetected, onSuccessTick)
		close(done)
	}()

	time.Sleep(1500 * time.Millisecond)
	cancel()
	<-done // wait for goroutine to exit before the test returns

	if got := atomic.LoadInt64(&callCount); got < 1 {
		t.Errorf("fn was called %d times, want at least 1", got)
	}
}

func TestRunHeartbeat_CancelsOnContext(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	cancel() // already cancelled

	var callCount int64
	fn := func(_ int) int {
		atomic.AddInt64(&callCount, 1)
		return 0
	}
	onReconnectDetected, onSuccessTick := noopRejoinHooks()

	done := make(chan struct{})
	go func() {
		runHeartbeat(ctx, fn, constantReconnectGen(0), onReconnectDetected, onSuccessTick)
		close(done)
	}()

	select {
	case <-done:
		// returned quickly as expected
	case <-time.After(100 * time.Millisecond):
		t.Error("runHeartbeat did not return promptly after context cancellation")
	}

	if got := atomic.LoadInt64(&callCount); got != 0 {
		t.Errorf("fn was called %d times, want 0 for already-cancelled context", got)
	}
}

// scriptedHeartbeatFn returns a runHeartbeat fn double that yields
// results[0], results[1], ... on successive calls, clamping to the last
// entry for any calls beyond len(results) (defensive against a stray extra
// tick firing before a test's cancel() takes effect -- see the tests below,
// which all stop reading/cancel as soon as the expected callback events are
// observed rather than relying on a fixed sleep window).
func scriptedHeartbeatFn(results []int) func(int) int {
	var idx int32
	return func(_ int) int {
		i := atomic.AddInt32(&idx, 1) - 1
		if int(i) >= len(results) {
			i = int32(len(results) - 1)
		}
		return results[i]
	}
}

// scriptedReconnectGen returns a currentReconnectGen double that yields
// gens[0], gens[1], ... on successive calls (one call per tick -- runHeartbeat
// reads it exactly once per tick, after fn), clamping to the last entry for
// any calls beyond len(gens). The very first call (the pre-loop baseline
// read in runHeartbeat, before any tick) is NOT scripted from this slice --
// callers should include their intended pre-loop baseline as gens[0] and
// note the first in-loop read is gens[1] onward matching a real
// endpoint_t.reconnect_gen starting at a fixed value and changing on later
// reconnects.
func scriptedReconnectGen(gens []uint64) func() uint64 {
	var idx int32
	return func() uint64 {
		i := atomic.AddInt32(&idx, 1) - 1
		if int(i) >= len(gens) {
			i = int32(len(gens) - 1)
		}
		return gens[i]
	}
}

// ════════════════════════════════════════════════════════════════════════
// component-failover-adoption Phase 6 (bug-1548) -- reconnect-generation-
// based idle-period rejoin trigger, replacing task 4.4's original failure-
// to-success transition detector (which structurally could not observe a
// reconnect that completed silently inside a single XCECLI_FF_CALL). These
// tests exercise runHeartbeat's own tick/generation-tracking logic via
// generic injected callbacks, entirely without CGO (the production wiring
// -- reconnect.go's real setNeedsRejoin/checkAndPerformRejoin plus clib.go's
// real currentReconnectGen reading C.xcecli_get_my_endpoint -- is exercised
// end-to-end in heartbeat_rejoin_test.go, //go:build cgo).
// ════════════════════════════════════════════════════════════════════════

// This is the literal shape of bug-1548: every heartbeat tick the Go loop
// observes reports SUCCESS (fn never returns a failure -- XCECLI_FF_CALL's
// silent inline reconnect+recover means the caller's own return code never
// surfaces one), yet a reconnect genuinely happened (the generation counter
// changed). The old failure-to-success transition detector could never fire
// here since it requires an observed failing tick first. This test is the
// actual proof the fix closes the gap the old detector's own green test
// suite could not catch -- it is still, deliberately, only proof of the
// DETECTOR's logic: whether the generation counter is reachable/reliable
// against the real XCECLI_FF_CALL path is proven only by a live E2E re-run
// (component-failover-adoption Phase 6, task 6.2), not by this test.
func TestRunHeartbeat_ReconnectDetected_FiresOnGenChangeEvenWhenAllTicksSucceed(t *testing.T) {
	atomic.StoreInt32(&heartbeatIntervalSecs, 1)

	fn := scriptedHeartbeatFn([]int{0, 0, 0}) // every tick succeeds
	gens := scriptedReconnectGen([]uint64{5, 5, 6, 6})

	events := make(chan string, 8)
	onReconnectDetected := func() { events <- "reconnect_detected" }
	onSuccessTick := func() { events <- "tick" }

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		runHeartbeat(ctx, fn, gens, onReconnectDetected, onSuccessTick)
		close(done)
	}()

	// tick 1: gen 5->5 (no change), tick succeeds -> "tick"
	// tick 2: gen 5->6 (changed) -> "reconnect_detected", then tick succeeds -> "tick"
	want := []string{"tick", "reconnect_detected", "tick"}
	got := make([]string, 0, len(want))
	for range want {
		select {
		case e := <-events:
			got = append(got, e)
		case <-time.After(6 * time.Second):
			t.Fatalf("timed out waiting for heartbeat callback events, got %v so far", got)
		}
	}
	cancel()
	<-done

	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("callback event %d = %q, want %q (full sequence got=%v want=%v) -- "+
				"onReconnectDetected must fire on a reconnect_gen change even though "+
				"every tick's own fn() call reported success",
				i, got[i], want[i], got, want)
		}
	}
}

// A reconnect can complete on a tick whose own send then fails for an
// unrelated reason -- the generation check must not be gated on tick
// success, unlike onSuccessTick.
func TestRunHeartbeat_GenChangeOnFailedTick_StillFiresOnReconnectDetected(t *testing.T) {
	atomic.StoreInt32(&heartbeatIntervalSecs, 1)

	fn := scriptedHeartbeatFn([]int{-1}) // the one tick this test observes fails
	gens := scriptedReconnectGen([]uint64{1, 2})

	events := make(chan string, 8)
	onReconnectDetected := func() { events <- "reconnect_detected" }
	onSuccessTick := func() { events <- "tick" }

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		runHeartbeat(ctx, fn, gens, onReconnectDetected, onSuccessTick)
		close(done)
	}()

	select {
	case e := <-events:
		if e != "reconnect_detected" {
			t.Fatalf("first event = %q, want %q -- a gen change must fire "+
				"onReconnectDetected even on a tick whose own fn() call failed", e, "reconnect_detected")
		}
	case <-time.After(5 * time.Second):
		t.Fatal("timed out waiting for onReconnectDetected")
	}
	cancel()
	<-done

	select {
	case e := <-events:
		t.Fatalf("unexpected extra event %q -- onSuccessTick must not fire on a failed tick", e)
	default:
	}
}

func TestRunHeartbeat_NoGenChange_NeverFiresOnReconnectDetected(t *testing.T) {
	atomic.StoreInt32(&heartbeatIntervalSecs, 1)

	fn := scriptedHeartbeatFn([]int{0, 0, 0})

	var reconnectCalls int32
	onReconnectDetected := func() { atomic.AddInt32(&reconnectCalls, 1) }

	var tickCalls int32
	tickEvents := make(chan struct{}, 8)
	onSuccessTick := func() {
		atomic.AddInt32(&tickCalls, 1)
		tickEvents <- struct{}{}
	}

	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() {
		runHeartbeat(ctx, fn, constantReconnectGen(42), onReconnectDetected, onSuccessTick)
		close(done)
	}()

	for i := 0; i < 3; i++ {
		select {
		case <-tickEvents:
		case <-time.After(5 * time.Second):
			t.Fatalf("timed out waiting for tick %d", i+1)
		}
	}
	cancel()
	<-done

	if got := atomic.LoadInt32(&reconnectCalls); got != 0 {
		t.Errorf("onReconnectDetected called %d times, want 0 -- the generation "+
			"never changed, so no reconnect ever happened", got)
	}
	if got := atomic.LoadInt32(&tickCalls); got < 3 {
		t.Errorf("onSuccessTick called %d times, want at least 3", got)
	}
}