feat: confirm slot degradation with active probe over the slot connection

AGENTS.md

@@ -121,7 +121,7 @@ make lint   # 等价: go tool golangci-lint run --timeout 10m --verbose
 2. **应用帧协议**：HTTP/2 stream 内封装 app 帧（HANDSHAKE/DATA/DATAGRAM/FIN/RST/PADDING/COVER），`cipher_len:3 + ciphertext` 作为 CryptoRecord 边界；`protocol.ReadFrame/WriteFrame` 处理帧级别的编解码
 3. **两阶段加密**：① Bootstrap 阶段：AES-256-GCM 加密初始握手帧（含目标地址和方法协商）；② Session 阶段：协商后的 AEAD（AES-256-GCM 或 ChaCha20-Poly1305），HKDF+salt 派生 C2S/S2C 方向独立密钥，每方向独立计数器 nonce
 4. **回落对抗**：服务端首个 CryptoRecord 解密/验证失败时，回落成正常 HTML 首页（伪装为普通网站）；一旦发送 octet-stream 响应头则只能 close/RST stream；fallback 支持 5 种视觉主题 ×5 类内容页面，按 URL hash 确定性生成
-5. **端点**：`POST /v3/tcp`、`POST /v3/udp`、`POST /v3/icmp`（强制 HTTP/2），其余路径返回 fallback HTML（允许 HTTP/1.1）
+5. **端点**：`POST /v3/tcp`、`POST /v3/udp`、`POST /v3/icmp`（强制 HTTP/2），其余路径返回 fallback HTML（允许 HTTP/1.1）；`GET /v3/probe` 返回启动时预生成的随机数据（需 `x-es` 携带 master key 派生的能力令牌），供客户端主动探测 slot 连接的真实下载速度
 6. **服务端**：需要 sudo 运行（443 端口 + ICMP）；TLS 证书通过 certmagic 自动管理（Let's Encrypt ACME）或手动指定证书文件
 7. **SSRF 防护**：服务端验证 HANDSHAKE 中的 target 地址，拒绝 LAN/私有 IP 目标，防止被用作跳板攻击内网
 8. **流量整形**：`shaper` 包将帧分批打包为 CryptoRecord，填充至固定大小档位（128/512/1500 字节），支持按预算比例注入 cover traffic（随机 COVER 帧），批处理窗口默认 3ms

client/client.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"sync"
 	"time"
@@ -81,6 +82,11 @@ func New(cfg *config.ClientConfig) (*Client, error) {
 		closeIdleDone: make(chan struct{}),
 	}
 
