webui4frpc/cmd/webui4frpc/rejoin_ring_test.go

package main

import (
	"context"
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"sync"
	"testing"
	"time"

	"webui4frpc/internal/cluster"
	"webui4frpc/internal/store"
)

// fakeJoinPeer stands in for a healthy cluster member: it answers the rejoin
// POST and records it, so the rejoin path can be exercised end to end.
type fakeJoinPeer struct {
	mu      sync.Mutex
	joined  int
	stateFn func() cluster.State
}

func (p *fakeJoinPeer) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	p.mu.Lock()
	p.joined++
	p.mu.Unlock()
	w.Header().Set("Content-Type", "application/json")
	_ = json.NewEncoder(w).Encode(map[string]any{"state": p.stateFn()})
}

func (p *fakeJoinPeer) joins() int {
	p.mu.Lock()
	defer p.mu.Unlock()
	return p.joined
}

// TestRejoinThenStartRingCoversTheRejoinPath is the regression test for the
// frozen ring, and it deliberately drives retryRejoinCached itself rather than
// calling StartRing directly — a test that calls the fix's own helper proves
// nothing about whether the fix is wired up (verified: such a test stayed green
// with the fix reverted).
//
// The failure, observed live on the three-node cluster: the leader restarted,
// rejoined successfully, logged "startup reconcile: reclaimed=0 submitted=3",
// and then never forwarded a token. cycle froze, topology stayed empty and the
// three forwards sat in pending forever, because rejoin returned immediately
// and only a leader may call StartRing.
func TestRejoinThenStartRingCoversTheRejoinPath(t *testing.T) {
	selfID := "192.168.2.30:7500"

	// The peer hands back a ring in which SELF is the leader — the situation a
	// rolling restart produces when the leader comes back last.
	var peerState cluster.State
	peer := &fakeJoinPeer{stateFn: func() cluster.State { return peerState }}

	// Bind on loopback so the engine's hardcoded peer address is reachable.
	// We therefore advertise 127.0.0.1 in the ring while keeping the same IDs.
	ln := httptest.NewServer(peer)
	defer ln.Close()

	peerState = cluster.State{
		LeaderID: selfID,
		Nodes: []cluster.Node{
			{ID: selfID, Addr: selfID, Alive: true, IsLeader: true},
			{ID: "peer:7500", Addr: "peer:7500", Alive: true},
		},
		Topology: map[string]*cluster.TopoEntry{},
		PendingTasks: map[string]*cluster.Task{
			"t1": {ID: "t1", Local: store.Local{Name: "web"},
				Remote: store.Remote{Name: "frps1", IP: "1.2.3.4", Port: 7000, Enabled: true},
				Link:   store.Link{RemotePort: 18081}},
		},
	}

	// Redirect the peer's advertised address at the transport level: the engine
	// dials whatever the ring says, so give the ring the test server's address.
	// JoinRing builds "http://<addr>/api/manager/cluster/join" itself, so the
	// cached entry must be a bare host:port, not a URL.
	realAddr := strings.TrimPrefix(ln.URL, "http://")
	for i := range peerState.Nodes {
		if peerState.Nodes[i].ID == "peer:7500" {
			peerState.Nodes[i].Addr = realAddr
		}
	}

	var mu sync.Mutex
	var emitted int
	eng := cluster.NewEngine(selfID, selfID, "u", "p", "0.1.2", nil,
		&cluster.AppHandler{ClaimFn: func(ctx context.Context, tk *cluster.Task) error { return nil }},
		func(ctx context.Context, next string, tk *cluster.Token) error {
			if tk != nil {
				mu.Lock()
				emitted++
				mu.Unlock()
			}
			return nil
		},
		selfID, true, "")

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// The rejoin path, as main() runs it. Cached peers point at the test server.
	peersJSON, err := json.Marshal([]map[string]string{{"addr": realAddr, "key": "peerkey"}})
	if err != nil {
		t.Fatal(err)
	}

	done := make(chan struct{})
	go func() {
		defer close(done)
		retryRejoinCached(ctx, eng, string(peersJSON))
	}()

	// retryRejoinCached returns once rejoin succeeds; give the goroutine a moment.
	for i := 0; i < 100; i++ {
		if peer.joins() > 0 {
			break
		}
		sleepShort()
	}
	if peer.joins() == 0 {
		t.Fatal("the rejoin never reached the peer — the test setup is wrong")
	}
	<-done

	// Rejoin succeeded. The engine must now consider itself the leader...
	if !eng.IsLeader() {
		t.Fatalf("precondition broken: rejoined node does not consider itself leader")
	}
	sleepShort()

	// ...and it must have restarted the ring: a token has to be emitted, or the
	// cluster sits frozen with its forwards stranded in pending.
	mu.Lock()
	got := emitted
	mu.Unlock()
	if got == 0 {
		t.Fatal("after a successful rejoin the leader emitted NO token: the ring is frozen " +
			"and pending forwards are never claimed (the live failure this guards)")
	}
}
func sleepShort() { time.Sleep(10 * time.Millisecond) }