webui4frpc/internal/cluster/ring_engine_test.go

package cluster

import (
	"context"
	"testing"

	"webui4frpc/internal/store"
)

type fakeHandler struct {
	load   Load
	claim  func(ctx context.Context, tk *Task) error
	revoke func(ctx context.Context, tk *Task) error
}

func (h *fakeHandler) Revoke(ctx context.Context, tk *Task) error {
	if h.revoke != nil {
		return h.revoke(ctx, tk)
	}
	return nil
}

func (h *fakeHandler) RuntimeLoad() Load { return h.load }
func (h *fakeHandler) Claim(ctx context.Context, tk *Task) error {
	if h.claim != nil {
		return h.claim(ctx, tk)
	}
	return nil
}

// sendNull is a no-op sender (single-node ring).
func sendNull(ctx context.Context, next string, tk *Token) error { return nil }

// TestSingleNodeCycle: leader starts ring, processes phases locally, and the
// task gets claimed (lowest load = only node).
func TestSingleNodeCycle(t *testing.T) {
	eng := NewEngine("n1", "n1:7500", "u", "p", "0.71.0", []string{"0.71.0"},
		&fakeHandler{load: Load{MemPct: 30, NetPct: 30}}, sendNull, "n1:7500", true, "")
	eng.StartRing(context.Background())

	// single node: token stays local; no successor so nothing travels.
	if eng.State().Nodes[0].ID != "n1" {
		t.Fatalf("nodes = %+v", eng.State().Nodes)
	}
}

// TestTwoNodesPhaseCollect: leader n1 sends to n2; n1 collects both after n2
// stamps; then phase flips and second round syncs.
func TestTwoNodesSingleRound(t *testing.T) {
	eng2 := NewEngine("n2", "n2:7500", "u", "p", "0.71.0", nil,
		&fakeHandler{load: Load{MemPct: 10, NetPct: 10}}, sendNull, "n2:7500", false, "")
	eng2.state.UpsertNode(Node{ID: "n1", Addr: "n1:7500", Alive: true, IsLeader: true, Load: Load{MemPct: 50, NetPct: 50}})

	// n1 sends a token carrying the cluster picture; single-round processing:
	// the receiving node adopts it and appends its own info in one pass.
	tk := &Token{Cycle: 1, State: State{
		Nodes: []Node{
			{ID: "n1", Addr: "n1:7500", Alive: true, IsLeader: true},
			{ID: "n2", Addr: "n2:7500", Alive: true},
		},
	}}

	out, err := eng2.OnToken(context.Background(), tk)
	if err != nil {
		t.Fatal(err)
	}
	if eng2.State().Find("n2") < 0 {
		t.Fatal("n2 should be present in adopted state after single-round")
	}
	if out == nil {
		t.Fatal("nil token after processing")
	}
}

