fscan/core/optimization_integration_test.go

package core

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

// =============================================================================
// 优化 1：target/ceiling 分离
// =============================================================================

func TestOpt1_TargetCeilingSeparation_TuneConfig(t *testing.T) {
	config := makeDefaultConfig()
	session := makeTestSession(config)

	ep := &EnvironmentProfile{
		Net: NetworkProfile{
			Env:       EnvLAN,
			RTTMedian: 1 * time.Millisecond,
			RTTStddev: 500 * time.Microsecond,
			LossRate:  0.0,
			Samples:   30,
		},
		System: SystemProfile{FDLimit: 65536, NumCPU: 8},
	}

	ep.TuneConfig(config, session)

	if config.ThreadCeiling <= 0 {
		t.Fatalf("ThreadCeiling 未被设置: %d", config.ThreadCeiling)
	}

	// 内网 factor=1.5，非显式 → target=ceiling=recommended
	// 但 ceiling 应该 >= target
	if config.ThreadCeiling < config.ThreadNum {
		t.Errorf("Ceiling(%d) < ThreadNum(%d)", config.ThreadCeiling, config.ThreadNum)
	}

	t.Logf("target=%d, ceiling=%d", config.ThreadNum, config.ThreadCeiling)
}

func TestOpt1_TargetCeilingSeparation_ExplicitT(t *testing.T) {
	config := makeDefaultConfig()
	config.ThreadNum = 200
	config.ThreadNumExplicit = true
	session := makeTestSession(config)

	ep := &EnvironmentProfile{
		Net: NetworkProfile{
			Env:       EnvInternet,
			RTTMedian: 100 * time.Millisecond,
			RTTStddev: 30 * time.Millisecond,
			LossRate:  0.0,
			Samples:   20,
		},
		System: SystemProfile{FDLimit: 65536, NumCPU: 8},
	}

	ep.TuneConfig(config, session)

	// 用户显式指定 -t → ceiling = threadNum = 200
	if config.ThreadCeiling != 200 {
		t.Errorf("显式 -t 200: ceiling=%d, want 200", config.ThreadCeiling)
	}
	if config.ThreadNum != 200 {
		t.Errorf("显式 -t 200: threadNum=%d, want 200", config.ThreadNum)
	}
}

func TestOpt1_PoolUsesCeiling(t *testing.T) {
	metrics := &ScanMetrics{}
	target, ceiling := 50, 200

	pool, err := NewAdaptivePool(target, ceiling, func(interface{}) {}, metrics)
	if err != nil {
		t.Fatalf("创建池失败: %v", err)
	}
	defer pool.Release()

	pool.inSlowStart = false
	pool.tune(target)

	// 注入健康 metrics 让池增长
	for i := 0; i < 200; i++ {
		metrics.RecordConnect(time.Millisecond)
	}

	// 多次 adjust，池应能增长超过 target 但不超过 ceiling
	for i := 0; i < 30; i++ {
		pool.lastCheck.Store(0)
		pool.adjust()
	}

	finalCap := pool.Cap()
	if finalCap <= target {
		t.Errorf("池应能超过 target(%d): cap=%d", target, finalCap)
	}
	if finalCap > ceiling {
		t.Errorf("池不应超过 ceiling(%d): cap=%d", ceiling, finalCap)
	}

	t.Logf("target=%d, ceiling=%d, finalCap=%d", target, ceiling, finalCap)
}

func TestOpt1_FDLimitConstraintsBothFields(t *testing.T) {
	config := makeDefaultConfig()
	config.ThreadNum = 1000
	session := makeTestSession(config)

	ep := &EnvironmentProfile{
		Net: NetworkProfile{
			Env:       EnvLAN,
			RTTMedian: 1 * time.Millisecond,
			RTTStddev: 500 * time.Microsecond,
			LossRate:  0.0,
			Samples:   30,
		},
		System: SystemProfile{FDLimit: 256, NumCPU: 4},
	}

	ep.TuneConfig(config, session)

	maxFD := 256 * 6 / 10
	if config.ThreadNum > maxFD {
		t.Errorf("ThreadNum(%d) 超过 fd 限制(%d)", config.ThreadNum, maxFD)
	}
	if config.ThreadCeiling > maxFD {
		t.Errorf("ThreadCeiling(%d) 超过 fd 限制(%d)", config.ThreadCeiling, maxFD)
	}
}

// =============================================================================
// 优化 2：RTT 漂移微调 target
// =============================================================================

