fix(component-failover-adoption): Phase 6 -- close apiserver's idle-period rejoin gap (bug-1548/bug-1549)
XCECLI_FF_CALL (backing xcecli_heartbeat) can silently rotate+reconnect the transport entirely inside a single call and still return success, so heartbeat.go's failure-to-success transition detector could never observe a reconnect that completes within one heartbeat tick -- the common case, since elections finish in single-digit seconds against a default 30s interval. apiserver's application-level rejoin never fired on a real idle-period failover, and job pushes silently stopped reaching it. Adds a monotonic reconnect_gen counter to endpoint_t (netio.h), bumped by xce_ctx_reconnect() (xcectx.c) on every successful reconnect regardless of trigger path. heartbeat.go now compares this generation every tick instead of tracking success/failure transitions. Re-validated live against task 5.2's exact scenario (apiserver alone, zero write traffic): rejoin fired 22s after a primary kill, 12s after election. Reopens and closes out component-failover-adoption's own NO-GO -> GO validation cycle from 2026-08-15. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
parent
0cb5918517
commit
d479b70764
SSH key fingerprint:
SHA256:8Vf2pZ3h6rJa7EhGK+WaAQ9Mta/apZvteW6g9U/WKKk
11 changed files with 430 additions and 146 deletions
|
|
@ -83,6 +83,7 @@ import (
|
|||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
"unsafe"
|
||||
|
||||
|
|
@ -350,19 +351,35 @@ func Cleanup() {
|
|||
C.idgen_ctx_fini(&gIDCtx)
|
||||
}
|
||||
|
||||
// currentReconnectGen reads endpoint_t.reconnect_gen (netio.h) off the
|
||||
// backend's live endpoint via the existing, previously-unused C accessor
|
||||
// xcecli_get_my_endpoint() -- no new C getter needed (component-failover-
|
||||
// adoption Phase 6, bug-1548). Uses an explicit atomic load (rather than a
|
||||
// bare field dereference) so the Go-side read is formally atomic and pairs
|
||||
// correctly with the C11 _Atomic write in xce_ctx_reconnect() (xcectx.c).
|
||||
func currentReconnectGen() uint64 {
|
||||
ep := C.xcecli_get_my_endpoint(gCtx.ctx)
|
||||
if ep == nil {
|
||||
return 0
|
||||
}
|
||||
return atomic.LoadUint64((*uint64)(unsafe.Pointer(&ep.reconnect_gen)))
|
||||
}
|
||||
|
||||
// StartHeartbeat launches a goroutine that sends periodic xcecli_heartbeat
|
||||
// keep-alive messages to the controller. Stops when ctx is cancelled.
|
||||
//
|
||||
// Wires runHeartbeat's failure-to-success and every-successful-tick hooks to
|
||||
// Wires runHeartbeat's reconnect-detected and every-successful-tick hooks to
|
||||
// reconnect.go's setNeedsRejoin/checkAndPerformRejoin (task 4.3) -- closing
|
||||
// the idle-period reconnect gap (task 4.4, Req 2 AC4/AC5): XCECLI_FF_CALL
|
||||
// (backing xcecli_heartbeat) never invokes the client library's rejoin
|
||||
// callback, so nothing else would set the pending-rejoin flag if this
|
||||
// heartbeat prober is the only traffic during a controller outage.
|
||||
// the idle-period reconnect gap (task 4.4/Phase 6, Req 2 AC4/AC5):
|
||||
// XCECLI_FF_CALL (backing xcecli_heartbeat) never invokes the client
|
||||
// library's rejoin callback, so nothing else would set the pending-rejoin
|
||||
// flag if this heartbeat prober is the only traffic during a controller
|
||||
// outage. currentReconnectGen (above) is what actually detects the
|
||||
// reconnect even when it completed silently inside a single XCECLI_FF_CALL.
|
||||
func StartHeartbeat(ctx context.Context) {
|
||||
go runHeartbeat(ctx, func(secs int) int {
|
||||
return int(C.xcecli_heartbeat(gCtx.ctx, C.int(secs)))
|
||||
}, setNeedsRejoin, checkAndPerformRejoin)
|
||||
}, currentReconnectGen, setNeedsRejoin, checkAndPerformRejoin)
|
||||
}
|
||||
|
||||
// go_push_handler is called by the C library on every incoming request.
|
||||
|
|
|
|||
|
|
@ -26,55 +26,68 @@ func SetHeartbeatInterval(secs int) {
|
|||
// consistent with this codebase's established "rc < 0 is the only failure"
|
||||
// join/rejoin convention (tasks.md Implementation Notes, task 2.3).
|
||||
//
|
||||
// component-failover-adoption task 4.4 (boundary: ApiserverBackendBridge,
|
||||
// Req 2 AC4/AC5): closes the idle-period reconnect gap. XCECLI_FF_CALL --
|
||||
// the macro backing xcecli_heartbeat -- performs a best-effort transport
|
||||
// reconnect on a known-dead connection but never invokes the client
|
||||
// library's rejoin callback (xcecli_invoke_rejoin is reached only from
|
||||
// xcecli_retry_next, the full-retry write path task 4.3 wired up) -- so
|
||||
// nothing else sets the pending-rejoin flag if this heartbeat prober is the
|
||||
// ONLY traffic during a controller outage (apiserver idle, no in-flight
|
||||
// write request).
|
||||
// component-failover-adoption Phase 6 (bug-1548): replaces task 4.4's
|
||||
// original failure-to-success TRANSITION detector, which could never
|
||||
// observe the idle-period reconnect gap it was built to close.
|
||||
// XCECLI_FF_CALL -- the macro backing xcecli_heartbeat -- can rotate +
|
||||
// reconnect the transport ENTIRELY INSIDE a single call (while holding
|
||||
// ac->failover.lock), before evaluating its own EXPR, and still return
|
||||
// success. So a two-tick fail-then-succeed edge on fn's own return value is
|
||||
// not a reliable "a reconnect happened" signal: whenever a controller
|
||||
// election completes faster than the heartbeat interval -- the common
|
||||
// case, elections finish in single-digit seconds against a default 30s
|
||||
// interval -- the whole reconnect+recovery happens silently within one
|
||||
// tick and no failing tick is ever observed by this loop.
|
||||
//
|
||||
// Tracks success/failure across ticks in a single local variable --
|
||||
// single-goroutine state, no atomic needed (unlike reconnect.go's
|
||||
// needsRejoin, which is genuinely shared across concurrent HTTP request
|
||||
// goroutines). On a genuine failure-to-success TRANSITION (previous tick
|
||||
// failed, this tick succeeded), onFailureToSuccess is called once --
|
||||
// wired to reconnect.go's setNeedsRejoin in production. On EVERY tick that
|
||||
// succeeds -- not only the transition tick -- onSuccessTick is called,
|
||||
// wired to checkAndPerformRejoin (task 4.3), which is itself a cheap no-op
|
||||
// once no rejoin is pending: this is what makes a rejoin attempt that
|
||||
// itself fails get retried on the next successful tick rather than
|
||||
// abandoned (Req 2 AC5). A failed tick does nothing beyond updating the
|
||||
// local failure-tracking state -- no rejoin logic runs on a failed tick.
|
||||
// currentReconnectGen reads endpoint_t.reconnect_gen (src/include/netio.h),
|
||||
// a monotonic counter bumped by xce_ctx_reconnect() (src/lib/xcectx.c) on
|
||||
// every successful transport reconnect, regardless of trigger path (write-
|
||||
// path XCECLI_FAILOVER_CALL retry or this heartbeat's own XCECLI_FF_CALL
|
||||
// best-effort self-heal) or whether the triggering call's own return code
|
||||
// ever surfaced a failure. Checked on EVERY tick, independent of that
|
||||
// tick's own fn() outcome -- a reconnect can complete on a tick whose own
|
||||
// send then fails for an unrelated reason, and that reconnect still needs
|
||||
// to be observed. Any change since the last-observed value means a
|
||||
// reconnect happened since this loop last checked, and fires
|
||||
// onReconnectDetected exactly once per change -- wired to reconnect.go's
|
||||
// setNeedsRejoin in production.
|
||||
//
|
||||
// Both callbacks are injected (rather than referenced by name) so this
|
||||
// loop's own tick/transition-tracking logic stays testable without CGO --
|
||||
// see heartbeat_test.go for that coverage and heartbeat_rejoin_test.go
|
||||
// (//go:build cgo) for the end-to-end wiring against the real
|
||||
// setNeedsRejoin/checkAndPerformRejoin. clib.go's StartHeartbeat wires the
|
||||
// real production callbacks (those two symbols only exist in the CGO
|
||||
// build).
|
||||
func runHeartbeat(ctx context.Context, fn func(int) int, onFailureToSuccess, onSuccessTick func()) {
|
||||
// onSuccessTick fires on every tick whose own fn() call succeeds -- wired
|
||||
// to checkAndPerformRejoin (task 4.3), which is itself a cheap no-op once
|
||||
// no rejoin is pending: this is what makes a rejoin attempt that itself
|
||||
// fails get retried on the next successful tick rather than abandoned
|
||||
// (Req 2 AC5). Gated on tick success, unlike the generation check above:
|
||||
// attempting a rejoin against a connection that just failed to even
|
||||
// heartbeat is not expected to succeed.
|
||||
//
|
||||
// currentReconnectGen and both callbacks are injected (rather than
|
||||
// referenced by name) so this loop's own tick-tracking logic stays
|
||||
// testable without CGO -- see heartbeat_test.go for that coverage and
|
||||
// heartbeat_rejoin_test.go (//go:build cgo) for the end-to-end wiring
|
||||
// against the real setNeedsRejoin/checkAndPerformRejoin. clib.go's
|
||||
// StartHeartbeat wires the real production callbacks and a real
|
||||
// currentReconnectGen reading C.xcecli_get_my_endpoint(gCtx.ctx).
|
||||
func runHeartbeat(ctx context.Context, fn func(int) int, currentReconnectGen func() uint64, onReconnectDetected, onSuccessTick func()) {
|
||||
interval := int(atomic.LoadInt32(&heartbeatIntervalSecs))
|
||||
t := time.NewTimer(time.Duration(interval) * time.Second)
|
||||
defer t.Stop()
|
||||
|
||||
prevFailed := false
|
||||
lastGen := currentReconnectGen()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
succeeded := fn(interval) >= 0
|
||||
|
||||
if gen := currentReconnectGen(); gen != lastGen {
|
||||
lastGen = gen
|
||||
onReconnectDetected()
|
||||
}
|
||||
|
||||
if succeeded {
|
||||
if prevFailed {
|
||||
onFailureToSuccess()
|
||||
}
|
||||
onSuccessTick()
|
||||
}
|
||||
prevFailed = !succeeded
|
||||
}
|
||||
interval = int(atomic.LoadInt32(&heartbeatIntervalSecs))
|
||||
t.Reset(time.Duration(interval) * time.Second)
|
||||
|
|
|
|||
|
|
@ -11,13 +11,18 @@ import (
|
|||
|
||||
// heartbeat_rejoin_test.go exercises runHeartbeat wired to reconnect.go's
|
||||
// REAL setNeedsRejoin/checkAndPerformRejoin (component-failover-adoption
|
||||
// task 4.4, boundary: ApiserverBackendBridge, Req 2 AC4/AC5) -- proving the
|
||||
// idle-period rejoin trigger end-to-end at real tick granularity, the same
|
||||
// way clib.go's StartHeartbeat wires them in production. heartbeat_test.go
|
||||
// covers runHeartbeat's own tick/transition-tracking logic via generic
|
||||
// injected callbacks, without CGO; this file proves those callbacks are
|
||||
// wired to the right production functions and that the composition
|
||||
// actually clears/retains needsRejoin correctly across real ticks.
|
||||
// task 4.4/Phase 6, boundary: ApiserverBackendBridge, Req 2 AC4/AC5) --
|
||||
// proving the idle-period rejoin trigger end-to-end at real tick
|
||||
// granularity, the same way clib.go's StartHeartbeat wires them in
|
||||
// production (currentReconnectGen itself -- the real
|
||||
// C.xcecli_get_my_endpoint(gCtx.ctx) read -- is NOT exercised here; these
|
||||
// tests inject a scripted currentReconnectGen double, same as
|
||||
// heartbeat_test.go, since gCtx has no live connection in a unit test).
|
||||
// heartbeat_test.go covers runHeartbeat's own tick/generation-tracking
|
||||
// logic via generic injected callbacks, without CGO; this file proves
|
||||
// those callbacks are wired to the right production functions and that the
|
||||
// composition actually clears/retains needsRejoin correctly across real
|
||||
// ticks.
|
||||
//
|
||||
// Kept //go:build cgo since needsRejoin/setNeedsRejoin/checkAndPerformRejoin/
|
||||
// performRejoinFn only exist in the CGO build (reconnect.go, task 4.3) --
|
||||
|
|
@ -28,10 +33,24 @@ import (
|
|||
// performRejoinFn (reconnect.go's existing test seam, task 4.3) is
|
||||
// overridden here to script the actual xcecli_join outcome without a live
|
||||
// controller -- same technique reconnect_test.go's own retry-on-failure
|
||||
// tests use. scriptedHeartbeatFn (heartbeat_test.go) scripts the heartbeat
|
||||
// send outcomes.
|
||||
// tests use. scriptedHeartbeatFn/scriptedReconnectGen (heartbeat_test.go)
|
||||
// script the heartbeat send outcomes and the reconnect-generation reads.
|
||||
|
||||
func TestRunHeartbeat_IdlePeriodRecovery_SetsNeedsRejoinAndRetriesUntilJoinSucceeds(t *testing.T) {
|
||||
// This is the CGO-level counterpart to
|
||||
// TestRunHeartbeat_ReconnectDetected_FiresOnGenChangeEvenWhenAllTicksSucceed
|
||||
// in heartbeat_test.go, and is the actual proof of closure for bug-1548 at
|
||||
// the level production code runs at: every heartbeat tick's own fn() call
|
||||
// reports SUCCESS (the literal shape XCECLI_FF_CALL's silent inline
|
||||
// reconnect+recover produces -- no failing tick is ever observed), yet a
|
||||
// reconnect genuinely happened (the generation counter changed). The old
|
||||
// failure-to-success transition detector could never set needsRejoin in
|
||||
// this scenario; this test proves the replacement generation-based
|
||||
// detector does. Still only proof of the DETECTOR's logic wired to the
|
||||
// real production callbacks -- whether currentReconnectGen's real
|
||||
// C.xcecli_get_my_endpoint read is itself reachable/reliable against a real
|
||||
// XCECLI_FF_CALL reconnect is proven only by the required live E2E re-run
|
||||
// of task 5.2's scenario (Phase 6, task 6.2), not by any unit test.
|
||||
func TestRunHeartbeat_IdlePeriodRecovery_GenChangeAloneSetsNeedsRejoinEvenWhenAllTicksSucceed(t *testing.T) {
|
||||
captureLog(t) // checkAndPerformRejoin logs via utils.Log; nil otherwise
|
||||
resetNeedsRejoinForTest()
|
||||
atomic.StoreInt32(&heartbeatIntervalSecs, 1)
|
||||
|
|
@ -39,6 +58,53 @@ func TestRunHeartbeat_IdlePeriodRecovery_SetsNeedsRejoinAndRetriesUntilJoinSucce
|
|||
origPerformRejoinFn := performRejoinFn
|
||||
defer func() { performRejoinFn = origPerformRejoinFn }()
|
||||
|
||||
var joinCalls int32
|
||||
joinDone := make(chan struct{}, 8)
|
||||
performRejoinFn = func() int {
|
||||
atomic.AddInt32(&joinCalls, 1)
|
||||
joinDone <- struct{}{}
|
||||
return 0 // succeeds on the first attempt
|
||||
}
|
||||
|
||||
// Every heartbeat tick this test observes reports success -- no failure
|
||||
// ever surfaces to Go, matching XCECLI_FF_CALL's silent self-heal.
|
||||
fn := scriptedHeartbeatFn([]int{0, 0, 0})
|
||||
// Generation changes on the second tick (pre-loop baseline=5, tick1=5
|
||||
// unchanged, tick2=6 changed) -- a reconnect happened even though the
|
||||
// caller never saw a failing tick.
|
||||
gens := scriptedReconnectGen([]uint64{5, 5, 6, 6})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
runHeartbeat(ctx, fn, gens, setNeedsRejoin, checkAndPerformRejoin)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-joinDone:
|
||||
case <-time.After(6 * time.Second):
|
||||
t.Fatal("timed out waiting for the rejoin attempt triggered by the generation change")
|
||||
}
|
||||
cancel()
|
||||
<-done
|
||||
|
||||
if got := atomic.LoadInt32(&joinCalls); got != 1 {
|
||||
t.Errorf("performRejoinFn called %d times, want exactly 1", got)
|
||||
}
|
||||
if got := atomic.LoadInt32(&needsRejoin); got != 0 {
|
||||
t.Error("expected needsRejoin to clear once the rejoin succeeds")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunHeartbeat_RejoinRetriesOnFailureUntilItSucceeds(t *testing.T) {
|
||||
captureLog(t)
|
||||
resetNeedsRejoinForTest()
|
||||
atomic.StoreInt32(&heartbeatIntervalSecs, 1)
|
||||
|
||||
origPerformRejoinFn := performRejoinFn
|
||||
defer func() { performRejoinFn = origPerformRejoinFn }()
|
||||
|
||||
var joinCalls int32
|
||||
joinDone := make(chan struct{}, 8)
|
||||
performRejoinFn = func() int {
|
||||
|
|
@ -50,22 +116,20 @@ func TestRunHeartbeat_IdlePeriodRecovery_SetsNeedsRejoinAndRetriesUntilJoinSucce
|
|||
return 0 // succeeds on the second, separate attempt
|
||||
}
|
||||
|
||||
// Heartbeat transport outcomes: fail (primary down), fail, succeed
|
||||
// (transparent reconnect -- fires the transition), succeed (retry
|
||||
// tick, while the first rejoin attempt is still pending).
|
||||
fn := scriptedHeartbeatFn([]int{-1, -1, 0, 0})
|
||||
fn := scriptedHeartbeatFn([]int{0, 0, 0})
|
||||
// Generation changes once, on tick 1 (baseline=1, tick1=2 changed,
|
||||
// stays 2 thereafter) -- one reconnect, but the resulting rejoin must
|
||||
// be retried across multiple successful ticks until it succeeds (Req 2
|
||||
// AC5), not abandoned after its first failed attempt.
|
||||
gens := scriptedReconnectGen([]uint64{1, 2, 2, 2})
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
runHeartbeat(ctx, fn, setNeedsRejoin, checkAndPerformRejoin)
|
||||
runHeartbeat(ctx, fn, gens, setNeedsRejoin, checkAndPerformRejoin)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
// Wait for both scripted rejoin attempts: one triggered by the
|
||||
// transition tick (fails), one retried on the next successful tick
|
||||
// (succeeds) -- proving the failed rejoin is retried, not abandoned
|
||||
// (Req 2 AC5), at real heartbeat-tick granularity.
|
||||
for i := 0; i < 2; i++ {
|
||||
select {
|
||||
case <-joinDone:
|
||||
|
|
@ -85,7 +149,7 @@ func TestRunHeartbeat_IdlePeriodRecovery_SetsNeedsRejoinAndRetriesUntilJoinSucce
|
|||
}
|
||||
}
|
||||
|
||||
func TestRunHeartbeat_AllHeartbeatsSucceed_NeverSetsNeedsRejoin(t *testing.T) {
|
||||
func TestRunHeartbeat_NoGenChange_NeverSetsNeedsRejoin(t *testing.T) {
|
||||
resetNeedsRejoinForTest()
|
||||
atomic.StoreInt32(&heartbeatIntervalSecs, 1)
|
||||
|
||||
|
|
@ -101,13 +165,13 @@ func TestRunHeartbeat_AllHeartbeatsSucceed_NeverSetsNeedsRejoin(t *testing.T) {
|
|||
tickEvents := make(chan struct{}, 8)
|
||||
fn := func(_ int) int {
|
||||
tickEvents <- struct{}{}
|
||||
return 0 // always succeeds -- no failure ever observed
|
||||
return 0 // always succeeds
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
runHeartbeat(ctx, fn, setNeedsRejoin, checkAndPerformRejoin)
|
||||
runHeartbeat(ctx, fn, constantReconnectGen(9), setNeedsRejoin, checkAndPerformRejoin)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
|
|
@ -122,8 +186,8 @@ func TestRunHeartbeat_AllHeartbeatsSucceed_NeverSetsNeedsRejoin(t *testing.T) {
|
|||
<-done
|
||||
|
||||
if got := atomic.LoadInt32(&needsRejoin); got != 0 {
|
||||
t.Error("needsRejoin must stay clear when every heartbeat tick succeeds -- " +
|
||||
"no failure ever occurred to transition from")
|
||||
t.Error("needsRejoin must stay clear when the reconnect generation never changes -- " +
|
||||
"no reconnect ever occurred")
|
||||
}
|
||||
if got := atomic.LoadInt32(&joinCalls); got != 0 {
|
||||
t.Errorf("performRejoinFn called %d times, want 0 (checkAndPerformRejoin "+
|
||||
|
|
|
|||
|
|
@ -31,10 +31,18 @@ func TestSetHeartbeatInterval_NegativeIgnored(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// 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 failure-to-success/pending-rejoin
|
||||
// callback behavior (see TestRunHeartbeat_FailureToSuccessTransition_* etc.
|
||||
// below for that coverage).
|
||||
// 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() {}
|
||||
}
|
||||
|
|
@ -47,12 +55,12 @@ func TestRunHeartbeat_CallsFunction(t *testing.T) {
|
|||
atomic.AddInt64(&callCount, 1)
|
||||
return 0
|
||||
}
|
||||
onFailureToSuccess, onSuccessTick := noopRejoinHooks()
|
||||
onReconnectDetected, onSuccessTick := noopRejoinHooks()
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick)
|
||||
runHeartbeat(ctx, fn, constantReconnectGen(0), onReconnectDetected, onSuccessTick)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
|
|
@ -74,11 +82,11 @@ func TestRunHeartbeat_CancelsOnContext(t *testing.T) {
|
|||
atomic.AddInt64(&callCount, 1)
|
||||
return 0
|
||||
}
|
||||
onFailureToSuccess, onSuccessTick := noopRejoinHooks()
|
||||
onReconnectDetected, onSuccessTick := noopRejoinHooks()
|
||||
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick)
|
||||
runHeartbeat(ctx, fn, constantReconnectGen(0), onReconnectDetected, onSuccessTick)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
|
|
@ -111,80 +119,69 @@ func scriptedHeartbeatFn(results []int) func(int) int {
|
|||
}
|
||||
}
|
||||
|
||||
// ════════════════════════════════════════════════════════════════════════
|
||||
// component-failover-adoption task 4.4 (Req 2 AC4/AC5) -- idle-period
|
||||
// rejoin trigger. These tests exercise runHeartbeat's own tick/transition-
|
||||
// tracking logic via generic injected callbacks, entirely without CGO (the
|
||||
// production wiring -- reconnect.go's real setNeedsRejoin/
|
||||
// checkAndPerformRejoin -- is exercised end-to-end in
|
||||
// heartbeat_rejoin_test.go, //go:build cgo).
|
||||
// ════════════════════════════════════════════════════════════════════════
|
||||
|
||||
func TestRunHeartbeat_FailureToSuccessTransition_FiresOnFailureToSuccessThenOnSuccessTick(t *testing.T) {
|
||||
atomic.StoreInt32(&heartbeatIntervalSecs, 1)
|
||||
|
||||
// fail, fail, succeed -- the transition happens on tick 3.
|
||||
fn := scriptedHeartbeatFn([]int{-1, -1, 0})
|
||||
|
||||
events := make(chan string, 8)
|
||||
onFailureToSuccess := func() { events <- "failure_to_success" }
|
||||
onSuccessTick := func() { events <- "success_tick" }
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
want := []string{"failure_to_success", "success_tick"}
|
||||
got := make([]string, 0, len(want))
|
||||
for range want {
|
||||
select {
|
||||
case e := <-events:
|
||||
got = append(got, e)
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatalf("timed out waiting for heartbeat callback events, got %v so far", got)
|
||||
}
|
||||
}
|
||||
// Cancel immediately once the expected events are observed, before any
|
||||
// further tick can fire -- keeps the assertion below exact regardless
|
||||
// of scheduling slop.
|
||||
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) -- "+
|
||||
"onFailureToSuccess must fire exactly once, before onSuccessTick, "+
|
||||
"only on the fail-then-succeed transition tick",
|
||||
i, got[i], want[i], got, want)
|
||||
// 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]
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunHeartbeat_EverySuccessfulTick_RetriesWhilePending(t *testing.T) {
|
||||
// ════════════════════════════════════════════════════════════════════════
|
||||
// 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)
|
||||
|
||||
// fail, then succeed 3 times in a row: onFailureToSuccess must fire
|
||||
// only on the first success (the transition), but onSuccessTick must
|
||||
// fire on ALL THREE successes -- proving the pending-rejoin check is
|
||||
// retried on every successful tick, not abandoned after the transition
|
||||
// tick (Req 2 AC5: "checked every tick, not only the transition edge").
|
||||
fn := scriptedHeartbeatFn([]int{-1, 0, 0, 0})
|
||||
fn := scriptedHeartbeatFn([]int{0, 0, 0}) // every tick succeeds
|
||||
gens := scriptedReconnectGen([]uint64{5, 5, 6, 6})
|
||||
|
||||
events := make(chan string, 8)
|
||||
onFailureToSuccess := func() { events <- "f2s" }
|
||||
onReconnectDetected := func() { events <- "reconnect_detected" }
|
||||
onSuccessTick := func() { events <- "tick" }
|
||||
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick)
|
||||
runHeartbeat(ctx, fn, gens, onReconnectDetected, onSuccessTick)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
want := []string{"f2s", "tick", "tick", "tick"}
|
||||
// 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 {
|
||||
|
|
@ -199,19 +196,60 @@ func TestRunHeartbeat_EverySuccessfulTick_RetriesWhilePending(t *testing.T) {
|
|||
|
||||
for i := range want {
|
||||
if got[i] != want[i] {
|
||||
t.Fatalf("callback event %d = %q, want %q (full sequence got=%v want=%v)",
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRunHeartbeat_AllTicksSucceed_NeverFiresOnFailureToSuccess(t *testing.T) {
|
||||
// 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 f2sCalls int32
|
||||
onFailureToSuccess := func() { atomic.AddInt32(&f2sCalls, 1) }
|
||||
var reconnectCalls int32
|
||||
onReconnectDetected := func() { atomic.AddInt32(&reconnectCalls, 1) }
|
||||
|
||||
var tickCalls int32
|
||||
tickEvents := make(chan struct{}, 8)
|
||||
|
|
@ -223,7 +261,7 @@ func TestRunHeartbeat_AllTicksSucceed_NeverFiresOnFailureToSuccess(t *testing.T)
|
|||
ctx, cancel := context.WithCancel(context.Background())
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
runHeartbeat(ctx, fn, onFailureToSuccess, onSuccessTick)
|
||||
runHeartbeat(ctx, fn, constantReconnectGen(42), onReconnectDetected, onSuccessTick)
|
||||
close(done)
|
||||
}()
|
||||
|
||||
|
|
@ -237,9 +275,9 @@ func TestRunHeartbeat_AllTicksSucceed_NeverFiresOnFailureToSuccess(t *testing.T)
|
|||
cancel()
|
||||
<-done
|
||||
|
||||
if got := atomic.LoadInt32(&f2sCalls); got != 0 {
|
||||
t.Errorf("onFailureToSuccess called %d times, want 0 for an all-successful "+
|
||||
"run -- no failure was ever observed, so there is no transition to fire on", got)
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -120,6 +120,21 @@ typedef struct endpoint_t {
|
|||
* their lifetime directly. */
|
||||
_Atomic int refcnt;
|
||||
int heap_alloc; /* 1 iff allocated by endpoint_get (unref may free) */
|
||||
/* bug-1548: bumped once, unconditionally, by xce_ctx_reconnect() (xcectx.c)
|
||||
* on every successful transport reconnect, regardless of trigger path
|
||||
* (write-path XCECLI_FAILOVER_CALL retry or heartbeat XCECLI_FF_CALL
|
||||
* best-effort self-heal). Neither endpoint_init() nor endpoint_fini()
|
||||
* (netio.c) touch this field, so it survives reconnect cycles the same
|
||||
* way refcnt/heap_alloc already do. Exists so a consumer (e.g. apiserver's
|
||||
* heartbeat prober, client.h's xcecli_get_my_endpoint) can detect that a
|
||||
* reconnect happened even when it completed silently inside a single
|
||||
* fire-and-forget call and the call's own return code never surfaced a
|
||||
* failure — XCECLI_FF_CALL can rotate+recover before evaluating its EXPR,
|
||||
* so a failure->success transition on the caller's own return code is not
|
||||
* a reliable signal (component-failover-adoption Phase 6). Does not
|
||||
* change reconnect timing/ordering/semantics — a passive observation
|
||||
* point at the existing success path, same shape as on_fini above. */
|
||||
_Atomic uint64_t reconnect_gen;
|
||||
} endpoint_t;
|
||||
|
||||
/* bug-1298 family: msg_handler runs SYNCHRONOUSLY on the io_worker thread (not
|
||||
|
|
|
|||
|
|
@ -178,6 +178,7 @@ libxce_test_src = ['testsuite/test.c',
|
|||
'testsuite/test_libxce_retry.c',
|
||||
'testsuite/test_netio_on_connected.c',
|
||||
'testsuite/test_netio_on_fini.c',
|
||||
'testsuite/test_netio_reconnect_gen.c',
|
||||
'testsuite/test_netio_arm_failure.c',
|
||||
'testsuite/test_netio_close_propagation.c',
|
||||
'testsuite/test_netio_bounded_connect.c',
|
||||
|
|
|
|||
|
|
@ -41,6 +41,7 @@ extern void test_cluster_parse(void);
|
|||
extern void test_libxce_retry(void);
|
||||
extern void test_netio_on_connected(void);
|
||||
extern void test_netio_on_fini(void);
|
||||
extern void test_netio_reconnect_gen(void);
|
||||
extern void test_netio_arm_failure(void);
|
||||
extern void test_netio_close_propagation(void);
|
||||
extern void test_netio_bounded_connect(void);
|
||||
|
|
@ -160,6 +161,7 @@ int main(int argc, char* argv[])
|
|||
TEST(test_libxce_retry);
|
||||
TEST(test_netio_on_connected);
|
||||
TEST(test_netio_on_fini);
|
||||
TEST(test_netio_reconnect_gen);
|
||||
TEST(test_netio_arm_failure);
|
||||
TEST(test_netio_close_propagation);
|
||||
TEST(test_netio_bounded_connect);
|
||||
|
|
|
|||
90
src/lib/testsuite/test_netio_reconnect_gen.c
Normal file
90
src/lib/testsuite/test_netio_reconnect_gen.c
Normal file
|
|
@ -0,0 +1,90 @@
|
|||
/*
|
||||
* test_netio_reconnect_gen.c — TDD tests for endpoint_t.reconnect_gen.
|
||||
*
|
||||
* component-failover-adoption Phase 6 (bug-1548): a monotonic counter bumped
|
||||
* by xce_ctx_reconnect() (xcectx.c) on every successful transport reconnect,
|
||||
* so a consumer (apiserver's heartbeat prober) can detect a reconnect even
|
||||
* when it completed silently inside a single XCECLI_FF_CALL and the call's
|
||||
* own return code never surfaced a failure.
|
||||
*
|
||||
* Mirrors test_netio_on_fini.c's structure for the same class of field
|
||||
* (a passive observation point added to endpoint_t). These tests exercise
|
||||
* only the field's lifecycle guarantees at the netio.c level — they do NOT
|
||||
* exercise xce_ctx_reconnect() itself (that requires a live socket/SSL
|
||||
* handshake) and are not a substitute for the required live E2E re-run of
|
||||
* task 5.2's scenario. Tests verify:
|
||||
* 1. A freshly heap-allocated endpoint_t starts with reconnect_gen == 0
|
||||
* 2. endpoint_init() does not reset a previously-bumped value
|
||||
* 3. endpoint_fini() does not reset a previously-bumped value
|
||||
*/
|
||||
|
||||
#define _GNU_SOURCE
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
#include <sys/socket.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
#include "netio.h"
|
||||
|
||||
/* ── Test 1: fresh endpoint_t starts at zero ─────────────────────── */
|
||||
|
||||
static void test_reconnect_gen_starts_at_zero(void)
|
||||
{
|
||||
endpoint_t *ep = endpoint_get();
|
||||
TEST_ASSERT_NOT_NULL(ep);
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT64(0, ep->reconnect_gen);
|
||||
|
||||
free(ep);
|
||||
printf("test_reconnect_gen_starts_at_zero: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test 2: endpoint_init() does not reset a bumped value ───────── */
|
||||
|
||||
static void test_reconnect_gen_survives_endpoint_init(void)
|
||||
{
|
||||
endpoint_t *ep = endpoint_get();
|
||||
TEST_ASSERT_NOT_NULL(ep);
|
||||
|
||||
ep->reconnect_gen = 7;
|
||||
|
||||
int fd = endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
|
||||
TEST_ASSERT_TRUE(fd >= 0);
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT64(7, ep->reconnect_gen);
|
||||
|
||||
endpoint_fini(ep);
|
||||
free(ep);
|
||||
printf("test_reconnect_gen_survives_endpoint_init: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test 3: endpoint_fini() does not reset a bumped value ───────── */
|
||||
|
||||
static void test_reconnect_gen_survives_endpoint_fini(void)
|
||||
{
|
||||
endpoint_t *ep = endpoint_get();
|
||||
TEST_ASSERT_NOT_NULL(ep);
|
||||
|
||||
int fd = endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
|
||||
TEST_ASSERT_TRUE(fd >= 0);
|
||||
|
||||
ep->reconnect_gen = 3;
|
||||
|
||||
endpoint_fini(ep);
|
||||
|
||||
TEST_ASSERT_EQUAL_UINT64(3, ep->reconnect_gen);
|
||||
|
||||
free(ep);
|
||||
printf("test_reconnect_gen_survives_endpoint_fini: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test suite entry point ───────────────────────────────────────── */
|
||||
|
||||
void test_netio_reconnect_gen(void)
|
||||
{
|
||||
printf("\n>> netio reconnect_gen field lifecycle tests\n\n");
|
||||
|
||||
RUN_TEST(test_reconnect_gen_starts_at_zero);
|
||||
RUN_TEST(test_reconnect_gen_survives_endpoint_init);
|
||||
RUN_TEST(test_reconnect_gen_survives_endpoint_fini);
|
||||
}
|
||||
|
|
@ -393,6 +393,14 @@ int xce_ctx_reconnect(xce_ctx_t* ctx, char* host, int port)
|
|||
return rc;
|
||||
}
|
||||
|
||||
/* bug-1548: mark that a reconnect actually completed, so a consumer
|
||||
* polling ep->reconnect_gen (e.g. apiserver's heartbeat prober) can
|
||||
* detect it even when this whole reconnect happened silently inside a
|
||||
* single XCECLI_FF_CALL and the caller's own return code never surfaced
|
||||
* a failure. Only bumped on this success path — a failed attempt above
|
||||
* already returned early without reaching here. */
|
||||
ctx->ep->reconnect_gen++;
|
||||
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue