feat: report per-connection status in transport stats

cmd/easyss/main.go

@@ -294,6 +294,7 @@ func (a *App) statsLoop() {
 				"conns", snap.Conns,
 				"priority_conns", snap.PriorityConns,
 				"bulk_conns", snap.BulkConns,
+				"conns_status", snap.ConnsStatus,
 				"active_streams", snap.ActiveStreams,
 				"priority_active", snap.PriorityActiveStreams,
 				"bulk_active", snap.BulkActiveStreams,

transport/http2/client.go

@@ -8,6 +8,9 @@ import (
 	"net"
 	"net/http"
 	"runtime"
+	"slices"
+	"strconv"
+	"strings"
 	"sync"
 	"time"
 
@@ -96,6 +99,7 @@ func New(cfg Config) (*HTTP2Transport, error) {
 	slots := make([]*transportSlot, maxSlots)
 	for i := range slots {
 		slots[i] = newSlot(cfg.TLSConfig, timeout, dialCtx, connLifetime)
+		slots[i].idx = i
 	}
 
 	sched := newScheduler(maxSlots, slots, threshold, prioritySlots)
@@ -304,6 +308,13 @@ func (t *HTTP2Transport) CloseIdle() {
 }
 
 func (t *HTTP2Transport) Stats() transport.TransportStats {
+	// Hold the scheduler read lock so the snapshot is consistent: shrink
+	// (swap-remove) and grow mutate liveCount and the live slot range under
+	// the write lock, so an unlocked render could read a stale liveCount
+	// and report more conns_status entries than Conns.
+	t.sched.mu.RLock()
+	defer t.sched.mu.RUnlock()
+
 	live := int(t.sched.liveCount.Load())
 	ts := transport.TransportStats{
 		Conns: live,
@@ -324,9 +335,77 @@ func (t *HTTP2Transport) Stats() transport.TransportStats {
 			ts.BulkActiveStreams += a
 		}
 	}
+	ts.ConnsStatus = slotStatusString(t.sched, live)
 	return ts
 }
 
+// slotStatus derives a live slot's connection status from its health flags.
+// Multiple flags are joined with "+" so no state is hidden (a heavy download
+// crossing the connection lifetime is both heavy and expiring); a slot with
+// no flags is "active".
+func slotStatus(s *transportSlot) string {
+	var parts []string
+	if s.heavy.Load() > 0 {
+		parts = append(parts, "heavy")
+	}
+	if s.degraded.Load() {
+		parts = append(parts, "degraded")
+	}
+	if s.expiring.Load() {
+		parts = append(parts, "expiring")
+	}
+	if len(parts) == 0 {
+		return "active"
+	}
+	return strings.Join(parts, "+")
+}
+
+// slotStatusString renders every live slot as "<index>:<active streams>:
+// <status>", wrapped in brackets, e.g. "[0:3:degraded, 1:2:expiring,
+// 2:1:active, 3:1:heavy]". Entries are ordered by the stable slot identity
+// (retire swap-removes scramble the live order) and then renumbered from 0,
+// so the rendered indices are always consecutive with no jumps. An empty
+// live set renders as "[]". live must be the liveCount value the caller
+// snapshot under the scheduler lock, so the rendered entry count always
+// matches Conns.
+func slotStatusString(sched *slotScheduler, live int) string {
+	if live == 0 {
+		return "[]"
+	}
+	type entry struct {
+		idx    int
+		active int
+		status string
+	}
+	entries := make([]entry, 0, live)
+	for i := 0; i < live; i++ {
+		s := sched.slots[i]
+		entries = append(entries, entry{
+			idx:    s.idx,
+			active: int(s.active.Load()),
+			status: slotStatus(s),
+		})
+	}
+	// Order by stable slot index regardless of the scrambled live order,
+	// then number entries 0..n-1 so the output indices never jump.
+	slices.SortFunc(entries, func(a, b entry) int { return a.idx - b.idx })
+
+	var b strings.Builder
+	b.WriteByte('[')
+	for i, e := range entries {
+		if i > 0 {
+			b.WriteString(", ")
+		}
+		b.WriteString(strconv.Itoa(i))
+		b.WriteByte(':')
+		b.WriteString(strconv.Itoa(e.active))
+		b.WriteByte(':')
+		b.WriteString(e.status)
+	}
+	b.WriteByte(']')
+	return b.String()
+}
+
 func (t *HTTP2Transport) Close() error {
 	t.cancel()
 	live := t.sched.liveCount.Load()

transport/http2/consistency_test.go

@@ -0,0 +1,102 @@
+package http2
+
+import (
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// TestStatsConnsStatusConsistencyUnderShrink stresses the scheduler with a
+// shrinker goroutine (simulating closeIdleLoop) while a stats reader
+// goroutine (simulating /stats polling) renders conns_status. It fails if the
+// rendered entry count ever exceeds the concurrently reported Conns value,
+// or if the rendered indices are not consecutive from 0.
+func TestStatsConnsStatusConsistencyUnderShrink(t *testing.T) {
+	slots := make([]*transportSlot, 6)
+	for i := range slots {
+		slots[i] = &transportSlot{idx: i}
+	}
+	sch := newScheduler(6, slots, 2, 1)
+	sch.liveCount.Store(6)
+
+	var over atomic.Int64
+	var badIdx atomic.Int64
+	var reads atomic.Int64
+	done := make(chan struct{})
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-done:
+				return
+			default:
+			}
+			sch.mu.Lock()
+			sch.shrinkIdleLocked()
+			sch.mu.Unlock()
+			// Re-grow like new streams arriving.
+			sch.grow(false)
+		}
+	}()
+
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-done:
+				return
+			default:
+			}
+			live := int(sch.liveCount.Load())
+			s := slotStatusString(sch, live)
+			idxs := parseIndices(s)
+			if len(idxs) > live {
+				over.Add(int64(len(idxs) - live))
+			}
+			for k, id := range idxs {
+				if id != k {
+					badIdx.Add(1)
+					break
+				}
+			}
+			reads.Add(1)
+		}
+	}()
+
+	time.Sleep(200 * time.Millisecond)
+	close(done)
+	wg.Wait()
+	t.Logf("reads=%d over_by_total=%d bad_idx_total=%d", reads.Load(), over.Load(), badIdx.Load())
+	if over.Load() > 0 {
+		t.Fatalf("conns_status rendered more entries than Conns: over=%d", over.Load())
+	}
+	if badIdx.Load() > 0 {
+		t.Fatalf("conns_status indices are not consecutive from 0: bad=%d", badIdx.Load())
+	}
+}
+
+// parseIndices extracts the leading "<index>:" of each entry, verifying the
+// array-like shape at the same time.
+func parseIndices(s string) []int {
+	if s == "[]" {
+		return nil
+	}
+	body := strings.TrimSuffix(strings.TrimPrefix(s, "["), "]")
+	parts := strings.Split(body, ", ")
+	idxs := make([]int, 0, len(parts))
+	for _, p := range parts {
+		colon := strings.IndexByte(p, ':')
+		n, err := strconv.Atoi(p[:colon])
+		if err != nil {
+			panic("malformed entry: " + p)
+		}
+		idxs = append(idxs, n)
+	}
+	return idxs
+}

