feat(autoselect): tag user-traffic failures by kind

adapter/autoselect_history.go

@@ -1,10 +1,63 @@
 package adapter
 
 import (
+	"encoding/json"
 	"sync"
 	"time"
 )
 
+// UserFailureKind distinguishes how a member's user traffic failed.
+type UserFailureKind string
+
+const (
+	// UserFailureDial is a failure to establish the connection to the member.
+	UserFailureDial UserFailureKind = "dial"
+	// UserFailureStall is a data-plane stall on an already-dialed connection:
+	// the tunnel carried traffic, then went silent for the idle window.
+	UserFailureStall UserFailureKind = "stall"
+	// UserFailureReset is a non-benign transport error (e.g. RST) on an
+	// already-established connection — the tunnel broke mid-stream rather
+	// than going quiet.
+	UserFailureReset   UserFailureKind = "reset"
+	UserFailureUnknown UserFailureKind = "unknown"
+)
+
+// UserFailure stores when a user-traffic failure happened and what kind it was.
+type UserFailure struct {
+	At   time.Time       `json:"at"`
+	Kind UserFailureKind `json:"kind,omitempty"`
+}
+
+func normalizeUserFailureKind(kind UserFailureKind) UserFailureKind {
+	switch kind {
+	case UserFailureDial, UserFailureStall, UserFailureReset:
+		return kind
+	default:
+		return UserFailureUnknown
+	}
+}
+
+// UnmarshalJSON accepts both the current object form and the legacy bare
+// timestamp form. Legacy entries are tagged as unknown.
+func (u *UserFailure) UnmarshalJSON(data []byte) error {
+	var at time.Time
+	if err := json.Unmarshal(data, &at); err == nil {
+		u.At = at
+		u.Kind = UserFailureUnknown
+		return nil
+	}
+
+	type rawUserFailure UserFailure
+	var raw rawUserFailure
+	if err := json.Unmarshal(data, &raw); err != nil {
+		return err
+	}
+
+	*u = UserFailure(raw)
+	u.Kind = normalizeUserFailureKind(u.Kind)
+	return nil
+}
+
 // TagHistory is a point-in-time snapshot of one member's selection
 // state, mirroring the in-memory history kept by the MutableAutoSelect
 // group. Probe outcomes feed the scalar fields; real user-traffic