func TestOpt2_RTTDriftReducesTarget(t *testing.T) {
	metrics := &ScanMetrics{}
	pool, err := NewAdaptivePool(200, 400, func(interface{}) {}, metrics)
	if err != nil {
		t.Fatalf("创建池失败: %v", err)
	}
	defer pool.Release()

	pool.inSlowStart = false
	pool.tune(200)

	// 建立基线：slow EMA 锚定在 1ms 附近
	for i := 0; i < 500; i++ {
		metrics.RecordConnect(1 * time.Millisecond)
	}

	origTarget := atomic.LoadInt32(&pool.target)

	// RTT 突增到 100ms（100 倍），大量喂入让 fast EMA 拉开差距
	for i := 0; i < 1000; i++ {
		metrics.RecordConnect(100 * time.Millisecond)
	}

	ratio := metrics.RTTRatio()
	t.Logf("RTT ratio after spike: %.2f", ratio)

	if ratio <= 3.0 {
		t.Skipf("RTT ratio=%.2f，EMA 差距不够大，跳过", ratio)
	}

	// 需要足够的新 metrics 让 assessHealth 的 deltaTotal >= 30
	for i := 0; i < 50; i++ {
		metrics.RecordConnect(100 * time.Millisecond)
	}

	// 多次 adjust 触发 maybeReduceTarget
	for i := 0; i < 10; i++ {
		pool.lastCheck.Store(0)
		pool.prevSnapshot = MetricsSnapshot{} // 重置快照让 delta 足够
		pool.adjust()
	}

	newTarget := atomic.LoadInt32(&pool.target)
	if newTarget >= origTarget {
		t.Errorf("RTT 漂移后 target 应降低: %d -> %d (ratio=%.2f)", origTarget, newTarget, ratio)
	}

	// 不应低于 ceiling/5
	minTarget := atomic.LoadInt32(&pool.ceiling) / 5
	if newTarget < minTarget {
		t.Errorf("target(%d) 低于下限(%d)", newTarget, minTarget)
	}

	t.Logf("RTT drift: ratio=%.2f, target %d -> %d (min=%d)", ratio, origTarget, newTarget, minTarget)
}

func TestOpt2_NoReductionWhenStable(t *testing.T) {
	metrics := &ScanMetrics{}
	pool, err := NewAdaptivePool(200, 400, func(interface{}) {}, metrics)
	if err != nil {
		t.Fatalf("创建池失败: %v", err)
	}
	defer pool.Release()

	pool.inSlowStart = false
	pool.tune(200)

	// 稳定 RTT
	for i := 0; i < 200; i++ {
		metrics.RecordConnect(10 * time.Millisecond)
	}

	origTarget := atomic.LoadInt32(&pool.target)

	for i := 0; i < 10; i++ {
		pool.lastCheck.Store(0)
		pool.adjust()
	}

	newTarget := atomic.LoadInt32(&pool.target)
	if newTarget != origTarget {
		t.Errorf("稳定 RTT 不应改变 target: %d -> %d", origTarget, newTarget)
	}
}

// =============================================================================
// 优化 3：assessHealth 阈值跟 NetworkEnv 关联
// =============================================================================

func TestOpt3_LANTighterThresholds(t *testing.T) {
	metrics := &ScanMetrics{}
	pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics, EnvLAN)
	if err != nil {
		t.Fatalf("创建池失败: %v", err)
	}
	defer pool.Release()

	pool.inSlowStart = false
	pool.tune(100)

	// 10% exhaust rate — 对 LAN 来说应该是 Congested（阈值 8%）
	for i := 0; i < 100; i++ {
		if i < 10 {
			metrics.RecordExhausted()
		} else {
			metrics.RecordConnect(time.Millisecond)
		}
	}

	pool.lastCheck.Store(0)
	pool.adjust()

	if pool.Cap() >= 100 {
		t.Errorf("LAN 10%% exhaust 应触发降速: cap=%d", pool.Cap())
	}

	t.Logf("LAN tight threshold: cap=%d (from 100)", pool.Cap())
}

func TestOpt3_InternetLooseThresholds(t *testing.T) {
	metrics := &ScanMetrics{}
	pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics, EnvInternet)
	if err != nil {
		t.Fatalf("创建池失败: %v", err)
	}
	defer pool.Release()

	pool.inSlowStart = false
	pool.tune(100)

	// 10% exhaust rate — 对 Internet 来说不算 Congested（阈值 25%），应是 Stressed
	for i := 0; i < 100; i++ {
		if i < 10 {
			metrics.RecordExhausted()
		} else {
			metrics.RecordConnect(10 * time.Millisecond)
		}
	}

	pool.lastCheck.Store(0)
	pool.adjust()
	capAfter := pool.Cap()

	// Internet 对 10% exhaust 只是 Stressed（×0.85），不是 Congested（×0.5）
	if capAfter < 80 {
		t.Errorf("Internet 10%% exhaust 不应大幅降速: cap=%d", capAfter)
	}

	t.Logf("Internet loose threshold: cap=%d (from 100)", capAfter)
}