transport/http2/scheduler_test.go

@@ -11,7 +11,7 @@ import (
 func newTestScheduler(specs ...[2]int32) *slotScheduler {
 	slots := make([]*transportSlot, len(specs))
 	for i, s := range specs {
-		slots[i] = &transportSlot{}
+		slots[i] = &transportSlot{idx: i}
 		slots[i].active.Store(s[0])
 		slots[i].heavy.Store(s[1])
 	}

transport/http2/slot.go

@@ -10,6 +10,11 @@ import (
 // to a single connection via MaxConnsPerHost=1) plus the state that the
 // scheduler and the connection lifecycle act on.
 type transportSlot struct {
+	// idx is the slot's stable index in the scheduler's pre-allocated
+	// array, set once at construction. Retire swap-removes slots, so the
+	// live position and idx diverge; idx is used for stats reporting.
+	idx int
+
 	t      *http.Transport
 	active atomic.Int32
 	heavy  atomic.Int32 // number of active heavy streams (>= HeavyStreamThreshold bytes)

transport/http2/slot_status_test.go

@@ -0,0 +1,94 @@
+package http2
+
+import (
+	"testing"
+)
+
+func TestSlotStatus(t *testing.T) {
+	tests := []struct {
+		name   string
+		heavy  int32
+		deg    bool
+		exp    bool
+		expect string
+	}{
+		{name: "no flags is active", expect: "active"},
+		{name: "heavy", heavy: 1, expect: "heavy"},
+		{name: "degraded", deg: true, expect: "degraded"},
+		{name: "expiring", exp: true, expect: "expiring"},
+		{name: "heavy and expiring", heavy: 1, exp: true, expect: "heavy+expiring"},
+		{name: "degraded and heavy", heavy: 2, deg: true, expect: "heavy+degraded"},
+		{name: "all flags", heavy: 1, deg: true, exp: true, expect: "heavy+degraded+expiring"},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			s := &transportSlot{}
+			s.heavy.Store(tt.heavy)
+			s.degraded.Store(tt.deg)
+			s.expiring.Store(tt.exp)
+			if got := slotStatus(s); got != tt.expect {
+				t.Fatalf("slotStatus() = %q, want %q", got, tt.expect)
+			}
+		})
+	}
+}
+
+func TestSlotStatusString(t *testing.T) {
+	t.Run("empty live set", func(t *testing.T) {
+		sch := newTestScheduler()
+		if got := slotStatusString(sch, int(sch.liveCount.Load())); got != "[]" {
+			t.Fatalf("slotStatusString() = %q, want %q", got, "[]")
+		}
+	})
+
+	t.Run("all active with stream counts", func(t *testing.T) {
+		sch := newTestScheduler([2]int32{1, 0}, [2]int32{2, 0})
+		if got, want := slotStatusString(sch, int(sch.liveCount.Load())), "[0:1:active, 1:2:active]"; got != want {
+			t.Fatalf("slotStatusString() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("mixed states", func(t *testing.T) {
+		sch := newTestScheduler([2]int32{3, 0}, [2]int32{2, 0}, [2]int32{1, 0}, [2]int32{1, 1})
+		sch.slots[0].degraded.Store(true)
+		sch.slots[1].expiring.Store(true)
+		want := "[0:3:degraded, 1:2:expiring, 2:1:active, 3:1:heavy]"
+		if got := slotStatusString(sch, int(sch.liveCount.Load())); got != want {
+			t.Fatalf("slotStatusString() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("combined flags", func(t *testing.T) {
+		sch := newTestScheduler([2]int32{1, 1}, [2]int32{0, 0})
+		sch.slots[0].expiring.Store(true)
+		sch.slots[1].degraded.Store(true)
+		want := "[0:1:heavy+expiring, 1:0:degraded]"
+		if got := slotStatusString(sch, int(sch.liveCount.Load())); got != want {
+			t.Fatalf("slotStatusString() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("respects liveCount bound", func(t *testing.T) {
+		sch := newTestScheduler([2]int32{1, 0}, [2]int32{2, 0}, [2]int32{3, 0})
+		sch.slots[2].degraded.Store(true)
+		sch.liveCount.Store(2)
+		want := "[0:1:active, 1:2:active]"
+		if got := slotStatusString(sch, int(sch.liveCount.Load())); got != want {
+			t.Fatalf("slotStatusString() = %q, want %q", got, want)
+		}
+	})
+
+	t.Run("scrambled live order is renumbered consecutively", func(t *testing.T) {
+		// Simulate a swap-remove of slot 1: positions [0,1,2] now hold
+		// slots with real indices 0,3,2 and liveCount dropped to 3. Entries
+		// stay ordered by real index but are renumbered 0..n-1, so the
+		// output indices never jump.
+		sch := newTestScheduler([2]int32{1, 0}, [2]int32{2, 0}, [2]int32{3, 0}, [2]int32{4, 0})
+		sch.slots[1], sch.slots[3] = sch.slots[3], sch.slots[1]
+		sch.liveCount.Store(3)
+		want := "[0:1:active, 1:3:active, 2:4:active]"
+		if got := slotStatusString(sch, int(sch.liveCount.Load())); got != want {
+			t.Fatalf("slotStatusString() = %q, want %q", got, want)
+		}
+	})
+}

transport/transport.go

@@ -51,6 +51,12 @@ type TransportStats struct {
 	BulkActiveStreams     int `json:"bulk_active_streams"`
 	PriorityConns         int `json:"priority_conns"`
 	BulkConns             int `json:"bulk_conns"`
+	// ConnsStatus is a compact per-connection status summary, e.g.
+	// "[0:3:degraded, 1:2:expiring, 2:1:active, 3:1:heavy]". Each element is
+	// "<index>:<active streams>:<status>" where indices are consecutive from
+	// 0 (ordered by stable connection identity) and status is one of
+	// active/heavy/degraded/expiring, with multiple flags joined by "+".
+	ConnsStatus string `json:"conns_status,omitempty"`
 }
 
 type Transport interface {