feat(stats): give the collector an outcome ledger

stats/collector.go

@@ -3,6 +3,7 @@ package stats
 import (
 	"cmp"
 	"fmt"
+	"log"
 	"slices"
 	"sort"
 	"sync"
@@ -13,6 +14,10 @@ import (
 type Collector struct {
 	mu sync.RWMutex
 
+	// loggedBadOutcome makes RecordOutcome's complaint fire once per run. A
+	// systematic classification bug puts every transaction through that branch.
+	loggedBadOutcome bool
+
 	// Transaction counts by scenario
 	txCounts map[string]uint64
 
@@ -88,6 +93,42 @@ func (c *Collector) RecordTransaction(scenario, operation string, latency time.D
 	c.recordWindowStats(latency)
 }
 
+// RecordOutcome counts one terminal outcome under the key the send path already
+// labels its metrics with. Call it once per transaction, for that transaction's
+// terminal state only. It adds rather than overwrites, because many
+// transactions share one key.
+//
+// Call it outside the tracker's registry lock. This method takes the collector
+// mutex, and the sender blocks on the registry lock at every send completion, so
+// a call made under both puts collector contention into the latency this package
+// reports. It is safe for concurrent callers; the nesting is what is not.
+//
+// An unset or out-of-range outcome counts as Unrecorded rather than vanishing,
+// so the conservation identity stays closed and the defect stays visible. The
+// first one also logs. A run never fails over a counting bug, and a systematic
+// one would otherwise write a line per transaction.
+func (c *Collector) RecordOutcome(key OperationKey, outcome Outcome) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	if outcome == outcomeUnset || outcome >= outcomeCount {
+		if !c.loggedBadOutcome {
+			c.loggedBadOutcome = true
+			log.Printf("stats: outcome %d for %s/%s is not a terminal state; "+
+				"counting it as %s. This is a bug in sei-load.",
+				outcome, key.Scenario, key.Operation, outcomeNames[outcomeUnset])
+		}
+		outcome = outcomeUnset
+	}
+
+	samples := c.perOperation[key]
+	if samples == nil {
+		samples = &operationSamples{}
+		c.perOperation[key] = samples
+	}
+	samples.outcomes[outcome]++
+}
+
 // recordOperation counts one attempt for key and, on success, adds its latency
 // to that operation's samples. The bound is the same one recordLatency applies
 // to the pooled window.
@@ -269,10 +310,17 @@ func (c *Collector) GetOperationStats() map[OperationKey]OperationStats {
 	out := make(map[OperationKey]OperationStats, len(c.perOperation))
 	for key, samples := range c.perOperation {
 		op := OperationStats{
-			Count:       samples.count,
-			Successes:   samples.successes,
-			SampleCount: len(samples.samples),
-			Window:      samples.window(),
+			Count:             samples.count,
+			Successes:         samples.successes,
+			SampleCount:       len(samples.samples),
+			Window:            samples.window(),
+			Committed:         samples.outcomes[OutcomeCommitted],
+			Failed:            samples.outcomes[OutcomeFailed],
+			Expired:           samples.outcomes[OutcomeExpired],
+			DroppedAtCap:      samples.outcomes[OutcomeDroppedAtCap],
+			DroppedAtHandoff:  samples.outcomes[OutcomeDroppedAtHandoff],
+			StatusUnavailable: samples.outcomes[OutcomeStatusUnavailable],
+			Unrecorded:        samples.outcomes[outcomeUnset],
 		}
 		if len(samples.samples) > 0 {
 			sorted := make([]time.Duration, len(samples.samples))
@@ -356,6 +404,28 @@ type OperationStats struct {
 	P99Latency  time.Duration
 	SampleCount int
 	Window      time.Duration
+
+	// Three layers, not one. Committed and Failed are what the chain did.
+	// Expired, DroppedAtCap and DroppedAtHandoff are what this run did to its
+	// own transactions, and none of the three is evidence about the chain.
+	// StatusUnavailable is what the run failed to see. Count and Successes above
+	// are the send path. Total them across layers and a generator throttling
+	// itself looks like a chain rejecting work.
+	//
+	// Failed here is the execution status a receipt reported. RunSummary.Failed
+	// is a send that returned an error. Opposite layers, same word.
+	//
+	// Unrecorded has no legitimate producer. A non-zero value means sei-load
+	// classified a transaction wrongly, and nothing else produces one.
+	//
+	// All of them stay zero until a tracker reports outcomes.
+	Committed         uint64
+	Failed            uint64
+	Expired           uint64
+	DroppedAtCap      uint64
+	DroppedAtHandoff  uint64
+	StatusUnavailable uint64
+	Unrecorded        uint64
 }
 
 // operationSamples accumulates one operation's counts and its most recent
@@ -369,6 +439,11 @@ type operationSamples struct {
 	count     uint64
 	successes uint64
 	samples   []latencySample
+
+	// results counts terminal states by Result. An array rather than a map:
+	// the set is closed, and indexing by the constant costs no allocation on a
+	// path the tracker walks once per transaction.
+	outcomes [outcomeCount]uint64
 }
 
 // latencySample is one successful transaction's latency and when it happened.

stats/collector_outcome_test.go

@@ -0,0 +1,162 @@
+package stats_test
+
+import (
+	"sync"
+	"testing"
+
+	"github.com/sei-protocol/sei-load/stats"
+	"github.com/stretchr/testify/require"
+)
+
+// TestOutcomesAccumulatePerKey is the property the ledger exists for: the
+// collector holds an outcome under the same key the send path already labels its
+// metrics with, so one operation's outcomes never land in another's count.
+//
+// It fails when two operations in one scenario share a count, and when a second
+// call overwrites the first instead of adding to it.
+func TestOutcomesAccumulatePerKey(t *testing.T) {
+	c := stats.NewCollector()
+	read := stats.OperationKey{Scenario: "storagerw", Operation: "read"}
+	rmw := stats.OperationKey{Scenario: "storagerw", Operation: "rmw"}
+
+	for i := 0; i < 3; i++ {
+		c.RecordOutcome(read, stats.OutcomeCommitted)
+	}
+	c.RecordOutcome(read, stats.OutcomeFailed)
+	c.RecordOutcome(rmw, stats.OutcomeCommitted)
+
+	got := c.GetOperationStats()
+	require.Equal(t, uint64(3), got[read].Committed, "repeat calls must add, not overwrite")
+	require.Equal(t, uint64(1), got[read].Failed)
+	require.Equal(t, uint64(1), got[rmw].Committed, "one operation's outcomes reached another's count")
+	require.Zero(t, got[rmw].Failed)
+}
+
+// outcomeReaders pairs each state with the count it must reach. Keeping them in
+// one table is what lets the tests below assert that a state reaches its own
+// count and no other.
+var outcomeReaders = []struct {
+	name    string
+	outcome stats.Outcome
+	read    func(stats.OperationStats) uint64
+}{
+	{"committed", stats.OutcomeCommitted, func(s stats.OperationStats) uint64 { return s.Committed }},
+	{"failed", stats.OutcomeFailed, func(s stats.OperationStats) uint64 { return s.Failed }},
+	{"expired", stats.OutcomeExpired, func(s stats.OperationStats) uint64 { return s.Expired }},
+	{"dropped_at_cap", stats.OutcomeDroppedAtCap, func(s stats.OperationStats) uint64 { return s.DroppedAtCap }},
+	{"dropped_at_handoff", stats.OutcomeDroppedAtHandoff, func(s stats.OperationStats) uint64 { return s.DroppedAtHandoff }},
+	{"status_unavailable", stats.OutcomeStatusUnavailable, func(s stats.OperationStats) uint64 { return s.StatusUnavailable }},
+}
+
+// TestEveryOutcomeReachesItsOwnCount covers the states individually, because a
+// switch that maps two of them to one field passes any test exercising only one.
+// The two drop states are one easily-conflated pair. Expired and
+// StatusUnavailable are the other: one says the chain did not take it, the other
+// says the run did not see.
+func TestEveryOutcomeReachesItsOwnCount(t *testing.T) {
+	key := stats.OperationKey{Scenario: "erc20", Operation: "erc20_transfer"}
+	for _, tc := range outcomeReaders {
+		t.Run(tc.name, func(t *testing.T) {
+			c := stats.NewCollector()
+			c.RecordOutcome(key, tc.outcome)
+			got := c.GetOperationStats()[key]
+			require.Equal(t, uint64(1), tc.read(got), "%s did not reach its own count", tc.name)
+
+			var total uint64
+			for _, other := range outcomeReaders {
+				total += other.read(got)
+			}
+			total += got.Unrecorded
+			require.Equal(t, uint64(1), total, "%s also incremented another count", tc.name)
+		})
+	}
+}
+
+// TestTheZeroValueIsNotASuccess is the reason a sentinel sits at index 0. A
+// transaction nobody classified — a switch that matched no case, an early return
+// — must not read as a commit, which is the failure this type exists to remove.
+func TestTheZeroValueIsNotASuccess(t *testing.T) {
+	var unclassified stats.Outcome // whatever a caller forgot to set
+
+	c := stats.NewCollector()
+	key := stats.OperationKey{Scenario: "erc20", Operation: "erc20_transfer"}
+	c.RecordOutcome(key, unclassified)
+
+	got := c.GetOperationStats()[key]
+	require.Zero(t, got.Committed, "an unclassified transaction counted as committed")
+	require.Equal(t, uint64(1), got.Unrecorded, "it must land somewhere visible")
+	require.Equal(t, "unrecorded", unclassified.String())
+}
+
+// TestAnUnknownOutcomeStaysVisible covers the other way a caller can be wrong. A
+// value past the known set must not vanish: the conservation identity would then
+// come up short with nothing to point at.
+func TestAnUnknownOutcomeStaysVisible(t *testing.T) {
+	c := stats.NewCollector()
+	key := stats.OperationKey{Scenario: "erc20", Operation: "erc20_transfer"}
+	c.RecordOutcome(key, stats.Outcome(200))
+	c.RecordOutcome(key, stats.Outcome(200))
+
+	all := c.GetOperationStats()
+	got, present := all[key]
+	require.True(t, present, "the operation key vanished along with the count")
+	require.Equal(t, uint64(2), got.Unrecorded)
+	require.Equal(t, "unrecorded", stats.Outcome(200).String())
+}
+
+// TestOutcomeNamesAreStable pins the strings, because a dashboard query and a
+// saved report both carry them. A rename orphans every one, so it has to be a
+// deliberate edit here rather than a side effect elsewhere.
+func TestOutcomeNamesAreStable(t *testing.T) {
+	require.Equal(t, "committed", stats.OutcomeCommitted.String())
+	require.Equal(t, "failed", stats.OutcomeFailed.String())
+	require.Equal(t, "expired", stats.OutcomeExpired.String())
+	require.Equal(t, "dropped_at_cap", stats.OutcomeDroppedAtCap.String())
+	require.Equal(t, "dropped_at_handoff", stats.OutcomeDroppedAtHandoff.String())
+	require.Equal(t, "status_unavailable", stats.OutcomeStatusUnavailable.String())
+}
+
+// TestConcurrentRecordingLosesNoCount fails when concurrent callers lose an
+// increment. It runs RecordOutcome against RecordTransaction and
+// GetOperationStats, because those are the pairs that can actually race: they
+// share one lock and touch the same map.
+//
+// It does not cover the tracker staying off the registry lock. That is TOT-010,
+// and no test here checks it.
+func TestConcurrentRecordingLosesNoCount(t *testing.T) {
+	c := stats.NewCollector()
+	key := stats.OperationKey{Scenario: "storagerw", Operation: "read"}
+
+	var wg sync.WaitGroup
+	for w := 0; w < 8; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < 250; i++ {
+				c.RecordOutcome(key, stats.OutcomeCommitted)
+			}
+		}()
+	}
+	for w := 0; w < 4; w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for i := 0; i < 500; i++ {
+				c.RecordTransaction(key.Scenario, key.Operation, 0, true)
+			}
+		}()
+	}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for i := 0; i < 200; i++ {
+			_ = c.GetOperationStats()
+		}
+	}()
+	wg.Wait()
+
+	got := c.GetOperationStats()[key]
+	require.Equal(t, uint64(2000), got.Committed)
+	require.Equal(t, uint64(2000), got.Count)
+	require.Zero(t, got.Unrecorded)
+}

