fix(subs/lane): 让「能连接但不能用」的死出口节点离开轮询 (dead-egress)

internal/lane/client.go

@@ -1117,12 +1117,25 @@ func (l *Lane) attemptModel(ctx context.Context, req Request, model, effort stri
 		// ledger only ever skips known-bad pairs — it is never used to prefer a
 		// "known good" one, which would pin every concurrent request onto the
 		// first success and burn that exit's allowance.
-		switch asUpstream(err).Code {
+		code := asUpstream(err).Code
+		switch code {
 		case CodeRegion:
 			ReportExit(plan.NodeID, base, "region")
 		case CodeQuota:
 			ReportExit(plan.NodeID, base, "limited")
-		case CodeTransport, CodeServer, CodeTimeout, CodeEmpty, CodeReset:
+		case CodeTransport, CodeTimeout, CodeReset:
+			// Connection-level failure: the egress tunnel itself is broken
+			// (dead upstream / reset / silent stall) though the node's local
+			// socks inbound still answers. Downgrade the node so it leaves
+			// rotation instead of hanging every request for ~90s.
+			ReportExit(plan.NodeID, base, "other")
+			if r := CurrentRotator(); r != nil && plan.NodeID != "" {
+				r.ReportDead(plan.NodeID)
+			}
+		case CodeServer, CodeEmpty:
+			// A 5xx or an empty-but-2xx body is not proof the egress tunnel is
+			// dead — the connection clearly worked — so it only books the
+			// per-(exit,model) ban and never downgrades node health.
 			ReportExit(plan.NodeID, base, "other")
 		}
 		// The caller decides what a 429 means: under rotation it re-asks the

internal/lane/http.go

@@ -135,6 +135,13 @@ type Rotator interface {
 	// Report books one failure against an (exit, model) pair. The ledger is
 	// only ever used to skip a combination known to be bad.
 	Report(nodeID, model, class string)
+	// ReportDead records that a real request through nodeID failed at the
+	// transport layer — the egress tunnel is broken though the node's local
+	// socks inbound still answers ("能连接但不能用"). Two consecutive hits
+	// downgrade the node to Alive=false so the rotation drops it and the
+	// dead-node recheck re-probes it, instead of letting every request hang
+	// for up to responseHeaderTimeout.
+	ReportDead(nodeID string)
 	// Revive clears the (exit, model) pair and the exit's transport cooldown
 	// after a real request for that model succeeded through it — real traffic
 	// is harder evidence than any probe sample. It never lifts bans recorded

internal/lane/rotate_test.go

@@ -51,6 +51,8 @@ func (r *countRotator) ExitKey(nodeID string) string { return nodeID }
 
 func (r *countRotator) Report(nodeID, model, class string) {}
 
+func (r *countRotator) ReportDead(nodeID string) {}
+
 func (r *countRotator) Revive(nodeID, model string) {}
 
 func (r *countRotator) seen() []string {

internal/subs/ledger.go

@@ -79,6 +79,47 @@ func (m *Manager) Report(nodeID, model, class string) {
 	m.bans[k] = now.Add(d)
 }
 
+// nodeDeadAfter is how many consecutive transport failures downgrade a node to
+// Alive=false. One is usually a transient stall (a cold-start first call has
+// been measured at 62s here), so two in a row is the bar for calling an egress
+// dead from live traffic.
+const nodeDeadAfter = 2
+
+// ReportDead records that a real request through nodeID failed at the transport
+// layer — the egress tunnel itself is broken (a dead upstream, a connection
+// reset, a silent stall) even though the node's local socks inbound still
+// answers. That is exactly the "能连接但不能用" case: the connect succeeds,
+// the request never returns. A probe is a sample and cannot catch it between
+// probe rounds, and the per-(exit,model) ban alone only parks the pair for
+// seconds and is keyed per model, so the node keeps being chosen for every
+// request and each attempt hangs for up to responseHeaderTimeout (90s).
+//
+// Two consecutive transport failures within failMemory downgrade the node to
+// Alive=false. That drops it from Pick/nextNode immediately AND makes it
+// eligible for the 2-minute dead-node recheck (ProbeDead only re-tests
+// !Alive nodes), so a genuinely dead egress leaves rotation within one or two
+// requests and a merely flaky one self-heals on the next probe instead of
+// lingering until the next full ProbeAll (~30 min).
+func (m *Manager) ReportDead(nodeID string) {
+	if nodeID == "" {
+		return
+	}
+	now := time.Now()
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if last, ok := m.nodeFailAt[nodeID]; ok && now.Sub(last) > failMemory {
+		m.nodeFails[nodeID] = 0
+	}
+	m.nodeFails[nodeID]++
+	m.nodeFailAt[nodeID] = now
+	if m.nodeFails[nodeID] >= nodeDeadAfter {
+		if h := m.health[nodeID]; h != nil {
+			h.Alive = false
+			h.CoolUntil = 0
+		}
+	}
+}
+
 // Revive clears an exit's failure state after a real request for one model
 // succeeded through it. A served request is harder evidence than any probe
 // sample, so there is no reason to sit out the rest of a cooldown.
@@ -89,6 +130,11 @@ func (m *Manager) Report(nodeID, model, class string) {
 // or has a separate quota for, so clearing sibling bans here would let a
 // model-B success re-arm a model-A wall. Only the pair that just succeeded is
 // lifted.
+//
+// A served request is also proof the egress is live, so Revive re-arms the
+// node's Alive flag and clears the transport-failure counter that ReportDead
+// may have accumulated — a node that answers one real request is not dead, even
+// if it had timed out twice just before.
 func (m *Manager) Revive(nodeID, model string) {
 	if nodeID == "" {
 		return
@@ -99,8 +145,11 @@ func (m *Manager) Revive(nodeID, model string) {
 	if h := m.health[nodeID]; h != nil {
 		h.CoolUntil = 0
 		h.LastOKMs = now.UnixMilli()
+		h.Alive = true
 		delete(m.nodeStreak, nodeID)
 	}
+	delete(m.nodeFails, nodeID)
+	delete(m.nodeFailAt, nodeID)
 	if model == "" {
 		return
 	}

internal/subs/ledger_test.go

@@ -133,3 +133,58 @@ func TestExpiredBanReleasesTheExit(t *testing.T) {
 		t.Error("the expired entry was not cleared")
 	}
 }
+
+// A dead egress ("能连接但不能用"): the node's socks inbound still answers, but
+// the upstream behind it never returns. One transport failure must not panic a
+// node out of rotation — cold-start stalls reach 62s here — but two consecutive
+// ones must, so the picker stops choosing it and the dead-node recheck re-probes
+// it instead of letting every request hang for ~90s.
+func TestReportDeadDowngradesNodeAfterTwoFailures(t *testing.T) {
+	m, health := poolForPick()
+	if !health["n3"].Alive {
+		t.Fatal("setup: n3 should start alive")
+	}
+	m.ReportDead("n3")
+	if !health["n3"].Alive {
+		t.Error("a single transport failure must not downgrade the node")
+	}
+	m.ReportDead("n3")
+	if health["n3"].Alive {
+		t.Fatal("two consecutive transport failures should downgrade the node to Alive=false")
+	}
+	// The downgraded node must leave the picker immediately.
+	if id := m.Pick("", "m"); id == "n3" {
+		t.Error("Pick returned a node downgraded for dead-egress")
+	}
+}
+
+// The consecutive-failure count restarts after a long quiet spell, so a node
+// that stalls once an hour is never treated as persistently dead.
+func TestReportDeadCountForgetsAfterMemoryWindow(t *testing.T) {
+	m, _ := poolForPick()
+	m.ReportDead("n3")
+	m.nodeFailAt["n3"] = time.Now().Add(-2 * failMemory)
+	m.ReportDead("n3")
+	if !m.health["n3"].Alive {
+		t.Error("the count should have restarted after the quiet spell, not downgraded the node")
+	}
+}
+
+// A served request is proof the egress is live: Revive must re-arm a node that
+// ReportDead had downgraded, and clear its transport-failure counter so a later
+// stall does not immediately re-kill it.
+func TestReviveReArmsDeadNode(t *testing.T) {
+	m, health := poolForPick()
+	m.ReportDead("n3")
+	m.ReportDead("n3")
+	if health["n3"].Alive {
+		t.Fatal("setup: n3 should be downgraded before Revive")
+	}
+	m.Revive("n3", "m")
+	if !health["n3"].Alive {
+		t.Error("Revive did not re-arm a dead node")
+	}
+	if _, still := m.nodeFails["n3"]; still {
+		t.Error("Revive did not clear the transport-failure counter")
+	}
+}

internal/subs/manager.go

@@ -61,6 +61,20 @@ type Manager struct {
 	// escalating cooldown in coolNode so a persistently dead node stops costing
 	// a full dialTimeout on every request. Keyed by node ID.
 	nodeStreak map[string]int
+	// nodeFails/nodeFailAt feed live transport failures back into node health.
+	// A subscription node whose egress died but whose local socks inbound
+	// still answers is the classic "能连接但不能用" case: the TCP/SOCKS
+	// connect succeeds, but the request never comes back. A probe sample cannot
+	// catch it between probe rounds, and the per-(exit,model) ban alone only
+	// parks the pair for seconds and is keyed per model, so the node keeps
+	// being chosen for every request and each attempt hangs for up to
+	// responseHeaderTimeout (90s). Two consecutive transport failures within
+	// failMemory downgrade the node to Alive=false, which drops it from
+	// Pick/nextNode immediately AND lets the 2-minute dead-node recheck
+	// re-probe it — a genuinely dead egress leaves rotation within one or two
+	// requests, and a merely flaky one self-heals on the next probe.
+	nodeFails  map[string]int
+	nodeFailAt map[string]time.Time
 	lastErr    string
 	version    string
 
@@ -94,10 +108,12 @@ func NewManager(st *store.Store, log func(string, ...any)) *Manager {
 		st: st, log: log,
 		health:   map[string]*NodeHealth{},
 		sticky:   map[string]string{},
-		bans:     map[string]time.Time{},
-		fails:    map[string]int{},
-		failAt:   map[string]time.Time{},
-		stickyAt: map[string]int64{},
+		bans:       map[string]time.Time{},
+		fails:      map[string]int{},
+		failAt:     map[string]time.Time{},
+		nodeFails:  map[string]int{},
+		nodeFailAt: map[string]time.Time{},
+		stickyAt:   map[string]int64{},
 	}
 }