func TestOpt3_EnvAffectsHealthDecision(t *testing.T) {
	envs := []struct {
		env  NetworkEnv
		name string
	}{
		{EnvLAN, "LAN"},
		{EnvWAN, "WAN"},
		{EnvInternet, "Internet"},
	}

	var caps []int

	for _, e := range envs {
		metrics := &ScanMetrics{}
		pool, err := NewAdaptivePool(100, 100, func(interface{}) {}, metrics, e.env)
		if err != nil {
			t.Fatalf("创建池失败: %v", err)
		}

		pool.inSlowStart = false
		pool.tune(100)

		// 相同的 12% exhaust rate
		for i := 0; i < 100; i++ {
			if i < 12 {
				metrics.RecordExhausted()
			} else {
				metrics.RecordConnect(time.Millisecond)
			}
		}

		pool.lastCheck.Store(0)
		pool.adjust()
		caps = append(caps, pool.Cap())
		pool.Release()

		t.Logf("%s: cap=%d (12%% exhaust)", e.name, caps[len(caps)-1])
	}

	// LAN 反应最激烈（cap 最低），Internet 最宽容（cap 最高）
	if caps[0] >= caps[2] {
		t.Errorf("LAN cap(%d) 应 < Internet cap(%d) for same exhaust rate", caps[0], caps[2])
	}
}

// =============================================================================
// 优化 4：去掉 semaphore，ants 池天然反压
// =============================================================================

func TestOpt4_SemaphoreRemoved(t *testing.T) {
	// 验证 portScanTask 结构体不再有 semaphore 字段
	// 如果 semaphore 被加回来，这段代码编译就会报 "unknown field"
	_ = portScanTask{
		host: "127.0.0.1",
		port: 80,
		addr: "127.0.0.1:80",
	}
	t.Log("portScanTask 无 semaphore 字段，反压由 ants pool 统一管理")
}

// =============================================================================
// 优化 5：扩充探测端口
// =============================================================================

func TestOpt5_ProbePortsExpanded(t *testing.T) {
	if len(probePorts) < 5 {
		t.Errorf("probePorts 只有 %d 个，应该扩充到至少 5 个", len(probePorts))
	}

	// 验证包含关键端口
	required := map[int]bool{80: false, 443: false, 22: false}
	for _, p := range probePorts {
		if _, ok := required[p]; ok {
			required[p] = true
		}
	}
	for port, found := range required {
		if !found {
			t.Errorf("probePorts 缺少关键端口 %d", port)
		}
	}

	// 验证没有重复
	seen := make(map[int]bool)
	for _, p := range probePorts {
		if seen[p] {
			t.Errorf("probePorts 有重复端口 %d", p)
		}
		seen[p] = true
	}

	t.Logf("probePorts = %v (%d 个)", probePorts, len(probePorts))
}

// =============================================================================
// 优化 6：computeRetries 环境自适应
// =============================================================================

func TestOpt6_RetriesEnvAware(t *testing.T) {
	lossRate := 0.3 // 30% 丢包

	lanRetry := computeRetries(lossRate, EnvLAN)
	wanRetry := computeRetries(lossRate, EnvWAN)
	inetRetry := computeRetries(lossRate, EnvInternet)

	// LAN 目标概率更严格(0.5%)，应该重试更多；但上限更低(4)
	// Internet 目标概率更宽松(2%)，应该重试更少；但上限更高(6)
	t.Logf("30%% loss: LAN=%d, WAN=%d, Internet=%d", lanRetry, wanRetry, inetRetry)

	if lanRetry < 1 || lanRetry > 4 {
		t.Errorf("LAN retry=%d, 应在 [1,4]", lanRetry)
	}
	if wanRetry < 1 || wanRetry > 5 {
		t.Errorf("WAN retry=%d, 应在 [1,5]", wanRetry)
	}
	if inetRetry < 1 || inetRetry > 6 {
		t.Errorf("Internet retry=%d, 应在 [1,6]", inetRetry)
	}
}

func TestOpt6_RetriesMaxByEnv(t *testing.T) {
	// 高丢包率，各环境应返回各自上限
	lanMax := computeRetries(0.99, EnvLAN)
	wanMax := computeRetries(0.99, EnvWAN)
	inetMax := computeRetries(0.99, EnvInternet)

	if lanMax != 4 {
		t.Errorf("LAN max retry=%d, want 4", lanMax)
	}
	if wanMax != 5 {
		t.Errorf("WAN max retry=%d, want 5", wanMax)
	}
	if inetMax != 6 {
		t.Errorf("Internet max retry=%d, want 6", inetMax)
	}
}

