diff --git a/AGENTS.md b/AGENTS.md index 8b252854..f437353d 100644 --- a/AGENTS.md +++ b/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 diff --git a/client/client.go b/client/client.go index 61f9746b..c35437b3 100644 --- a/client/client.go +++ b/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) }, diff --git a/client/proxy/stream.go b/client/proxy/stream.go index d0aabcd6..ab32851c 100644 --- a/client/proxy/stream.go +++ b/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 { diff --git a/cmd/easyss/main.go b/cmd/easyss/main.go index 67358795..9eb0adde 100644 --- a/cmd/easyss/main.go +++ b/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: diff --git a/config/types.go b/config/types.go index eab4a81c..bec00007 100644 --- a/config/types.go +++ b/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 // 链路参考速度有效窗口 ) diff --git a/crypto/kdf.go b/crypto/kdf.go index f2bdc39b..9319d145 100644 --- a/crypto/kdf.go +++ b/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 +} diff --git a/server/handler/probe.go b/server/handler/probe.go new file mode 100644 index 00000000..f3ad75d5 --- /dev/null +++ b/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() + } +} diff --git a/server/handler/probe_test.go b/server/handler/probe_test.go new file mode 100644 index 00000000..506ada70 --- /dev/null +++ b/server/handler/probe_test.go @@ -0,0 +1,149 @@ +package handler + +import ( + "bytes" + "crypto/rand" + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/nange/easyss/v3/crypto" +) + +// h2Request builds an HTTP/2 request; httptest.NewRequest defaults to +// HTTP/1.1, which the probe handler (like the proxy handler) rejects with +// the fallback page. +func h2Request(method, target string) *http.Request { + req := httptest.NewRequest(method, target, nil) + req.Proto = "HTTP/2.0" + req.ProtoMajor = 2 + req.ProtoMinor = 0 + return req +} + +func newTestProbeHandler(t *testing.T) (*ProbeHandler, string, []byte) { + t.Helper() + masterKey, err := crypto.DeriveMasterKey("test-password") + if err != nil { + t.Fatal(err) + } + token, err := crypto.ProbeToken(masterKey) + if err != nil { + t.Fatal(err) + } + payload := make([]byte, 4096) + if _, err := io.ReadFull(rand.Reader, payload); err != nil { + t.Fatal(err) + } + h, err := NewProbeHandler(masterKey, payload) + if err != nil { + t.Fatal(err) + } + return h, token, payload +} + +func TestProbeHandlerValidToken(t *testing.T) { + h, token, payload := newTestProbeHandler(t) + + req := h2Request(http.MethodGet, "/v3/probe") + req.Header.Set("x-es", token) + rr := httptest.NewRecorder() + + h.ServeHTTP(rr, req) + + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want 200", rr.Code) + } + if ct := rr.Header().Get("Content-Type"); ct != "application/octet-stream" { + t.Fatalf("Content-Type = %q, want application/octet-stream", ct) + } + if cc := rr.Header().Get("Cache-Control"); cc != "no-store" { + t.Fatalf("Cache-Control = %q, want no-store", cc) + } + if cl := rr.Header().Get("Content-Length"); cl != "4096" { + t.Fatalf("Content-Length = %q, want 4096", cl) + } + if !bytes.Equal(rr.Body.Bytes(), payload) { + t.Fatal("body differs from the pre-generated payload") + } + + // A second request must serve the same buffer. + req2 := h2Request(http.MethodGet, "/v3/probe") + req2.Header.Set("x-es", token) + rr2 := httptest.NewRecorder() + h.ServeHTTP(rr2, req2) + if !bytes.Equal(rr2.Body.Bytes(), payload) { + t.Fatal("second response differs from the pre-generated payload") + } +} + +func TestProbeHandlerRejectsInvalidToken(t *testing.T) { + h, _, _ := newTestProbeHandler(t) + + cases := []struct { + name string + token string + }{ + {"missing token", ""}, + {"garbage token", "not-a-valid-token"}, + {"wrong token", "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + req := h2Request(http.MethodGet, "/v3/probe") + if tc.token != "" { + req.Header.Set("x-es", tc.token) + } + rr := httptest.NewRecorder() + h.ServeHTTP(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want 200 fallback", rr.Code) + } + if ct := rr.Header().Get("Content-Type"); !strings.HasPrefix(ct, "text/html") { + t.Fatalf("Content-Type = %q, want fallback HTML", ct) + } + if rr.Body.Len() == 4096 { + t.Fatal("fallback response must not contain the probe payload") + } + }) + } +} + +func TestProbeHandlerRejectsHTTP1(t *testing.T) { + h, token, _ := newTestProbeHandler(t) + + req := httptest.NewRequest(http.MethodGet, "/v3/probe", nil) // HTTP/1.1 + req.Header.Set("x-es", token) + rr := httptest.NewRecorder() + + h.ServeHTTP(rr, req) + + if ct := rr.Header().Get("Content-Type"); !strings.HasPrefix(ct, "text/html") { + t.Fatalf("Content-Type = %q, want fallback HTML for HTTP/1.1", ct) + } +} + +func TestProbeHandlerRateLimit(t *testing.T) { + h, token, _ := newTestProbeHandler(t) + + // Drain the per-IP bucket (capacity 100), then the next request is + // rejected with 429. + ip := "203.0.113.7" + for i := 0; i < 100; i++ { + if !h.limiter.Allow(ip) { + t.Fatalf("bucket drained earlier than expected at request %d", i+1) + } + } + + req := h2Request(http.MethodGet, "/v3/probe") + req.Header.Set("x-es", token) + req.RemoteAddr = ip + ":12345" + rr := httptest.NewRecorder() + h.ServeHTTP(rr, req) + + if rr.Code != http.StatusTooManyRequests { + t.Fatalf("status = %d, want 429", rr.Code) + } +} diff --git a/server/server.go b/server/server.go index ffdb1b4e..e9a81a4e 100644 --- a/server/server.go +++ b/server/server.go @@ -6,6 +6,7 @@ import ( "crypto/tls" "encoding/hex" "fmt" + "io" stdlog "log" "net/http" "os" @@ -145,6 +146,7 @@ func (s *Server) statsLoop() { "icmp", snap.ServerICMPStreams, "hserr", snap.ServerHandshakeErrors, "fallback", snap.ServerFallbackPages, + "probe", snap.ServerProbes, "padding", stats.HumanBytes(snap.PaddingBytes), "records", snap.RecordsWritten, ) @@ -298,6 +300,15 @@ func (s *Server) Start() error { NextProxy: np, }) + probePayload := make([]byte, sharedconfig.ProbePayloadSize) + if _, err := io.ReadFull(rand.Reader, probePayload); err != nil { + return fmt.Errorf("generate probe payload: %w", err) + } + probeHandler, err := handler.NewProbeHandler(masterKey, probePayload) + if err != nil { + return fmt.Errorf("probe handler: %w", err) + } + s.mux = http.NewServeMux() s.mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { handler.ServeFallback(w, r) @@ -305,10 +316,11 @@ func (s *Server) Start() error { s.mux.Handle(sharedconfig.EndpointTCP, proxyHandler) s.mux.Handle(sharedconfig.EndpointUDP, proxyHandler) s.mux.Handle(sharedconfig.EndpointICMP, proxyHandler) + s.mux.Handle(sharedconfig.EndpointProbe, probeHandler) s.httpServer = buildHTTPServer(cfg, tlsConfig, s.mux, timeout) - log.Info("[SERVER] listening", "addr", s.cfg.Listen, "routes", []string{"/", sharedconfig.EndpointTCP, sharedconfig.EndpointUDP, sharedconfig.EndpointICMP}) + log.Info("[SERVER] listening", "addr", s.cfg.Listen, "routes", []string{"/", sharedconfig.EndpointTCP, sharedconfig.EndpointUDP, sharedconfig.EndpointICMP, sharedconfig.EndpointProbe}) s.statsDone = make(chan struct{}) go s.statsLoop() return s.httpServer.ListenAndServeTLS("", "") diff --git a/stats/stats.go b/stats/stats.go index cfbf19ed..4c5115a1 100644 --- a/stats/stats.go +++ b/stats/stats.go @@ -44,12 +44,15 @@ type stats struct { bulkFallback atomic.Int64 // Transport health (client-side) - slotDegraded atomic.Int64 - slotRetiredDegraded atomic.Int64 - connRotated atomic.Int64 + slotDegraded atomic.Int64 + slotRetiredDegraded atomic.Int64 + connRotated atomic.Int64 + slotProbes atomic.Int64 + slotProbeSlow atomic.Int64 + slotProbeUnsupported atomic.Int64 rttMu sync.Mutex - rttEWMA int64 // nanoseconds, EWMA-smoothed RTT + rttEWMA int64 // nanoseconds, EWMA-smoothed pure path RTT rttCount atomic.Int64 // Speed tracking (bytes/sec, EWMA-smoothed) @@ -64,6 +67,7 @@ type stats struct { serverICMPStreams atomic.Int64 serverHandshakeErrors atomic.Int64 serverFallbackPages atomic.Int64 + serverProbes atomic.Int64 // startTime keeps the monotonic clock reading so time.Since stays // immune to wall-clock adjustments; nil means no active session. @@ -98,9 +102,18 @@ func RecordBulkFallback() { g.bulkFallback.Add(1) } func RecordSlotDegraded() { g.slotDegraded.Add(1) } func RecordSlotRetiredDegraded() { g.slotRetiredDegraded.Add(1) } func RecordConnRotated() { g.connRotated.Add(1) } +func RecordSlotProbe() { g.slotProbes.Add(1) } +func RecordSlotProbeSlow() { g.slotProbeSlow.Add(1) } +func RecordSlotProbeUnsupported() { + g.slotProbeUnsupported.Add(1) +} const rttAlpha = 0.35 +// RecordRTT feeds a pure client<->server path RTT sample: the time between +// the client flushing its bootstrap record (or the probe request reaching +// the server) and the response arriving. The server commits its response +// before dialing the origin, so origin latency never enters the sample. func RecordRTT(d time.Duration) { g.rttMu.Lock() if g.rttCount.Load() == 0 { @@ -117,6 +130,7 @@ func RecordServerUDPStream() { g.serverUDPStreams.Add(1) } func RecordServerICMPStream() { g.serverICMPStreams.Add(1) } func RecordServerHandshakeError() { g.serverHandshakeErrors.Add(1) } func RecordServerFallbackPage() { g.serverFallbackPages.Add(1) } +func RecordServerProbe() { g.serverProbes.Add(1) } // --- session lifecycle --- @@ -154,6 +168,9 @@ func ResetCounters() { g.slotDegraded.Store(0) g.slotRetiredDegraded.Store(0) g.connRotated.Store(0) + g.slotProbes.Store(0) + g.slotProbeSlow.Store(0) + g.slotProbeUnsupported.Store(0) g.rttMu.Lock() g.rttEWMA = 0 @@ -170,6 +187,7 @@ func ResetCounters() { g.serverICMPStreams.Store(0) g.serverHandshakeErrors.Store(0) g.serverFallbackPages.Store(0) + g.serverProbes.Store(0) } // --- snapshot --- @@ -198,15 +216,19 @@ type Snapshot struct { ServerICMPStreams int64 `json:"server_icmp_streams,omitempty"` ServerHandshakeErrors int64 `json:"server_handshake_errors,omitempty"` ServerFallbackPages int64 `json:"server_fallback_pages,omitempty"` + ServerProbes int64 `json:"server_probes,omitempty"` PriorityStreamsOpened int64 `json:"priority_streams_opened"` BulkStreamsOpened int64 `json:"bulk_streams_opened"` PriorityFallback int64 `json:"priority_fallback"` BulkFallback int64 `json:"bulk_fallback"` // Transport health (client-side only; zero on server) - SlotDegraded int64 `json:"slot_degraded"` - SlotRetiredDegraded int64 `json:"slot_retired_degraded"` - ConnRotated int64 `json:"conn_rotated"` + SlotDegraded int64 `json:"slot_degraded"` + SlotRetiredDegraded int64 `json:"slot_retired_degraded"` + ConnRotated int64 `json:"conn_rotated"` + SlotProbes int64 `json:"slot_probes"` + SlotProbeSlow int64 `json:"slot_probe_slow"` + SlotProbeUnsupported int64 `json:"slot_probe_unsupported"` // Speed UploadSpeed int64 `json:"upload_speed"` @@ -285,6 +307,7 @@ func Collect() Snapshot { ServerICMPStreams: g.serverICMPStreams.Load(), ServerHandshakeErrors: g.serverHandshakeErrors.Load(), ServerFallbackPages: g.serverFallbackPages.Load(), + ServerProbes: g.serverProbes.Load(), PriorityStreamsOpened: g.priorityStreamsOpened.Load(), BulkStreamsOpened: g.bulkStreamsOpened.Load(), PriorityFallback: g.priorityFallback.Load(), @@ -292,6 +315,9 @@ func Collect() Snapshot { SlotDegraded: g.slotDegraded.Load(), SlotRetiredDegraded: g.slotRetiredDegraded.Load(), ConnRotated: g.connRotated.Load(), + SlotProbes: g.slotProbes.Load(), + SlotProbeSlow: g.slotProbeSlow.Load(), + SlotProbeUnsupported: g.slotProbeUnsupported.Load(), UploadSpeed: upSpeed, DownloadSpeed: downSpeed, UploadSpeedHuman: HumanBytes(upSpeed) + "/s", diff --git a/transport/http2/client.go b/transport/http2/client.go index 555f53e8..2f3f1ca5 100644 --- a/transport/http2/client.go +++ b/transport/http2/client.go @@ -42,6 +42,10 @@ type Config struct { ConnMaxBytes int64 // max bytes carried by a connection in either direction before rotation (0: default) Timeout time.Duration DialContext func(ctx context.Context, network, addr string) (net.Conn, error) + // ProbeToken is the capability token for the server's /v3/probe + // endpoint (derived from the master key). Empty disables active + // probing, leaving passive-only degraded detection. + ProbeToken string } func New(cfg Config) (*HTTP2Transport, error) { @@ -96,13 +100,23 @@ func New(cfg Config) (*HTTP2Transport, error) { sched := newScheduler(maxSlots, slots, threshold, prioritySlots) + lc := &slotLifecycle{ + sched: sched, + connLifetime: connLifetime, + connMaxBytes: connMaxBytes, + } + if cfg.ProbeToken != "" { + prober := &slotProber{ + serverURL: cfg.ServerURL, + token: cfg.ProbeToken, + payloadSize: int64(sharedconfig.ProbePayloadSize), + } + lc.probeFunc = prober.probe + } + tr := &HTTP2Transport{ - sched: sched, - lifecycle: &slotLifecycle{ - sched: sched, - connLifetime: connLifetime, - connMaxBytes: connMaxBytes, - }, + sched: sched, + lifecycle: lc, serverURL: cfg.ServerURL, ctx: ctx, cancel: cancel, diff --git a/transport/http2/lifecycle.go b/transport/http2/lifecycle.go index ee3c6db8..662fb3ab 100644 --- a/transport/http2/lifecycle.go +++ b/transport/http2/lifecycle.go @@ -11,13 +11,24 @@ import ( ) // slotLifecycle owns the per-slot connection lifecycle: a health loop that -// samples download throughput for degraded detection, and connection -// rotation once the lifetime or bytes limit is exceeded. It reuses the -// scheduler's pool management to retire idle degraded slots. +// samples download throughput for degradation suspicion and confirms it with +// an active probe over the slot's own connection, plus connection rotation +// once the lifetime or bytes limit is exceeded. It reuses the scheduler's +// pool management to retire idle degraded slots. type slotLifecycle struct { sched *slotScheduler connLifetime time.Duration // max age of a connection before rotation connMaxBytes int64 // max bytes per connection before rotation + + // probeFunc actively measures one slot's connection throughput; nil + // disables probing (no probe token configured). + probeFunc func(ctx context.Context, slot *transportSlot) (speedBps float64, verdict probeVerdict) + + // Probe state, touched only from the health loop goroutine. + probeUnsupported bool // server does not serve /v3/probe: fall back to passive-only detection + unsupportedCount int // unsupported probe verdicts seen + linkRefSpeed float64 // best probe speed observed on this link recently (bytes/s) + linkRefAt time.Time } // run drives the periodic health evaluation until ctx is cancelled. @@ -35,9 +46,9 @@ func (lc *slotLifecycle) run(ctx context.Context) { } } -// evaluate walks the live slots: download throughput feeds the degraded -// detector, connection age/bytes feed rotation, and idle degraded slots are -// retired. +// evaluate walks the live slots: download throughput feeds the degradation +// suspicion detector, active probes confirm suspicions, connection age/bytes +// feed rotation, and idle degraded slots are retired. func (lc *slotLifecycle) evaluate(interval time.Duration) { // A congested link (high RTT) makes every connection slow: marking or // retiring slots then only adds handshake churn without recovering @@ -59,16 +70,22 @@ func (lc *slotLifecycle) evaluate(interval time.Duration) { i-- } } + + lc.evaluateProbes(linkOK) } -// evaluateSlotHealth updates one slot's degraded state from its recent -// download throughput. Only slots hosting heavy streams are considered — -// idle or short-lived slots naturally carry zero throughput. The mark is -// set after DegradedPersistCycles consecutive slow intervals and cleared -// after DegradedRecoverCycles healthy ones. +// evaluateSlotHealth updates one slot's suspicion from its recent download +// throughput. Only slots hosting heavy streams are considered — idle or +// short-lived slots naturally carry zero throughput. When probing is active +// (server serves /v3/probe), persistent low throughput only marks the slot +// as suspected and an active probe confirms or refutes it; without probing +// (server does not support it, or no probe token configured), the suspicion +// directly marks the slot degraded (legacy behavior). The mark is cleared +// after DegradedRecoverCycles healthy intervals in both modes. func (lc *slotLifecycle) evaluateSlotHealth(idx int, s *transportSlot, interval time.Duration, linkOK bool) { if s.heavy.Load() == 0 { s.lastHeavy = 0 + s.suspected = false return } if s.lastHeavy == 0 { @@ -100,6 +117,7 @@ func (lc *slotLifecycle) evaluateSlotHealth(idx int, s *transportSlot, interval if throughput >= int64(sharedconfig.DegradedThroughputThreshold) { s.lowCycles = 0 + s.suspected = false if s.degraded.Load() { s.recoverCycles++ if s.recoverCycles >= sharedconfig.DegradedRecoverCycles { @@ -114,10 +132,104 @@ func (lc *slotLifecycle) evaluateSlotHealth(idx int, s *transportSlot, interval s.recoverCycles = 0 s.lowCycles++ if s.lowCycles >= sharedconfig.DegradedPersistCycles && !s.degraded.Load() { - s.degraded.Store(true) s.lowCycles = 0 - stats.RecordSlotDegraded() - log.Info("[TRANSPORT] slot degraded", "slot", idx, "throughput_kb_s", throughput/1024) + if lc.probeUnsupported || lc.probeFunc == nil { + // No active probing (server does not serve /v3/probe, or no + // probe token configured): the passive sampler marks the slot + // degraded directly, as before probing existed. + s.degraded.Store(true) + stats.RecordSlotDegraded() + log.Info("[TRANSPORT] slot degraded", "slot", idx, "throughput_kb_s", throughput/1024) + } else { + s.suspected = true + } + } +} + +// evaluateProbes confirms degradation suspicions with active probes. Only +// slots suspected by the passive sampler are probed, and only while the link +// RTT is healthy (a congested link makes every probe slow). Each tick probes +// at most ProbeMaxPerInterval slots; a slot is re-probed at most once per +// ProbeCooldown. +func (lc *slotLifecycle) evaluateProbes(linkOK bool) { + if !linkOK || lc.probeUnsupported || lc.probeFunc == nil { + return + } + + now := time.Now() + live := int(lc.sched.liveCount.Load()) + probed := 0 + for i := 0; i < live && probed < sharedconfig.ProbeMaxPerInterval; i++ { + s := lc.sched.slots[i] + if !s.suspected || s.degraded.Load() { + continue + } + if now.Sub(s.lastProbeAt) < sharedconfig.ProbeCooldown { + continue + } + lc.probeSlot(i, s, now) + probed++ + } +} + +// probeSlot runs one probe and folds its verdict into the slot's degraded +// state: +// - slow: confirmed after ProbeConfirmCycles consecutive slow probes — +// unless the link reference speed shows the whole link is the +// bottleneck, in which case the slot is not to blame; +// - fast: the connection is healthy (slow traffic is an origin or stream +// property, not a connection property); clears suspicion; +// - inconclusive: no state change; +// - unsupported: after two such verdicts the server is treated as not +// serving the probe endpoint and detection falls back to passive-only. +func (lc *slotLifecycle) probeSlot(idx int, s *transportSlot, now time.Time) { + ctx, cancel := context.WithTimeout(context.Background(), sharedconfig.ProbeTimeout) + defer cancel() + + speed, verdict := lc.probeFunc(ctx, s) + s.lastProbeAt = now + stats.RecordSlotProbe() + + switch verdict { + case probeFast: + s.probeLowCycles = 0 + s.suspected = false + // The link reference is the best throughput this link achieved + // recently; it only refreshes from fast probes. + if speed > lc.linkRefSpeed || now.Sub(lc.linkRefAt) > sharedconfig.ProbeLinkRefWindow { + lc.linkRefSpeed = speed + lc.linkRefAt = now + } + case probeSlow: + // A fresh link reference below the degraded threshold means the + // whole link is the bottleneck right now: every connection is + // slow, so blaming this slot would only add handshake churn. + if now.Sub(lc.linkRefAt) <= sharedconfig.ProbeLinkRefWindow && + lc.linkRefSpeed < float64(sharedconfig.DegradedThroughputThreshold) { + s.probeLowCycles = 0 + s.suspected = false + return + } + stats.RecordSlotProbeSlow() + s.probeLowCycles++ + if s.probeLowCycles >= sharedconfig.ProbeConfirmCycles { + s.probeLowCycles = 0 + s.suspected = false + s.degraded.Store(true) + stats.RecordSlotDegraded() + log.Info("[TRANSPORT] slot degraded", "slot", idx, "probe_kb_s", int64(speed)/1024) + } + case probeUnsupported: + lc.unsupportedCount++ + if lc.unsupportedCount >= 2 { + lc.probeUnsupported = true + stats.RecordSlotProbeUnsupported() + log.Info("[TRANSPORT] server does not serve /v3/probe, falling back to passive detection") + } + case probeInconclusive: + // Dead connection (stream errors and rotation handle it) or + // transient rejection (429): keep the current state, re-probe + // once the cooldown elapses. } } diff --git a/transport/http2/lifecycle_test.go b/transport/http2/lifecycle_test.go index 07dc52ff..29e66c2f 100644 --- a/transport/http2/lifecycle_test.go +++ b/transport/http2/lifecycle_test.go @@ -1,6 +1,7 @@ package http2 import ( + "context" "net/http" "testing" "time" @@ -10,7 +11,9 @@ import ( func TestEvaluateSlotHealth(t *testing.T) { interval := sharedconfig.HealthCheckInterval - lc := &slotLifecycle{} + // Legacy mode (probeUnsupported): the passive sampler marks slots + // degraded directly, as before probing existed. + lc := &slotLifecycle{probeUnsupported: true} newHeavySlot := func() *transportSlot { s := &transportSlot{t: &http.Transport{}} @@ -117,6 +120,273 @@ func TestEvaluateSlotHealth(t *testing.T) { }) } +// newTestLifecycle builds a lifecycle over n live slots with the given +// probe function (nil disables probing). +func newTestLifecycle(n int, probe func(context.Context, *transportSlot) (float64, probeVerdict)) (*slotLifecycle, *slotScheduler) { + slots := make([]*transportSlot, n) + for i := range slots { + slots[i] = &transportSlot{t: &http.Transport{}} + } + sch := newScheduler(n, slots, 8, 1) + sch.liveCount.Store(int32(n)) + lc := &slotLifecycle{sched: sch, probeFunc: probe} + return lc, sch +} + +func TestSuspicionInsteadOfDirectMark(t *testing.T) { + interval := sharedconfig.HealthCheckInterval + // Probe mode: a probe function is configured, so low passive + // throughput only raises suspicion; the degraded mark is confirmed by + // probes. (The fake is never called here — only the passive sampler + // runs.) + lc := &slotLifecycle{probeFunc: func(context.Context, *transportSlot) (float64, probeVerdict) { + return 0, probeInconclusive + }} + s := &transportSlot{t: &http.Transport{}} + s.heavy.Store(1) + + low := func() { + s.bytesRecv.Add(10 * 1024) + lc.evaluateSlotHealth(0, s, interval, true) + } + high := func() { + s.bytesRecv.Add(2 * 1024 * 1024) + lc.evaluateSlotHealth(0, s, interval, true) + } + + low() // baseline reset + for i := 0; i < sharedconfig.DegradedPersistCycles; i++ { + low() + } + if s.degraded.Load() { + t.Fatal("probe mode must not mark degraded directly") + } + if !s.suspected { + t.Fatal("expected suspicion after persistent low throughput") + } + + // A healthy interval clears the suspicion without any probe. + high() + if s.suspected { + t.Fatal("expected suspicion cleared by healthy throughput") + } +} + +func TestNoProbeFuncFallsBackToPassive(t *testing.T) { + interval := sharedconfig.HealthCheckInterval + // No probe function configured (e.g. no probe token): the passive + // sampler keeps its legacy direct marking. + lc := &slotLifecycle{} + s := &transportSlot{t: &http.Transport{}} + s.heavy.Store(1) + + lc.evaluateSlotHealth(0, s, interval, true) // baseline reset + for i := 0; i < sharedconfig.DegradedPersistCycles; i++ { + s.bytesRecv.Add(10 * 1024) + lc.evaluateSlotHealth(0, s, interval, true) + } + if !s.degraded.Load() { + t.Fatal("expected legacy degraded marking without a probe function") + } + if s.suspected { + t.Fatal("legacy mode must not set suspicion") + } +} + +func TestProbeConfirmDegraded(t *testing.T) { + lc, sch := newTestLifecycle(2, func(context.Context, *transportSlot) (float64, probeVerdict) { + return 10 * 1024, probeSlow // 10KB/s, well below 64KB/s + }) + s := sch.slots[0] + s.suspected = true + + lc.evaluateProbes(true) + if s.degraded.Load() { + t.Fatal("degraded after a single slow probe") + } + if s.probeLowCycles != 1 { + t.Fatalf("probeLowCycles = %d, want 1", s.probeLowCycles) + } + + s.lastProbeAt = time.Time{} // bypass cooldown + lc.evaluateProbes(true) + if !s.degraded.Load() { + t.Fatal("expected degraded after ProbeConfirmCycles slow probes") + } + if s.suspected { + t.Fatal("expected suspicion cleared once degraded") + } +} + +func TestProbeFastClearsSuspicion(t *testing.T) { + const fastSpeed = 10 * 1024 * 1024 // 10MB/s + calls := 0 + lc, sch := newTestLifecycle(1, func(context.Context, *transportSlot) (float64, probeVerdict) { + calls++ + return fastSpeed, probeFast + }) + s := sch.slots[0] + s.suspected = true + + lc.evaluateProbes(true) + + if calls != 1 { + t.Fatalf("probe calls = %d, want 1", calls) + } + if s.suspected { + t.Fatal("fast probe must clear suspicion") + } + if s.degraded.Load() { + t.Fatal("fast probe must not mark degraded") + } + if lc.linkRefSpeed != fastSpeed { + t.Fatalf("linkRefSpeed = %v, want %v", lc.linkRefSpeed, fastSpeed) + } +} + +func TestProbeSlowRespectsLinkReference(t *testing.T) { + lc, sch := newTestLifecycle(1, func(context.Context, *transportSlot) (float64, probeVerdict) { + return 30 * 1024, probeSlow + }) + s := sch.slots[0] + + // A fresh link reference below the degraded threshold means the whole + // link is the bottleneck: the slow probe must not blame the slot. + lc.linkRefSpeed = 32 * 1024 + lc.linkRefAt = time.Now() + s.suspected = true + lc.evaluateProbes(true) + if s.degraded.Load() || s.probeLowCycles != 0 || s.suspected { + t.Fatal("must not mark or keep suspicion while the link itself is slow") + } + + // A healthy link reference: the slot's connection is to blame. + lc.linkRefSpeed = 1024 * 1024 + lc.linkRefAt = time.Now() + s.suspected = true + s.lastProbeAt = time.Time{} + lc.evaluateProbes(true) + s.lastProbeAt = time.Time{} + lc.evaluateProbes(true) + if !s.degraded.Load() { + t.Fatal("expected degraded with a healthy link reference") + } +} + +func TestProbeUnsupportedFallsBackToPassive(t *testing.T) { + lc, sch := newTestLifecycle(1, func(context.Context, *transportSlot) (float64, probeVerdict) { + return 0, probeUnsupported + }) + s := sch.slots[0] + s.suspected = true + + lc.evaluateProbes(true) + if lc.probeUnsupported { + t.Fatal("unsupported too early after a single verdict") + } + s.lastProbeAt = time.Time{} + lc.evaluateProbes(true) + if !lc.probeUnsupported { + t.Fatal("expected probeUnsupported after two verdicts") + } + + // The passive sampler now marks degraded directly (legacy behavior). + interval := sharedconfig.HealthCheckInterval + s.suspected = false + s.heavy.Store(1) + lc.evaluateSlotHealth(0, s, interval, true) // baseline reset + for i := 0; i < sharedconfig.DegradedPersistCycles; i++ { + s.bytesRecv.Add(10 * 1024) + lc.evaluateSlotHealth(0, s, interval, true) + } + if !s.degraded.Load() { + t.Fatal("expected legacy degraded marking after fallback") + } +} + +func TestProbeInconclusiveKeepsState(t *testing.T) { + lc, sch := newTestLifecycle(1, func(context.Context, *transportSlot) (float64, probeVerdict) { + return 0, probeInconclusive + }) + s := sch.slots[0] + s.suspected = true + + lc.evaluateProbes(true) + + if !s.suspected { + t.Fatal("inconclusive probe must keep suspicion") + } + if s.probeLowCycles != 0 { + t.Fatalf("probeLowCycles = %d, want 0", s.probeLowCycles) + } + if lc.probeUnsupported { + t.Fatal("inconclusive must not count as unsupported") + } +} + +func TestProbeCooldown(t *testing.T) { + calls := 0 + lc, sch := newTestLifecycle(1, func(context.Context, *transportSlot) (float64, probeVerdict) { + calls++ + return 10 * 1024, probeSlow + }) + s := sch.slots[0] + s.suspected = true + s.lastProbeAt = time.Now() // probed moments ago + + lc.evaluateProbes(true) + + if calls != 0 { + t.Fatal("probe must be skipped within the cooldown") + } +} + +func TestProbeMaxPerInterval(t *testing.T) { + calls := 0 + lc, sch := newTestLifecycle(3, func(context.Context, *transportSlot) (float64, probeVerdict) { + calls++ + return 10 * 1024, probeSlow + }) + for i := 0; i < 3; i++ { + sch.slots[i].suspected = true + } + + lc.evaluateProbes(true) + + if calls != sharedconfig.ProbeMaxPerInterval { + t.Fatalf("probe calls = %d, want %d", calls, sharedconfig.ProbeMaxPerInterval) + } + if sch.slots[2].probeLowCycles != 0 { + t.Fatal("the third suspect must wait for the next tick") + } +} + +func TestProbesPausedOnCongestedLink(t *testing.T) { + calls := 0 + lc, sch := newTestLifecycle(1, func(context.Context, *transportSlot) (float64, probeVerdict) { + calls++ + return 10 * 1024, probeSlow + }) + sch.slots[0].suspected = true + + lc.evaluateProbes(false) + + if calls != 0 { + t.Fatal("no probes while the link is congested") + } +} + +func TestProbesDisabledWithoutProbeFunc(t *testing.T) { + lc, sch := newTestLifecycle(1, nil) + sch.slots[0].suspected = true + + lc.evaluateProbes(true) + + if sch.slots[0].probeLowCycles != 0 { + t.Fatal("no probing without a probe function") + } +} + func TestRotationDue(t *testing.T) { now := time.Now() t.Run("lifetime exceeded", func(t *testing.T) { diff --git a/transport/http2/probe.go b/transport/http2/probe.go new file mode 100644 index 00000000..b177654d --- /dev/null +++ b/transport/http2/probe.go @@ -0,0 +1,130 @@ +package http2 + +import ( + "context" + "net/http" + "time" + + sharedconfig "github.com/nange/easyss/v3/config" + "github.com/nange/easyss/v3/stats" +) + +// probeVerdict classifies a single probe result. +type probeVerdict int + +const ( + // probeInconclusive: the probe failed before any body bytes arrived + // (RoundTrip error, non-200 status): the connection is either dead + // (handled by stream errors/rotation) or transiently rejected (429), + // so no verdict is produced. + probeInconclusive probeVerdict = iota + // probeFast: the payload was delivered at or above the degraded + // throughput threshold. + probeFast + // probeSlow: no bytes arrived within the probe timeout, or the body + // throughput stayed below the degraded threshold. + probeSlow + // probeUnsupported: the server answered 200 but not with the probe + // payload (wrong Content-Type/Content-Length), i.e. it does not serve + // the /v3/probe endpoint and the client must fall back to passive + // detection. + probeUnsupported +) + +// maxProbeSpeed caps the measured throughput so a single instantaneous +// (elapsed==0) probe cannot latch the link reference speed to an absurd +// value. +const maxProbeSpeed = 1 << 30 // 1GB/s + +// slotProber actively measures the download throughput of one slot's own +// connection by downloading the server's pre-generated random payload +// through the slot's http.Transport (MaxConnsPerHost=1 pins the transport +// to that single connection, so the probe bytes traverse exactly the +// connection under test). +type slotProber struct { + serverURL string + token string + payloadSize int64 +} + +// probe downloads the probe payload over the slot's connection and reports +// the body throughput (excluding TTFB: timing starts with the first body +// chunk). The verdict is decided against the absolute degraded threshold; +// the lifecycle applies the link-reference refinement on top. +func (p *slotProber) probe(ctx context.Context, slot *transportSlot) (float64, probeVerdict) { + probeCtx, cancel := context.WithTimeout(ctx, sharedconfig.ProbeTimeout) + defer cancel() + + req, err := http.NewRequestWithContext(probeCtx, http.MethodGet, p.serverURL+sharedconfig.EndpointProbe, nil) + if err != nil { + return 0, probeInconclusive + } + req.Header.Set("x-es", p.token) + req.Header.Set("Cache-Control", "no-store") + req.Header.Set("User-Agent", chromeUserAgent()) + + resp, err := slot.t.RoundTrip(req) + if err != nil { + // RoundTrip failed (dial/TLS/stream error, or the timeout expired + // before the response headers): the connection is dead or the + // server is unreachable — no verdict, stream errors and rotation + // handle those. A healthy redial inside RoundTrip completes the + // request, so a fresh connection is never misjudged as slow. + return 0, probeInconclusive + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + // Rejection (e.g. 429 rate limit): transient, re-probe later. + return 0, probeInconclusive + } + if resp.ContentLength != p.payloadSize || + resp.Header.Get("Content-Type") != "application/octet-stream" { + // The server answered 200 but not with the probe payload (e.g. a + // fallback HTML page from a server without /v3/probe). + return 0, probeUnsupported + } + + buf := make([]byte, 32*1024) + var total int64 + var start time.Time + timed := false + ttfbStart := time.Now() + for { + n, rErr := resp.Body.Read(buf) + if n > 0 { + total += int64(n) + if !timed { + timed = true + start = time.Now() + // The response headers have arrived and the server writes the + // payload immediately (no origin involved), so the time to the + // first body chunk is the pure path RTT — same measurement + // basis as the per-request bootstrap round trip. + stats.RecordRTT(time.Since(ttfbStart)) + } + } + if rErr != nil { + break // EOF, timeout or connection error: measure what arrived + } + } + + if total == 0 { + // Nothing arrived within the probe timeout (or the body was cut + // short immediately): on any sane link the first bytes of 128KB + // arrive well within 3s, so treat this as slow evidence. + return 0, probeSlow + } + + var speed float64 = maxProbeSpeed + if elapsed := time.Since(start); elapsed > 0 { + speed = float64(total) / elapsed.Seconds() + if speed > maxProbeSpeed { + speed = maxProbeSpeed + } + } + if speed < float64(sharedconfig.DegradedThroughputThreshold) { + return speed, probeSlow + } + return speed, probeFast +} diff --git a/transport/http2/probe_test.go b/transport/http2/probe_test.go new file mode 100644 index 00000000..6c463139 --- /dev/null +++ b/transport/http2/probe_test.go @@ -0,0 +1,140 @@ +package http2 + +import ( + "context" + "crypto/tls" + "net/http" + "net/http/httptest" + "testing" + + sharedconfig "github.com/nange/easyss/v3/config" + "github.com/nange/easyss/v3/crypto" + "github.com/nange/easyss/v3/server/handler" + "github.com/nange/easyss/v3/stats" +) + +const testProbePayloadSize = 4096 + +// newProbeServer starts a real TLS server serving the probe endpoint. +func newProbeServer(t *testing.T) (*httptest.Server, string) { + t.Helper() + masterKey, err := crypto.DeriveMasterKey("test-password") + if err != nil { + t.Fatal(err) + } + token, err := crypto.ProbeToken(masterKey) + if err != nil { + t.Fatal(err) + } + payload := make([]byte, testProbePayloadSize) + for i := range payload { + payload[i] = byte(i) + } + h, err := handler.NewProbeHandler(masterKey, payload) + if err != nil { + t.Fatal(err) + } + ts := httptest.NewUnstartedServer(h) + ts.EnableHTTP2 = true + ts.StartTLS() + t.Cleanup(ts.Close) + return ts, token +} + +// newProbeSlot builds a slot whose transport talks plain TLS to the test +// server (the production transport uses uTLS, which is irrelevant here). +func newProbeSlot() *transportSlot { + tr := &http.Transport{ + TLSClientConfig: &tls.Config{InsecureSkipVerify: true}, //nolint:gosec // test server cert + ForceAttemptHTTP2: true, + MaxConnsPerHost: 1, + } + return &transportSlot{t: tr} +} + +func TestSlotProberFast(t *testing.T) { + ts, token := newProbeServer(t) + prober := &slotProber{serverURL: ts.URL, token: token, payloadSize: testProbePayloadSize} + + stats.ResetCounters() + speed, verdict := prober.probe(context.Background(), newProbeSlot()) + + if verdict != probeFast { + t.Fatalf("verdict = %v, want probeFast (speed %v)", verdict, speed) + } + if speed < float64(sharedconfig.DegradedThroughputThreshold) { + t.Fatalf("speed %v below the degraded threshold", speed) + } + // A successful probe must feed a pure path RTT sample (response headers + // arrived -> first body chunk), same basis as the per-request sampling. + if got := stats.Collect().RTTCount; got != 1 { + t.Fatalf("RTTCount = %d, want 1 after a fast probe", got) + } +} + +func TestSlotProberSlowOnEmptyBody(t *testing.T) { + // A 200 octet-stream response that delivers nothing within the probe + // timeout counts as slow evidence. + ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/octet-stream") + w.Header().Set("Content-Length", "4096") + w.WriteHeader(http.StatusOK) + _ = http.NewResponseController(w).Flush() + })) + t.Cleanup(ts.Close) + prober := &slotProber{serverURL: ts.URL, token: "unused", payloadSize: testProbePayloadSize} + + speed, verdict := prober.probe(context.Background(), newProbeSlot()) + + if verdict != probeSlow { + t.Fatalf("verdict = %v, want probeSlow (speed %v)", verdict, speed) + } + if speed != 0 { + t.Fatalf("speed = %v, want 0", speed) + } +} + +func TestSlotProberUnsupportedOnHTML(t *testing.T) { + // A 200 page that is not the probe payload (e.g. an old server's + // fallback HTML) marks the probe unsupported. + ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/html; charset=utf-8") + _, _ = w.Write([]byte("fallback")) + })) + t.Cleanup(ts.Close) + prober := &slotProber{serverURL: ts.URL, token: "unused", payloadSize: testProbePayloadSize} + + _, verdict := prober.probe(context.Background(), newProbeSlot()) + + if verdict != probeUnsupported { + t.Fatalf("verdict = %v, want probeUnsupported", verdict) + } +} + +func TestSlotProberInconclusiveOnErrorStatus(t *testing.T) { + ts := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "rate limited", http.StatusTooManyRequests) + })) + t.Cleanup(ts.Close) + prober := &slotProber{serverURL: ts.URL, token: "unused", payloadSize: testProbePayloadSize} + + _, verdict := prober.probe(context.Background(), newProbeSlot()) + + if verdict != probeInconclusive { + t.Fatalf("verdict = %v, want probeInconclusive", verdict) + } +} + +func TestSlotProberInconclusiveOnDialError(t *testing.T) { + ts, _ := newProbeServer(t) + deadURL := ts.URL + ts.Close() // connection refused from now on + + prober := &slotProber{serverURL: deadURL, token: "unused", payloadSize: testProbePayloadSize} + + _, verdict := prober.probe(context.Background(), newProbeSlot()) + + if verdict != probeInconclusive { + t.Fatalf("verdict = %v, want probeInconclusive", verdict) + } +} diff --git a/transport/http2/slot.go b/transport/http2/slot.go index 882c8867..b115288c 100644 --- a/transport/http2/slot.go +++ b/transport/http2/slot.go @@ -35,6 +35,14 @@ type transportSlot struct { lastHeavy int // last observed heavy count, tracks heavy 0->1 transitions lowCycles int recoverCycles int + + // Probe state, touched only from the health loop goroutine. suspected + // marks a slot whose passive throughput stayed low long enough to + // deserve an active probe confirmation; probeLowCycles counts + // consecutive slow probes; lastProbeAt enforces the probe cooldown. + suspected bool + probeLowCycles int + lastProbeAt time.Time } // eligible reports whether the slot may host a new stream under the given diff --git a/transport/http2/stream.go b/transport/http2/stream.go index 80cecbfe..3ae4dcd1 100644 --- a/transport/http2/stream.go +++ b/transport/http2/stream.go @@ -11,6 +11,7 @@ import ( "time" sharedconfig "github.com/nange/easyss/v3/config" + "github.com/nange/easyss/v3/stats" "github.com/nange/easyss/v3/transport" ) @@ -26,6 +27,13 @@ type HTTP2Stream struct { done func() slot *transportSlot + // bootstrapSentAt is the moment the client finished flushing the + // encrypted bootstrap record (the request body). The server answers + // with the response headers before dialing the origin, so the time + // between this stamp and the response arrival is the pure client<->server + // path RTT (no origin time). Stored as UnixNano; 0 means never stamped. + bootstrapSentAt atomic.Int64 + mu sync.Mutex r io.ReadCloser respErr error @@ -133,6 +141,14 @@ func (s *HTTP2Stream) setRoundTripErr(err error) { s.rtErrMu.Unlock() } +// MarkBootstrapSent stamps the moment the bootstrap record was flushed to +// the transport. The response headers arrive roughly one path RTT later (the +// server answers before dialing the origin), which Read() records as the +// pure client<->server RTT sample. +func (s *HTTP2Stream) MarkBootstrapSent() { + s.bootstrapSentAt.Store(time.Now().UnixNano()) +} + func (s *HTTP2Stream) Read(p []byte) (int, error) { s.mu.Lock() if s.closed { @@ -150,6 +166,12 @@ func (s *HTTP2Stream) Read(p []byte) (int, error) { s.respErr = res.err } else { s.r = res.resp.Body + // Response headers arrived: with MarkBootstrapSent stamped + // (bootstrap record flushed) this is the pure path RTT — + // the server commits the response before dialing the origin. + if t0 := s.bootstrapSentAt.Load(); t0 > 0 { + stats.RecordRTT(time.Since(time.Unix(0, t0))) + } } // If Close() ran while we were blocked waiting for the response, the // response body just arrived but nobody will ever read or close it. diff --git a/transport/http2/stream_test.go b/transport/http2/stream_test.go index 72ed3454..ffbbd153 100644 --- a/transport/http2/stream_test.go +++ b/transport/http2/stream_test.go @@ -3,9 +3,13 @@ package http2 import ( "errors" "io" + "net/http" + "strings" "sync" "testing" "time" + + "github.com/nange/easyss/v3/stats" ) func newTestStream() (*HTTP2Stream, *io.PipeReader) { @@ -151,3 +155,65 @@ func TestHTTP2Stream_TrackWriteNilSlotNoOp(t *testing.T) { t.Fatal("nil slot must not mark heavy") } } + +// TestHTTP2Stream_RecordsPathRTTOnResponse verifies that a successful +// response arriving after MarkBootstrapSent feeds one pure path RTT sample +// (bootstrap record flushed -> response headers arrived), and that a stream +// never stamped stays silent. +func TestHTTP2Stream_RecordsPathRTTOnResponse(t *testing.T) { + // newTestStream exposes respCh as read-only, so keep a writable handle + // to inject the response. + newStream := func() (*HTTP2Stream, chan roundTripResult) { + s, pr := newTestStream() + _ = pr + respCh := make(chan roundTripResult, 1) + s.respCh = respCh + return s, respCh + } + + t.Run("stamped stream records the sample", func(t *testing.T) { + stats.ResetCounters() + s, respCh := newStream() + defer s.Close() + + s.MarkBootstrapSent() + time.Sleep(2 * time.Millisecond) + respCh <- roundTripResult{ + resp: &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(""))}, + err: nil, + } + + buf := make([]byte, 16) + if _, err := s.Read(buf); err != io.EOF { + t.Fatalf("Read = %v, want io.EOF", err) + } + + snap := stats.Collect() + if snap.RTTCount != 1 { + t.Fatalf("RTTCount = %d, want 1", snap.RTTCount) + } + if snap.AvgRTT() < time.Millisecond { + t.Fatalf("AvgRTT = %v, want >= 1ms", snap.AvgRTT()) + } + }) + + t.Run("unstamped stream records nothing", func(t *testing.T) { + stats.ResetCounters() + s, respCh := newStream() + defer s.Close() + + respCh <- roundTripResult{ + resp: &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(""))}, + err: nil, + } + + buf := make([]byte, 16) + if _, err := s.Read(buf); err != io.EOF { + t.Fatalf("Read = %v, want io.EOF", err) + } + + if got := stats.Collect().RTTCount; got != 0 { + t.Fatalf("RTTCount = %d, want 0 without MarkBootstrapSent", got) + } + }) +}