// TestLeaderRoundTripNoResurrection: when the leader submits a task and the
// token completes a full round (leader→n2 claims→n3→leader), the consumed task
// MUST NOT be resurrected on the leader's next OnToken, and the topology must
// still attribute the forward to the single claimer (n2).
//
// This holds NOT via a published-set mark on StartRing, but via Go map
// reference semantics: StartRing sets tk.State = e.state, so the token's
// PendingTasks map IS the leader's own map. When n2 calls ClaimPending it
// deletes from that shared map — the deletion is visible to the leader too.
// By the time the token returns, the claimed task is already gone from the
// leader's e.state.PendingTasks, so the localPending re-merge has nothing to
// re-inject. Single token + remove-on-claim = single claim (plan §M6).
func TestLeaderRoundTripNoResurrection(t *testing.T) {
	// 3-node ring: n1 (leader+submitter), n2 (lowest, claims), n3 (idle).
	var sent *Token
	sendCap := func(ctx context.Context, next string, tk *Token) error {
		sent = tk
		return nil
	}
	n1 := NewEngine("n1", "n1:7500", "u", "p", "v", nil,
		&fakeHandler{load: Load{MemPct: 50, NetPct: 50}}, sendCap, "n1:7500", true, "")
	// n2/n3 carry a lower stored load than n1's NewEngine default ({10,10})
	// so LowestAlive picks n2 (first among the tied low nodes) as the claimer.
	n1.state.UpsertNode(Node{ID: "n2", Addr: "n2:7500", Alive: true, Load: Load{MemPct: 1, NetPct: 1}})
	n1.state.UpsertNode(Node{ID: "n3", Addr: "n3:7500", Alive: true, Load: Load{MemPct: 1, NetPct: 1}})

	task := n1.SubmitTask(store.Local{Name: "l1"}, store.Remote{Name: "r1"}, store.Link{RemotePort: 100})
	if task == nil {
		t.Fatal("submit returned nil")
	}
	n1.StartRing(context.Background())
	if sent == nil {
		t.Fatal("StartRing did not send a token")
	}

	// n2 receives, claims t1 (lowest load), establishes topology.
	n2 := NewEngine("n2", "n2:7500", "u", "p", "v", nil,
		&fakeHandler{load: Load{MemPct: 10, NetPct: 10}}, sendCap, "n2:7500", false, "")
	out2, err := n2.OnToken(context.Background(), sent)
	if err != nil {
		t.Fatal(err)
	}
	if got := len(n2.state.TopologyList()); got != 1 {
		t.Fatalf("n2 should own 1 forward, topo=%+v", n2.state.TopologyList())
	}
	if len(n2.state.PendingList()) != 0 {
		t.Fatalf("pending should be empty after n2 claim, got %+v", n2.state.PendingList())
	}

	// n3 receives, nothing to claim, forwards.
	n3 := NewEngine("n3", "n3:7500", "u", "p", "v", nil,
		&fakeHandler{load: Load{MemPct: 10, NetPct: 10}}, sendCap, "n3:7500", false, "")
	out3, err := n3.OnToken(context.Background(), out2)
	if err != nil {
		t.Fatal(err)
	}

	// Token returns to n1. The consumed task t1 MUST NOT be resurrected.
	if _, err := n1.OnToken(context.Background(), out3); err != nil {
		t.Fatal(err)
	}
	if got := len(n1.state.PendingList()); got != 0 {
		t.Fatalf("n1 resurrected a consumed task: pending=%+v (single-token + shared-map must prevent this)",
			n1.state.PendingList())
	}
	// Topology still attributes t1 to n2 (not overwritten by a re-claim).
	topo := n1.state.TopologyList()
	if len(topo) != 1 || topo[0].OwnerID != "n2" {
		t.Fatalf("topology should be t1@n2, got %+v", topo)
	}
}

// TestClaimGuardDropsDuplicateForward: when a node (lowest load) receives a
// pending task whose forward is ALREADY in the topology owned by another node
// (a resurrected/stale copy), the claim path must drop it WITHOUT spawning a
// worker or overwriting the topology. Without the guard a second worker for
// the same forward would be spawned and orphaned (the topology entry is keyed
// by task id, so AddTopology would silently overwrite the owner).
func TestClaimGuardDropsDuplicateForward(t *testing.T) {
	// n1 is lowest load and would otherwise claim; the fake Claim hook fails
	// the test if ever called.
	n1 := NewEngine("n1", "n1:7500", "u", "p", "v", nil,
		&fakeHandler{
			load: Load{MemPct: 10, NetPct: 10},
			claim: func(ctx context.Context, tk *Task) error {
				t.Fatalf("Claim must not be called for an already-owned forward: %s", tk.ID)
				return nil
			},
		}, sendNull, "n1:7500", true, "")

	dup := &Task{ID: "t1", Local: store.Local{Name: "l1"}, Remote: store.Remote{Name: "r1"}, Link: store.Link{RemotePort: 100}}
	tk := &Token{Cycle: 1, State: State{
		Nodes: []Node{
			{ID: "n1", Addr: "n1:7500", Alive: true, Load: Load{MemPct: 10, NetPct: 10}},
			{ID: "n2", Addr: "n2:7500", Alive: true, Load: Load{MemPct: 50, NetPct: 50}},
		},
		PendingTasks: map[string]*Task{"t1": dup},
		Topology: map[string]*TopoEntry{"t1": {
			TaskID: "t1", OwnerID: "n2", Local: store.Local{Name: "l1"},
			Remote: store.Remote{Name: "r1"}, Link: store.Link{RemotePort: 100}, Active: true,
		}},
	}}
	if _, err := n1.OnToken(context.Background(), tk); err != nil {
		t.Fatal(err)
	}
	// Pending drained (the stale task was consumed/dropped, not left to ride).
	if got := len(n1.state.PendingList()); got != 0 {
		t.Fatalf("stale task should be dropped, pending=%+v", n1.state.PendingList())
	}
	// Topology untouched: still owned by n2.
	topo := n1.state.TopologyList()
	if len(topo) != 1 || topo[0].OwnerID != "n2" {
		t.Fatalf("topology should remain t1@n2, got %+v", topo)
	}
}