webui4frpc/internal/cluster/ring_log_test.go

package cluster

import (
	"context"
	"testing"

	"webui4frpc/internal/store"
)

// Helper: build an engine for tests.
func newTestEngine(id string, isLeader bool) *Engine {
	return NewEngine(id, id+":7500", "u", "p", "0.1.0", nil,
		&fakeHandler{load: Load{MemPct: 20, NetPct: 20}},
		func(ctx context.Context, next string, tk *Token) error { return nil },
		id+":7500", isLeader, "")
}

// TestLogAppendDeltaReplay: append entries, extract delta after a watermark,
// replay on another node idempotently + in order.
func TestLogAppendDeltaReplay(t *testing.T) {
	l := NewClusterLog()
	if _, err := l.Append("n1", LogForwardAdd, map[string]string{"k": "v"}); err != nil {
		t.Fatal(err)
	}
	e2, _ := l.Append("n1", LogNodeJoin, nil)
	if e2.Seq != 2 {
		t.Fatalf("seq = %d want 2", e2.Seq)
	}

	dst := NewClusterLog()
	if _, err := dst.ApplyDelta(l.EntriesAfter(0)); err != nil {
		t.Fatalf("apply: %v", err)
	}
	if len(dst.Snapshot()) != 2 {
		t.Fatalf("dst log = %+v", dst.Snapshot())
	}
	if _, err := dst.ApplyDelta(l.EntriesAfter(0)); err != nil {
		t.Fatalf("idempotent apply: %v", err)
	}
	if len(dst.Snapshot()) != 2 {
		t.Fatalf("dedupe failed: %+v", dst.Snapshot())
	}
}

// TestLogGapDetection: a delta with a seq gap must error.
func TestLogGapDetection(t *testing.T) {
	l := NewClusterLog()
	if _, err := l.ApplyDelta([]LogEntry{{Seq: 1, Kind: LogNodeJoin}, {Seq: 3, Kind: LogForwardAdd}}); err == nil {
		t.Fatal("expected gap error, got nil")
	}
}

// TestLogAfterWatermark: delta only includes entries newer than watermark.
func TestLogAfterWatermark(t *testing.T) {
	l := NewClusterLog()
	_, _ = l.Append("a", LogNodeJoin, nil)
	_, _ = l.Append("a", LogForwardAdd, nil)
	after := l.EntriesAfter(1)
	if len(after) != 1 || after[0].Seq != 2 {
		t.Fatalf("after(1) = %+v want [seq 2]", after)
	}
}

// TestClaimLogsToEngine: a claimed task appends LogForwardAdd to engine log.
func TestClaimLogsToEngine(t *testing.T) {
	eng := newTestEngine("n1", true)
	eng.state.AddPending(store.Local{Name: "l1"}, store.Remote{Name: "r1"}, store.Link{})
	out, err := eng.OnToken(context.Background(), &Token{Cycle: 1, State: eng.state})
	if err != nil {
		t.Fatal(err)
	}
	if out == nil {
		t.Fatal("nil token after OnToken")
	}
	snap := eng.Log.Snapshot()
	if len(snap) != 1 || snap[0].Kind != LogForwardAdd {
		t.Fatalf("engine log = %+v", snap)
	}
}

// TestFullLogBackfillAfterGap: a node that missed entries while offline
// (Synced stuck below the ring's max) must backfill from a peer's FULL log
// attachment. This is the regression test for the "log delta gap: want 1 got
// N" livelock where offline nodes could never rejoin the log history.
func TestFullLogBackfillAfterGap(t *testing.T) {
	// Peer with complete history [1..4].
	var peer ClusterLog
	for i := 1; i <= 4; i++ {
		if _, err := peer.Append("n1", LogNodeJoin, map[string]int{"i": i}); err != nil {
			t.Fatal(err)
		}
	}
	// Straggler that only has [1]; it missed [2..3] while offline and now
	// receives the peer's FULL attachment [1..4].
	straggler := NewClusterLog()
	if _, err := straggler.Append("n2", LogNodeJoin, nil); err != nil {
		t.Fatal(err)
	}
	// Force straggler to look like it has seq1 only (Synced=1).
	straggler.Synced = 1

	wm, err := straggler.ApplyDelta(peer.Snapshot())
	if err != nil {
		t.Fatalf("full backfill failed: %v", err)
	}
	if wm != 4 {
		t.Fatalf("watermark = %d, want 4", wm)
	}
	if len(straggler.Snapshot()) != 4 {
		t.Fatalf("log length = %d, want 4 (no dupes)", len(straggler.Snapshot()))
	}
}

// TestApplyDeltaIdempotentOnFullResend: applying the same full attachment
// twice must not duplicate entries or error — peers re-attach their full log
// every cycle now.
func TestApplyDeltaIdempotentOnFullResend(t *testing.T) {
	var src ClusterLog
	for i := 1; i <= 3; i++ {
		if _, err := src.Append("n1", LogNodeJoin, nil); err != nil {
			t.Fatal(err)
		}
	}
	full := src.Snapshot()
	dst := NewClusterLog()
	if _, err := dst.ApplyDelta(full); err != nil {
		t.Fatalf("first apply: %v", err)
	}
	wm, err := dst.ApplyDelta(full)
	if err != nil {
		t.Fatalf("second apply (idempotence): %v", err)
	}
	if wm != 3 || len(dst.Snapshot()) != 3 {
		t.Fatalf("wm=%d len=%d, want 3/3", wm, len(dst.Snapshot()))
	}
}

// TestLocalAppendNotDuplicatedByPeerFullLog: a locally appended entry (seq=N)
// must NOT be re-applied when a peer's full-log attachment carries the same
// seq — this was the source of duplicate rows in the audit log.
func TestLocalAppendNotDuplicatedByPeerFullLog(t *testing.T) {
	var mine ClusterLog
	if _, err := mine.Append("n1", LogForwardAdd, map[string]string{"local": "web"}); err != nil {
		t.Fatal(err)
	}
	// Simulate the pre-fix bug window: peer attaches [1..N] including our N.
	peerHas := mine.Snapshot()
	before := len(mine.Snapshot())
	if _, err := mine.ApplyDelta(peerHas); err != nil {
		t.Fatalf("apply own entry back: %v", err)
	}
	if got := len(mine.Snapshot()); got != before {
		t.Fatalf("log grew from %d to %d after re-applying own entry", before, got)
	}
}