@@ -13,22 +66,19 @@ import (
 // lets the probe URL through while dropping user traffic shows up as
 // elevated UserFailures with healthy probe scalars simultaneously.
 type TagHistory struct {
-	// LastSuccessDelayMs is the most recent successful probe RTT
-	// (clamped to ≥1 ms). 0 means "no successful probe yet" — rank
-	// uses it as a sentinel to deprioritize against measured peers.
+	// LastSuccessDelayMs is the most recent successful probe RTT.
+	// 0 means "no successful probe yet".
 	LastSuccessDelayMs uint32 `json:"last_success_delay_ms,omitempty"`
-	// LastOutcomeAt is the timestamp of the most recent probe outcome
-	// (success or failure). Used by the probe-cycle freshSince gate to
-	// distinguish "probed this cycle" from "stale outcome."
+	// LastOutcomeAt is the timestamp of the most recent probe outcome.
 	LastOutcomeAt time.Time `json:"last_outcome_at,omitempty"`
 	// ConsecutiveFailures counts probe failures since the last probe
 	// success. Resets to zero on probe success.
 	ConsecutiveFailures uint32 `json:"consecutive_failures,omitempty"`
-	// UserFailures is the sliding window of user-traffic failure
-	// timestamps (dial errors and data-plane stalls). Probe successes
-	// never enter this window; entries age out naturally on the next
-	// mutation older than the group's userFailureWindow.
-	UserFailures []time.Time `json:"user_failures,omitempty"`
+	// UserFailures is the sliding window of user-traffic failures, each
+	// tagged with its kind. Probe successes never enter this window; entries
+	// age out naturally on the next mutation older than the group's
+	// userFailureWindow.
+	UserFailures []UserFailure `json:"user_failures,omitempty"`
 	// HardDemoted reports whether the member has reached the hard-demote
 	// tier from its own recorded probe and user-traffic failures.
 	HardDemoted bool      `json:"hard_demoted,omitempty"`
@@ -164,7 +214,7 @@ func cloneTagHistory(h *TagHistory) *TagHistory {
 	}
 	out := *h
 	if len(h.UserFailures) > 0 {
-		out.UserFailures = make([]time.Time, len(h.UserFailures))
+		out.UserFailures = make([]UserFailure, len(h.UserFailures))
 		copy(out.UserFailures, h.UserFailures)
 	}
 	return &out

adapter/autoselect_history_test.go

@@ -0,0 +1,47 @@
+package adapter
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestUserFailure_UnmarshalJSON_ObjectForm(t *testing.T) {
+	at := time.Date(2026, 7, 6, 12, 0, 0, 0, time.UTC)
+	data, err := json.Marshal(UserFailure{At: at, Kind: UserFailureStall})
+	require.NoError(t, err)
+
+	var got UserFailure
+	require.NoError(t, json.Unmarshal(data, &got))
+	assert.True(t, got.At.Equal(at))
+	assert.Equal(t, UserFailureStall, got.Kind)
+}
+
+func TestUserFailure_UnmarshalJSON_LegacyTimestamp(t *testing.T) {
+	// Snapshots persisted before the kind was tracked stored each failure
+	// as a bare RFC3339 timestamp string. Those must still load, tagged as
+	// unknown, so an in-place upgrade doesn't fail to read history.
+	at := time.Date(2026, 7, 6, 12, 0, 0, 0, time.UTC)
+	legacy, err := json.Marshal(at)
+	require.NoError(t, err)
+
+	var got UserFailure
+	require.NoError(t, json.Unmarshal(legacy, &got))
+	assert.True(t, got.At.Equal(at))
+	assert.Equal(t, UserFailureUnknown, got.Kind)
+}
+
+func TestTagHistory_UnmarshalJSON_LegacyUserFailures(t *testing.T) {
+	// A whole TagHistory persisted with the legacy []time.Time window must
+	// decode without error.
+	legacy := `{"user_failures":["2026-07-06T12:00:00Z","2026-07-06T12:01:00Z"]}`
+	var got TagHistory
+	require.NoError(t, json.Unmarshal([]byte(legacy), &got))
+	require.Len(t, got.UserFailures, 2)
+	for _, f := range got.UserFailures {
+		assert.Equal(t, UserFailureUnknown, f.Kind)
+	}
+}

protocol/group/mutableautoselect.go

@@ -469,7 +469,7 @@ func (s *MutableAutoSelect) DialContext(ctx context.Context, network string, des
 	// runLadder's gating both see it. For nested groups, the inner tag
 	// isn't a member of this group, so recordUserFailure would drop on
 	// the member gate.
-	s.recordUserFailure(outerTag)
+	s.recordUserFailure(outerTag, adapter.UserFailureDial)
 
 	// Fast failover: try one alternate from the current rank before
 	// kicking the ladder. The active 3-min cadence keeps the probe
@@ -483,7 +483,7 @@ func (s *MutableAutoSelect) DialContext(ctx context.Context, network string, des
 			return s.wrapStream(conn, alt), nil
 		}
 		s.logger.ErrorContext(ctx, err)
-		s.recordUserFailure(altTag)
+		s.recordUserFailure(altTag, adapter.UserFailureDial)
 		outerTag = altTag
 	}
 
@@ -512,7 +512,7 @@ func (s *MutableAutoSelect) ListenPacket(ctx context.Context, destination M.Sock
 		return s.wrapPacket(conn, o), nil
 	}
 	s.logger.ErrorContext(ctx, err)
-	s.recordUserFailure(outerTag)
+	s.recordUserFailure(outerTag, adapter.UserFailureDial)
 
 	alt, altErr := s.selectForExcluding("udp", outerTag)
 	if altErr == nil {
@@ -523,7 +523,7 @@ func (s *MutableAutoSelect) ListenPacket(ctx context.Context, destination M.Sock
 			return s.wrapPacket(conn, alt), nil
 		}
 		s.logger.ErrorContext(ctx, err)
-		s.recordUserFailure(altTag)
+		s.recordUserFailure(altTag, adapter.UserFailureDial)
 		outerTag = altTag
 	}
 
@@ -957,17 +957,17 @@ func (s *MutableAutoSelect) mutateHistory(tag string, fn func(*localHistory, tim
 	return true
 }
 
-// recordUserFailure appends a single user-traffic failure timestamp to
-// the member's sliding window and persists the snapshot. Dial errors
-// and confirmed data-plane stalls both pass through here; each counts
+// recordUserFailure appends a single user-traffic failure of the given
+// kind to the member's sliding window and persists the snapshot. Dial
+// errors and confirmed data-plane stalls both pass through here; each counts
 // as one failure, with softFailLimit tripping the soft tier and
 // consecutiveFailLimit tripping the hard tier. Returns true if the
 // failure was persisted; false on dedup or non-member tag, so callers
 // can suppress downstream side effects (e.g. ladder kicks) on collapsed
 // events.
-func (s *MutableAutoSelect) recordUserFailure(tag string) bool {
+func (s *MutableAutoSelect) recordUserFailure(tag string, kind adapter.UserFailureKind) bool {
 	return s.mutateHistory(tag, func(h *localHistory, now time.Time) bool {
-		return h.addUserFailure(now, s.hist.userFailureWindow, s.hist.userFailureDedupeWindow)
+		return h.addUserFailure(adapter.UserFailure{At: now, Kind: kind}, s.hist.userFailureWindow, s.hist.userFailureDedupeWindow)
 	})
 }
 

protocol/group/mutableautoselect_dataplane.go

@@ -8,23 +8,25 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/getlantern/lantern-box/adapter"
 )
 
-// makeHooks returns the (onStall, onActivity) callback pair wired into
+// makeHooks returns the (onFailure, onActivity) callback pair wired into
 // the data-plane wrappers. Callbacks re-look-up the member's history
 // at fire time so a Remove+Add cycle can't strand writes on a stale
-// entry. The stall callback short-circuits when recordUserFailure
+// entry. The failure callback short-circuits when recordUserFailure
 // reports a dedup/non-member, so a single broken outbound with N
 // orphan idle conns doesn't trigger N redundant full-fleet probe
 // sweeps.
-func (s *MutableAutoSelect) makeHooks(outerTag string) (onStall, onActivity func()) {
-	onStall = func() {
-		if !s.recordUserFailure(outerTag) {
+func (s *MutableAutoSelect) makeHooks(outerTag string) (onFailure func(adapter.UserFailureKind), onActivity func()) {
+	onFailure = func(kind adapter.UserFailureKind) {
+		if !s.recordUserFailure(outerTag, kind) {
 			return
 		}
 		go s.runLadder(outerTag)
 	}
-	return onStall, s.bumpActive
+	return onFailure, s.bumpActive
 }
 
 // dataPlaneWatchdog is the no-traffic stall timer shared by the stream
@@ -45,7 +47,7 @@ func (s *MutableAutoSelect) makeHooks(outerTag string) (onStall, onActivity func
 // identical to a broken tunnel without this gate.
 type dataPlaneWatchdog struct {
 	idle            time.Duration
-	onStall         func()
+	onFailure       func(adapter.UserFailureKind)
 	onActivity      func()
 	provedReadBytes uint64
 	readBytes       atomic.Uint64
@@ -57,10 +59,10 @@ type dataPlaneWatchdog struct {
 	closeOnce       sync.Once
 }
 
-func (w *dataPlaneWatchdog) init(idle time.Duration, provedReadBytes uint64, onStall, onActivity func()) {
+func (w *dataPlaneWatchdog) init(idle time.Duration, provedReadBytes uint64, onFailure func(adapter.UserFailureKind), onActivity func()) {
 	w.idle = idle
 	w.provedReadBytes = provedReadBytes
-	w.onStall = onStall
+	w.onFailure = onFailure
 	w.onActivity = onActivity
 	w.timer = time.AfterFunc(idle, w.fireStall)
 }
@@ -87,7 +89,7 @@ func (w *dataPlaneWatchdog) noteIO(n int, err error, isRead bool) {
 	}
 	// Attribute mid-stream transport failures immediately.
 	if isDataPlaneFailure(err) {
-		w.fireFailure()
+		w.fireResetFailure()
 		return
 	}
 	// Ignore empty I/O and benign terminal conditions.
@@ -122,7 +124,7 @@ func (w *dataPlaneWatchdog) noteIO(n int, err error, isRead bool) {
 
 // closeWatchdog sets stalled=true before Stop so any concurrent noteIO
 // short-circuits and any concurrent fireStall CAS-fails — late I/O can't
-// deliver a phantom onStall after Close.
+// deliver a phantom onFailure after Close.
 func (w *dataPlaneWatchdog) closeWatchdog() (firstClose bool) {
 	w.closeOnce.Do(func() {
 		w.stalled.Store(true)
@@ -148,25 +150,25 @@ func (w *dataPlaneWatchdog) fireStall() {
 	if !w.fired.CompareAndSwap(false, true) {
 		return
 	}
-	if w.onStall != nil {
-		w.onStall()
+	if w.onFailure != nil {
+		w.onFailure(adapter.UserFailureStall)
 	}
 }
 
-// fireFailure demotes on a mid-stream transport error. Unlike fireStall it skips
+// fireResetFailure demotes on a mid-stream transport error. Unlike fireStall it skips
 // the proven/lastWasWrite gates — an explicit reset is unambiguous even before a
 // conn is proven. Fire and attribution happens at most once.
-func (w *dataPlaneWatchdog) fireFailure() {
+func (w *dataPlaneWatchdog) fireResetFailure() {
 	if !w.stalled.CompareAndSwap(false, true) {
 		return
 	}
 	if !w.fired.CompareAndSwap(false, true) {
 		return
 	}
-	if w.onStall != nil {
-		// run onStall in a goroutine to avoid deadlocks: the callback may call back
+	if w.onFailure != nil {
+		// run onFailure in a goroutine to avoid deadlocks: the callback may call back
 		// into the selector, which may hold locks that the data-plane IO path also needs.
-		go w.onStall()
+		go w.onFailure(adapter.UserFailureReset)
 	}
 }
 
@@ -177,9 +179,9 @@ type dataPlaneStream struct {
 	dataPlaneWatchdog
 }
 
-func newDataPlaneStream(c net.Conn, idle time.Duration, provedReadBytes uint64, onStall, onActivity func()) *dataPlaneStream {
+func newDataPlaneStream(c net.Conn, idle time.Duration, provedReadBytes uint64, onFailure func(adapter.UserFailureKind), onActivity func()) *dataPlaneStream {
 	d := &dataPlaneStream{Conn: c}
-	d.init(idle, provedReadBytes, onStall, onActivity)
+	d.init(idle, provedReadBytes, onFailure, onActivity)
 	return d
 }
 
@@ -215,9 +217,15 @@ type dataPlanePacket struct {
 	dataPlaneWatchdog
 }
 
-func newDataPlanePacket(c net.PacketConn, idle time.Duration, provedReadBytes uint64, onStall, onActivity func()) *dataPlanePacket {
+func newDataPlanePacket(
+	c net.PacketConn,
+	idle time.Duration,
+	provedReadBytes uint64,
+	onFailure func(adapter.UserFailureKind),
+	onActivity func(),
+) *dataPlanePacket {
 	d := &dataPlanePacket{PacketConn: c}
-	d.init(idle, provedReadBytes, onStall, onActivity)
+	d.init(idle, provedReadBytes, onFailure, onActivity)
 	return d
 }