+	probeToken, err := crypto.ProbeToken(masterKey)
+	if err != nil {
+		return nil, fmt.Errorf("probe token: %w", err)
+	}
+
 	tr, err := http2.New(http2.Config{
 		ServerURL:         cfg.ServerURL(),
 		TLSConfig:         tlsCfg,
@@ -90,6 +96,7 @@ func New(cfg *config.ClientConfig) (*Client, error) {
 		ConnLifetime:      time.Duration(cfg.Transport.ConnLifetimeSec) * time.Second,
 		ConnMaxBytes:      cfg.Transport.ConnMaxBytes,
 		Timeout:           cfg.TimeoutDuration(),
+		ProbeToken:        probeToken,
 		DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) {
 			return dialWithConfig(ctx, cfg, client.dialer, rt, network, addr)
 		},

client/proxy/stream.go

@@ -136,6 +136,14 @@ func (h *StreamHandler) openAndBootstrap(ctx context.Context, endpoint string, p
 		}
 		rw.Flush()
 
+		// Stamp the moment the bootstrap record left the client: the server
+		// answers with the response headers before dialing the origin, so the
+		// transport records the pure client<->server path RTT when they
+		// arrive (see HTTP2Stream.MarkBootstrapSent).
+		if m, ok := stream.(interface{ MarkBootstrapSent() }); ok {
+			m.MarkBootstrapSent()
+		}
+
 		return &bootstrapSession{stream: stream, sk: sk, salt: salt}, nil
 	}
 
@@ -325,14 +333,11 @@ func (h *StreamHandler) copyRemoteToLocal(rx *crypto.DecryptedReader, dst net.Co
 
 	go func() {
 		defer close(ch)
-		start := time.Now()
 		first := true
 		for {
 			frame, err := rx.ReadFrame()
 			if first {
 				first = false
-				rtt := time.Since(start)
-				stats.RecordRTT(rtt)
 				err = classifyFirstReadError(err)
 			}
 			if err != nil {

cmd/easyss/main.go

@@ -320,6 +320,8 @@ func (a *App) statsLoop() {
 				"avg_rtt", snap.AvgRTT().Round(time.Millisecond),
 				"slot_degraded", snap.SlotDegraded,
 				"slot_retired_degraded", snap.SlotRetiredDegraded,
+				"slot_probes", snap.SlotProbes,
+				"slot_probe_slow", snap.SlotProbeSlow,
 				"conn_rotated", snap.ConnRotated,
 			)
 		case <-a.statsCloser:

config/types.go

@@ -45,18 +45,22 @@ const (
 
 	// Degraded-slot detection: a slot hosting heavy streams whose download
 	// throughput stays below DegradedThroughputThreshold for
-	// DegradedPersistCycles consecutive health-check intervals is marked
-	// degraded — new streams avoid it and its idle connection is retired
-	// early instead of lingering. The mark clears after
+	// DegradedPersistCycles consecutive health-check intervals is *suspected*
+	// of degradation and gets confirmed by an active probe over the slot's
+	// own connection (see EndpointProbe). The mark clears after
 	// DegradedRecoverCycles healthy intervals. Detection only runs while
 	// the link RTT is at most DegradedMaxRTT: a congested link makes every
 	// connection slow, so retiring connections then only adds handshake
-	// churn without recovering anything.
+	// churn without recovering anything. The RTT is the pure client<->server
+	// path RTT (bootstrap round trip, origin latency excluded); a slow
+	// origin no longer suppresses detection. When the server does not
+	// support probing, the suspicion directly marks the slot (legacy
+	// behavior).
 	HealthCheckInterval         = 5 * time.Second
-	DegradedThroughputThreshold = 64 * 1024 // 64KB/s
+	DegradedThroughputThreshold = 128 * 1024 // 128KB/s
 	DegradedPersistCycles       = 3
 	DegradedRecoverCycles       = 2
-	DegradedMaxRTT              = 700 * time.Millisecond
+	DegradedMaxRTT              = 900 * time.Millisecond
 
 	// Connection rotation: long-lived TCP+TLS connections are frequently
 	// throttled by middleboxes — especially during peak hours — which is
@@ -85,7 +89,21 @@ const (
 	TCPStreamBufferSize       = 15 * 1024 // 客户端，4帧/record (4*(15360+3)=61452 < 64KB)
 	ServerTCPStreamBufferSize = 31 * 1024 // 服务端，2帧/record (2*(31744+3)=63494 < 64KB)
 
-	EndpointTCP  = "/v3/tcp"
-	EndpointUDP  = "/v3/udp"
-	EndpointICMP = "/v3/icmp"
+	EndpointTCP   = "/v3/tcp"
+	EndpointUDP   = "/v3/udp"
+	EndpointICMP  = "/v3/icmp"
+	EndpointProbe = "/v3/probe"
+
+	// Active slot probing: a slot suspected of degradation (passive
+	// throughput below DegradedThroughputThreshold) is confirmed by
+	// downloading a pre-generated random payload over the slot's own
+	// connection; only a slow probe verdict marks the slot degraded. The
+	// probe measures the client<->server path only, so a slow origin or a
+	// stalled-but-open stream no longer causes misjudgment.
+	ProbePayloadSize    = 128 * 1024       // 128KB，服务端启动时预生成
+	ProbeTimeout        = 3 * time.Second  // 单次探测超时（含透明重拨）
+	ProbeConfirmCycles  = 2                // 连续慢探测次数 → 标记 degraded
+	ProbeCooldown       = 15 * time.Second // 同一 slot 两次探测最小间隔
+	ProbeMaxPerInterval = 2                // 每个健康周期最多探测数
+	ProbeLinkRefWindow  = 60 * time.Second // 链路参考速度有效窗口
 )

crypto/kdf.go

@@ -3,6 +3,7 @@ package crypto
 import (
 	"crypto/rand"
 	"crypto/sha256"
+	"encoding/base64"
 	"errors"
 	"io"
 
@@ -18,6 +19,7 @@ const (
 	masterKDFInfo    = "easyss-v3-master"
 	bootstrapKDFInfo = "easyss-v3-bootstrap"
 	sessionKDFInfo   = "easyss-v3-session"
+	probeKDFInfo     = "easyss-v3-probe"
 )
 
 func DeriveMasterKey(password string) ([]byte, error) {
@@ -82,3 +84,16 @@ func GenerateSalt() ([]byte, error) {
 	}
 	return salt, nil
 }
+
+// ProbeToken derives the capability token for the /v3/probe endpoint.
+// Client and server derive the same 16-byte value from the master key;
+// base64url-encoded it is wire-identical to the x-es salt shape used by
+// proxy handshakes.
+func ProbeToken(masterKey []byte) (string, error) {
+	reader := hkdf.New(sha256.New, masterKey, nil, []byte(probeKDFInfo))
+	b := make([]byte, saltSize)
+	if _, err := io.ReadFull(reader, b); err != nil {
+		return "", err
+	}
+	return base64.RawURLEncoding.EncodeToString(b), nil
+}

server/handler/probe.go

@@ -0,0 +1,93 @@
+package handler
+
+import (
+	"crypto/subtle"
+	"encoding/base64"
+	"net/http"
+	"strconv"
+
+	"github.com/nange/easyss/v3/crypto"
+	"github.com/nange/easyss/v3/log"
+	"github.com/nange/easyss/v3/stats"
+)
+
+// probeChunkSize is the write granularity for the probe payload: each chunk
+// is flushed so the payload reaches the client at its real network pace
+// instead of being buffered server-side.
+const probeChunkSize = 32 * 1024
+
+// ProbeHandler serves the pre-generated random payload used by clients to
+// actively measure the download throughput of their own connection. A valid
+// request must carry the capability token derived from the master key in the
+// x-es header (same header name and wire shape as the proxy handshake salt);
+// anything else is answered with the camouflaged fallback page so the server
+// stays indistinguishable from a real site.
+type ProbeHandler struct {
+	payload []byte
+	token   []byte
+	limiter *ipRateLimiter
+}
+
+// NewProbeHandler builds the /v3/probe handler. The payload must have been
+// generated at server startup; serving the same buffer keeps the endpoint
+// cheap and uncacheable (Cache-Control: no-store).
+func NewProbeHandler(masterKey, payload []byte) (*ProbeHandler, error) {
+	tokenB64, err := crypto.ProbeToken(masterKey)
+	if err != nil {
+		return nil, err
+	}
+	token, err := base64.RawURLEncoding.DecodeString(tokenB64)
+	if err != nil {
+		return nil, err
+	}
+	return &ProbeHandler{
+		payload: payload,
+		token:   token,
+		limiter: newIPRateLimiter(),
+	}, nil
+}
+
+func (h *ProbeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	if !r.ProtoAtLeast(2, 0) {
+		ServeFallback(w, r)
+		return
+	}
+
+	tokenB64 := r.Header.Get("x-es")
+	if tokenB64 == "" {
+		ServeFallback(w, r)
+		return
+	}
+	token, err := base64.RawURLEncoding.DecodeString(tokenB64)
+	if err != nil || len(token) != len(h.token) ||
+		subtle.ConstantTimeCompare(token, h.token) != 1 {
+		ServeFallback(w, r)
+		return
+	}
+
+	// Bound probe downloads per source IP; wrong-token requests never reach
+	// this point (they get the cheap fallback page instead).
+	if !h.limiter.Allow(clientIP(r)) {
+		log.Error("[SERVER] probe rate limited", "remote", r.RemoteAddr)
+		serveReject(w, http.StatusTooManyRequests)
+		return
+	}
+
+	stats.RecordServerProbe()
+
+	w.Header().Set("Content-Type", "application/octet-stream")
+	w.Header().Set("Cache-Control", "no-store")
+	w.Header().Set("Content-Length", strconv.Itoa(len(h.payload)))
+	w.WriteHeader(http.StatusOK)
+
+	rc := http.NewResponseController(w)
+	_ = rc.Flush()
+
+	for off := 0; off < len(h.payload); off += probeChunkSize {
+		end := min(off+probeChunkSize, len(h.payload))
+		if _, err := w.Write(h.payload[off:end]); err != nil {
+			return
+		}
+		_ = rc.Flush()
+	}
+}