diff --git a/internal/lane/client.go b/internal/lane/client.go index 129bbe2..5872ff5 100644 --- a/internal/lane/client.go +++ b/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 diff --git a/internal/lane/http.go b/internal/lane/http.go index 96c81d0..4086993 100644 --- a/internal/lane/http.go +++ b/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 diff --git a/internal/lane/rotate_test.go b/internal/lane/rotate_test.go index 9d8f84d..6bcb3c9 100644 --- a/internal/lane/rotate_test.go +++ b/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 { diff --git a/internal/subs/ledger.go b/internal/subs/ledger.go index 80917c7..6dee989 100644 --- a/internal/subs/ledger.go +++ b/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 } diff --git a/internal/subs/ledger_test.go b/internal/subs/ledger_test.go index 829544c..5edd8e8 100644 --- a/internal/subs/ledger_test.go +++ b/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") + } +} diff --git a/internal/subs/manager.go b/internal/subs/manager.go index 9d34290..88ea522 100644 --- a/internal/subs/manager.go +++ b/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{}, } }