stats/doc.go

@@ -0,0 +1,61 @@
+// Package stats holds what a run measured: what it submitted, what became of
+// those submissions, and the report that states both.
+//
+// # Types
+//
+//   - Collector — the run's ledger. Counts submissions per scenario and per
+//     operation, keeps recent latencies, and holds the terminal outcome of each
+//     transaction once a tracker reports one.
+//   - OperationKey, OperationStats — the per-operation dimension. The key is
+//     what the send path already labels its metrics with; the stats are what a
+//     report reads.
+//   - Outcome — the six terminal states a transaction reaches, and the sentinel
+//     that catches one nobody classified.
+//   - InclusionTracker — the registry of transactions sent and not yet
+//     accounted for, and the loops that match blocks and reap stragglers.
+//   - RunSummary, InclusionSummary — the end-of-run tallies.
+//   - BlockCollector — per-block chain data, gathered independently of the
+//     transactions this run sent.
+//
+// # Zero values and sentinels
+//
+// Three of them carry meaning, and each one exists because the obvious default
+// would lie.
+//
+//   - Outcome's zero value is outcomeUnset, not a real state. A transaction
+//     nobody classified counts as Unrecorded, which has no legitimate producer:
+//     a non-zero count means sei-load has a bug.
+//   - InclusionSummary.InflightAtShutdown is meaningful only after both the
+//     sender and the tracker have joined. Read earlier it is a race.
+//   - RunSummary.InclusionTracked separates a run with no tracker from a tracked
+//     run that saw no inclusions. Both would otherwise report zero.
+//
+// # Concurrency
+//
+// Three lock domains, and one rule that spans them.
+//
+//   - Collector.mu guards every counter and sample. RecordTransaction and
+//     RecordOutcome take it themselves; recordOperation is called with it
+//     already held.
+//   - TPSWindow.mu guards the rolling rate window.
+//   - InclusionTracker.state guards the in-flight registry.
+//
+// The rule: report an outcome to the collector outside the tracker's registry
+// lock. The sender blocks on that lock at every send completion, so work held
+// under both puts collector contention into the latency this package reports.
+//
+// # Invariants
+//
+// sender/doc.go owns the conservation identity and states it there.
+// TestInclusion_Conservation asserts it.
+//
+// The claim that Outcome's states partition every accepted transaction is
+// documented and not yet guarded. The test belongs with the change that wires
+// the tracker.
+//
+// # Not documented yet
+//
+// Lifecycle, and ownership boundaries between the collector and the tracker.
+// Both describe the tracker loop that reports outcomes, which does not exist.
+// They arrive with it.
+package stats