func TestOpt6_RetriesMathCorrectness(t *testing.T) {
	envs := []struct {
		env        NetworkEnv
		targetProb float64
		name       string
	}{
		{EnvLAN, 0.005, "LAN"},
		{EnvWAN, 0.01, "WAN"},
		{EnvInternet, 0.02, "Internet"},
	}

	for _, e := range envs {
		for _, loss := range []float64{0.05, 0.10, 0.20, 0.30} {
			retries := computeRetries(loss, e.env)
			prob := 1.0
			for i := 0; i < retries; i++ {
				prob *= loss
			}
			// 重试后全失败概率应 < targetProb（除非被 clamp 了）
			if prob >= e.targetProb && retries < 4 {
				t.Errorf("%s loss=%.0f%% retries=%d: P=%.6f >= %.3f",
					e.name, loss*100, retries, prob, e.targetProb)
			}
		}
	}
}

// =============================================================================
// 端到端集成：全链路验证
// =============================================================================

func TestOptAll_EndToEnd_LANToPool(t *testing.T) {
	// 模拟内网探测 → TuneConfig → 创建池 → 池根据 env 自适应
	profile := classifyNetwork(
		makeDurations([]int{1, 1, 2, 2, 2, 3, 3, 3, 4, 5}),
		0, 10,
	)

	config := makeDefaultConfig()
	session := makeTestSession(config)
	ep := &EnvironmentProfile{Net: *profile, System: SystemProfile{FDLimit: 65536, NumCPU: 8}}
	ep.TuneConfig(config, session)

	// 验证 env 被存储
	if config.DetectedNetworkEnv != int(EnvLAN) {
		t.Errorf("DetectedNetworkEnv=%d, want %d(LAN)", config.DetectedNetworkEnv, int(EnvLAN))
	}

	// 验证 ceiling 合理
	if config.ThreadCeiling < config.ThreadNum {
		t.Errorf("ceiling(%d) < target(%d)", config.ThreadCeiling, config.ThreadNum)
	}

	// 创建池并验证 env 传递
	netEnv := NetworkEnv(config.DetectedNetworkEnv)
	metrics := &ScanMetrics{}
	pool, err := NewAdaptivePool(config.ThreadNum, config.ThreadCeiling, func(interface{}) {}, metrics, netEnv)
	if err != nil {
		t.Fatalf("创建池失败: %v", err)
	}
	defer pool.Release()

	if pool.networkEnv != EnvLAN {
		t.Errorf("池的 networkEnv=%v, want LAN", pool.networkEnv)
	}

	t.Logf("端到端 LAN: target=%d ceiling=%d env=%v maxRetry=%d",
		config.ThreadNum, config.ThreadCeiling, netEnv, config.MaxRetries)
}

func TestOptAll_EndToEnd_InternetToPool(t *testing.T) {
	profile := classifyNetwork(
		makeDurations([]int{60, 70, 80, 90, 100, 110, 120, 130, 140, 150}),
		0, 10,
	)

	config := makeDefaultConfig()
	session := makeTestSession(config)
	ep := &EnvironmentProfile{Net: *profile, System: SystemProfile{FDLimit: 4096, NumCPU: 4}}
	ep.TuneConfig(config, session)

	if config.DetectedNetworkEnv != int(EnvInternet) {
		t.Errorf("DetectedNetworkEnv=%d, want %d(Internet)", config.DetectedNetworkEnv, int(EnvInternet))
	}

	// 公网 target 应明显低于默认 600
	if config.ThreadNum >= 600 {
		t.Errorf("公网 threadNum=%d, 应 < 600", config.ThreadNum)
	}

	// ceiling 应 == target（非显式模式）
	if config.ThreadCeiling != config.ThreadNum {
		t.Errorf("非显式模式 ceiling(%d) != target(%d)", config.ThreadCeiling, config.ThreadNum)
	}

	netEnv := NetworkEnv(config.DetectedNetworkEnv)
	metrics := &ScanMetrics{}
	pool, err := NewAdaptivePool(config.ThreadNum, config.ThreadCeiling, func(interface{}) {}, metrics, netEnv)
	if err != nil {
		t.Fatalf("创建池失败: %v", err)
	}
	defer pool.Release()

	// 注入 12% exhaust，Internet 环境应只是 Stressed 而不是 Congested
	pool.inSlowStart = false
	pool.tune(config.ThreadNum)
	for i := 0; i < 100; i++ {
		if i < 12 {
			metrics.RecordExhausted()
		} else {
			metrics.RecordConnect(80 * time.Millisecond)
		}
	}
	pool.lastCheck.Store(0)
	pool.adjust()

	// cap 不应被砍到一半以下（Stressed 只降 15%）
	if pool.Cap() < config.ThreadNum*7/10 {
		t.Errorf("Internet 12%% exhaust 降速过猛: %d -> %d", config.ThreadNum, pool.Cap())
	}

	t.Logf("端到端 Internet: target=%d ceiling=%d cap_after_stress=%d",
		config.ThreadNum, config.ThreadCeiling, pool.Cap())
}