stats/inclusion_tracker.go

@@ -124,20 +124,24 @@ func (t *InclusionTracker) Register(tx *types.LoadTx) {
 		s.inflight[hash] = &entry{tx: tx, registeredAt: time.Now()}
 	}
 	if droppedAtCap {
-		t.recordOutcome("dropped_at_cap", tx.Scenario)
+		t.recordOutcome(OutcomeDroppedAtCap, tx.Scenario)
 	}
 }
 
 // recordOutcome counts one tx that left the registry un-included. Callers emit
 // outside the registry lock: the sender blocks on that lock at every send
 // completion, so time spent holding it lands in the latency this package
 // reports.
-func (t *InclusionTracker) recordOutcome(outcome string, scenario *types.TxScenario) {
+// The parameter is an Outcome rather than a string so that one type owns every
+// value this label can take. Passing a literal here would leave the metric and
+// Outcome.String() as two independent sources for one wire contract, and a
+// rename of either would orphan a dashboard while the other's test stayed green.
+func (t *InclusionTracker) recordOutcome(outcome Outcome, scenario *types.TxScenario) {
 	inclusionOutcome.Add(context.Background(), 1, metric.WithAttributes(
 		attribute.String("chain_id", t.seiChainID),
 		attribute.String("scenario", scenario.Name),
 		attribute.String("operation", scenario.Operation),
-		attribute.String("outcome", outcome),
+		attribute.String("outcome", outcome.String()),
 	))
 }
 
@@ -289,7 +293,7 @@ func (t *InclusionTracker) reap() {
 	// A stalled chain can expire the whole registry in one sweep, so the counter
 	// per victim runs after the lock is released.
 	for _, scenario := range expired {
-		t.recordOutcome("expired", scenario)
+		t.recordOutcome(OutcomeExpired, scenario)
 	}
 }
 

stats/metrics.go

@@ -68,7 +68,7 @@ var (
 
 	inclusionOutcome = must(meter.Int64Counter(
 		"inclusion_outcome",
-		metric.WithDescription("In-flight txs that left the registry un-included, by outcome (expired, dropped_at_cap)"),
+		metric.WithDescription("In-flight txs that left the registry un-included, by outcome. See stats.Outcome for the values."),
 		metric.WithUnit("{transactions}")))
 
 	inclusionBlockGaps = must(meter.Int64Counter(