refactor(stats): rename InclusionTracker to OutcomeTracker, and close out the spec

main.go

@@ -232,7 +232,7 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command) error {
 	logger := stats.NewLogger(collector, cfg.Settings.StatsInterval.ToDuration(), cfg.Settings.ReportPath, cfg.Settings.Debug)
 	rng := generator.ResolveSeed(cfg)
 	var ramper *sender.Ramper
-	inclusion := utils.None[*stats.InclusionTracker]()
+	inclusion := utils.None[*stats.OutcomeTracker]()
 
 	err = scope.Run(ctx, func(ctx context.Context, s scope.Scope) error {
 		// The generator deploys as it is built, so resolve who signs those
@@ -293,7 +293,7 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command) error {
 		// would all reap as expired and pollute the inclusion stats.
 		if len(cfg.Endpoints) > 0 && cfg.Settings.TrackReceipts && !cfg.Settings.DryRun {
 			reapAfter := cfg.Settings.InclusionReapAfter.ToDuration()
-			inclusionTracker := stats.NewInclusionTracker(
+			inclusionTracker := stats.NewOutcomeTracker(
 				cfg.SeiChainID,
 				reapAfter,
 				inclusionRegistryCap(cfg.Settings.MaxInFlight, cfg.Settings.TPS, reapAfter),
@@ -309,9 +309,8 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command) error {
 			// and a node trailing its peers by less than the readiness threshold
 			// stays in service while it does. Reading both from one node makes
 			// the run internally consistent, which is what TOT-022 requires.
-			trackingEndpoint := cfg.ReceiptEndpoint
-			if trackingEndpoint == "" {
-				trackingEndpoint = cfg.Endpoints[0]
+			trackingEndpoint, shared := trackingEndpoint(cfg)
+			if shared {
 				log.Printf("⚠️  Tracking from the load endpoint %s. Set "+
 					"receiptEndpoint to a node taking no send load: a receipts "+
 					"read costs the serving node work that grows with the block's "+
@@ -461,6 +460,21 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command) error {
 	return err
 }
 
+// trackingEndpoint picks the node the tracker reads status and heads from, and
+// reports whether that is the same node the run sends load to.
+//
+// Both signals come from this one value, so the caller cannot take heads from
+// one node and status from another (TOT-022). Falling back to the load endpoint
+// is what makes the tracker work on a single-node deployment (TOT-021); the
+// caller says so out loud, because the reads then land on the system under test
+// and count against the capacity the run exists to measure.
+func trackingEndpoint(cfg *config.LoadConfig) (endpoint string, sharedWithLoad bool) {
+	if cfg.ReceiptEndpoint != "" {
+		return cfg.ReceiptEndpoint, false
+	}
+	return cfg.Endpoints[0], true
+}
+
 // endedOnRunContext reports whether err is just the run finishing: its duration
 // elapsed, or the operator signalled it. Both are success.
 //

main_test.go

@@ -4,6 +4,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/sei-protocol/sei-load/config"
+	"github.com/sei-protocol/sei-load/utils"
 	"os"
 	"path/filepath"
 	"testing"
@@ -55,3 +57,57 @@ func TestEndedOnRunContext(t *testing.T) {
 		})
 	}
 }
+
+// TestStatusReadsLandOnTheTrackingNode covers T018 / TOT-021.
+//
+// The tracker's own reads otherwise land on the system under test, where they
+// count against the capacity the run exists to measure. A receipts call costs
+// the serving node work that grows with the block's transaction count, which is
+// exactly what a load run makes large.
+func TestStatusReadsLandOnTheTrackingNode(t *testing.T) {
+	got, shared := trackingEndpoint(&config.LoadConfig{
+		Endpoints:       []string{"http://load-node:8545"},
+		ReceiptEndpoint: "http://peer-node:8545",
+	})
+
+	require.Equal(t, "http://peer-node:8545", got,
+		"the tracker read from the node under load while a peer holding the same blocks sat idle")
+	require.False(t, shared)
+}
+
+// TestOneNodeStillWorks is the other half of TOT-021: the run MUST still work
+// when only one node is available. A tracker that refused to start without a
+// second endpoint would make outcome tracking unavailable on every single-node
+// deployment, which is most of them.
+func TestOneNodeStillWorks(t *testing.T) {
+	got, shared := trackingEndpoint(&config.LoadConfig{
+		Endpoints: []string{"http://only-node:8545"},
+	})
+
+	require.Equal(t, "http://only-node:8545", got)
+	require.True(t, shared,
+		"the run took its reads from the load node without saying so, so nobody knows the "+
+			"measurement is competing with the workload")
+}
+
+// TestBothSignalsShareOneNode covers T020 / TOT-022.
+//
+// A head notification carries the raw committed height. A status read resolves
+// through a watermark behind it. The two disagree on one node already, and
+// taking them from different nodes adds peer lag on top, which can reach minutes
+// while both nodes still report healthy.
+//
+// The tracker takes one endpoint and derives the head stream from it, so the
+// caller cannot split them. This pins that: the WebSocket endpoint follows the
+// tracking node, never the load node.
+func TestBothSignalsShareOneNode(t *testing.T) {
+	tracking, _ := trackingEndpoint(&config.LoadConfig{
+		Endpoints:       []string{"http://load-node:8545"},
+		ReceiptEndpoint: "http://peer-node:8545",
+	})
+
+	require.Equal(t, "ws://peer-node:8546", utils.GetWSEndpoint(tracking),
+		"the head stream came from a different node than the status reads")
+	require.NotEqual(t, utils.GetWSEndpoint("http://load-node:8545"), utils.GetWSEndpoint(tracking),
+		"the two nodes resolve to one WebSocket endpoint, so this test proves nothing")
+}

sender/doc.go

@@ -6,7 +6,7 @@
 // the sender queue contains unsigned transactions; the sender loop stamps the
 // attempt, signs the tx, and calls the go-ethereum client (eth_sendRawTransaction).
 // Inclusion, when tracked, is observed by the
-// block-indexed [stats.InclusionTracker] (see Inclusion stage below), not by
+// block-indexed [stats.OutcomeTracker] (see Inclusion stage below), not by
 // per-tx receipt polling. A shared [golang.org/x/time/rate.Limiter] is
 // the single rate authority for the whole pipeline; the [Ramper] drives its
 // limit up or down via SetLimit.
@@ -89,7 +89,7 @@
 // # Inclusion stage
 //
 // When enabled (--track-receipts), the sender hands each successful send to the
-// [stats.InclusionTracker] at send-completion (after OnComplete, only on a nil
+// [stats.OutcomeTracker] at send-completion (after OnComplete, only on a nil
 // send error). The tracker subscribes to new heads, reads each arriving block's
 // receipts once (one request per block, whatever the block's tx count), and
 // stamps InclusionTime on every matched in-flight tx with the block's
@@ -108,8 +108,9 @@
 // and accepted ⊆ succeeded (only a successful send is accepted). The left side
 // is accepted rather than registered, because a tx refused at the cap is
 // counted by dropped_at_cap and never entered the registry.
-// dropped_at_handoff is a term of the identity and nothing produces it yet; the
-// hand-off channel is what will. The inclusion
+// dropped_at_handoff is what the sender could not hand over: Submit never
+// blocks, so a full channel drops and counts rather than stalling the send path
+// it was called from. The inclusion
 // denominator is succeeded (txs_accepted), never a minted "registered" series;
 // dropped_at_cap txs are excluded from it. inflight_at_shutdown is read only
 // after both the senders and the tracker have joined.

sender/sender_test.go

@@ -144,7 +144,7 @@ func TestShardedSender_DryRunWithoutEndpoints(t *testing.T) {
 		ChainID:    1,
 		SeiChainID: "test-chain",
 		Settings:   &settings,
-	}, rate.NewLimiter(rate.Inf, 1), stats.NewCollector(), utils.None[*stats.InclusionTracker]())
+	}, rate.NewLimiter(rate.Inf, 1), stats.NewCollector(), utils.None[*stats.OutcomeTracker]())
 	account := types.NewAccount(true)
 
 	require.NoError(t, scope.Run(t.Context(), func(ctx context.Context, s scope.Scope) error {
@@ -345,7 +345,7 @@ func newTestShardedSender(endpoints []string) *ShardedSender {
 		ChainID:    1,
 		Endpoints:  endpoints,
 		Settings:   &settings,
-	}, rate.NewLimiter(rate.Inf, 1), stats.NewCollector(), utils.None[*stats.InclusionTracker]())
+	}, rate.NewLimiter(rate.Inf, 1), stats.NewCollector(), utils.None[*stats.OutcomeTracker]())
 }
 
 func testGeneratorConfigWithAccounts(endpoints []string, accountCount int, newAccountRate float64) *config.LoadConfig {

sender/sharded_sender.go

@@ -24,12 +24,12 @@ type ShardedSender struct {
 	queue     *TxsQueue
 	limiter   *rate.Limiter // Shared rate limiter for transaction sending
 	collector *stats.Collector
-	inclusion utils.Option[*stats.InclusionTracker]
+	inclusion utils.Option[*stats.OutcomeTracker]
 }
 
 // NewShardedSender creates a new sharded sender.
 // Txs of each shard are sent sequentially, using a single eth client.
-func NewShardedSender(cfg *config.LoadConfig, limiter *rate.Limiter, collector *stats.Collector, inclusion utils.Option[*stats.InclusionTracker]) *ShardedSender {
+func NewShardedSender(cfg *config.LoadConfig, limiter *rate.Limiter, collector *stats.Collector, inclusion utils.Option[*stats.OutcomeTracker]) *ShardedSender {
 	return &ShardedSender{
 		cfg:       cfg,
 		queue:     NewTxsQueue(cfg.Settings.MaxInFlight),

stats/doc.go

@@ -11,7 +11,7 @@
 //     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
+//   - OutcomeTracker — 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
@@ -42,7 +42,7 @@
 //     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 and the watermarks a
+//   - OutcomeTracker.state guards the in-flight registry and the watermarks a
 //     reap consults: the first height the run read, the highest head it has
 //     taken off the wire, and the highest whose block it has read.
 //
@@ -60,7 +60,7 @@
 // and reads it after they join, so a read of its counts is only final at that
 // point.
 //
-// InclusionTracker runs for the length of the run. Run dials the receipt
+// OutcomeTracker runs for the length of the run. Run dials the receipt
 // endpoint and proves it answers. That proof can refuse the run: an endpoint
 // that never answers, refuses the method, or replies with something other than
 // JSON-RPC ends the run with ErrEndpointUnusable rather than letting it report

stats/inclusion_run_test.go

@@ -17,7 +17,7 @@ import (
 )
 
 // headServer serves eth_subscribe("newHeads") over WebSocket, so a test can
-// drive InclusionTracker.Run itself rather than the helpers underneath it.
+// drive OutcomeTracker.Run itself rather than the helpers underneath it.
 //
 // Run is where the ordering lives: the pump stamps a head's arrival, the head
 // loop reads its block, the reap decides what to call a transaction, and a dead
@@ -112,10 +112,10 @@ func (h *headServer) kill() {
 // and returns the tracker plus a stop function.
 func runTracker(
 	t *testing.T, hs *headServer, src receiptSource, reapAfter time.Duration,
-) (*InclusionTracker, context.CancelFunc, <-chan error) {
+) (*OutcomeTracker, context.CancelFunc, <-chan error) {
 	t.Helper()
-	tr := newInclusionTrackerWithSource(
-		NewInclusionTracker("test-chain", reapAfter, 1000, testTPS, true, NewCollector()), src)
+	tr := newOutcomeTrackerWithSource(
+		NewOutcomeTracker("test-chain", reapAfter, 1000, testTPS, true, NewCollector()), src)
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan error, 1)
 	go func() { done <- tr.Run(ctx, hs.endpoint()) }()
@@ -238,7 +238,7 @@ func TestADeadSubscriptionStopsTheRunSayingAnything(t *testing.T) {
 }
 
 // headsSeen reports the highest head the tracker has taken off the wire.
-func headsSeen(tr *InclusionTracker) uint64 {
+func headsSeen(tr *OutcomeTracker) uint64 {
 	for s := range tr.state.Lock() {
 		return s.headsReceived
 	}

stats/metrics.go

@@ -108,17 +108,17 @@ var (
 		metric.WithUnit("{transactions}")))
 )
 
-// meteredInclusionTrackers backs the inclusion_inflight gauge: each tracker
+// meteredOutcomeTrackers backs the inclusion_inflight gauge: each tracker
 // registers so the callback can sample its in-flight map under lock.
-var meteredInclusionTrackers = struct {
+var meteredOutcomeTrackers = struct {
 	lock     sync.RWMutex
-	trackers []*InclusionTracker
+	trackers []*OutcomeTracker
 }{}
 
-func meterInclusionInflight(t *InclusionTracker) {
-	meteredInclusionTrackers.lock.Lock()
-	defer meteredInclusionTrackers.lock.Unlock()
-	meteredInclusionTrackers.trackers = append(meteredInclusionTrackers.trackers, t)
+func meterInclusionInflight(t *OutcomeTracker) {
+	meteredOutcomeTrackers.lock.Lock()
+	defer meteredOutcomeTrackers.lock.Unlock()
+	meteredOutcomeTrackers.trackers = append(meteredOutcomeTrackers.trackers, t)
 }
 
 func init() {
@@ -127,9 +127,9 @@ func init() {
 		metric.WithDescription("Current size of the inclusion tracker's in-flight tx registry"),
 		metric.WithUnit("{transactions}"),
 		metric.WithInt64Callback(func(_ context.Context, observer metric.Int64Observer) error {
-			meteredInclusionTrackers.lock.RLock()
-			defer meteredInclusionTrackers.lock.RUnlock()
-			for _, t := range meteredInclusionTrackers.trackers {
+			meteredOutcomeTrackers.lock.RLock()
+			defer meteredOutcomeTrackers.lock.RUnlock()
+			for _, t := range meteredOutcomeTrackers.trackers {
 				for s := range t.state.Lock() {
 					observer.Observe(int64(len(s.inflight)),
 						metric.WithAttributes(attribute.String("chain_id", t.seiChainID)))

stats/outcome.go

@@ -2,7 +2,7 @@ package stats
 
 // Outcome is what became of one transaction the endpoint accepted.
 //
-// InclusionTracker reports one of these for every transaction it registers, to
+// OutcomeTracker reports one of these for every transaction it registers, to
 // the inclusion_outcome metric and to Collector.RecordOutcome. The run report
 // does not read the collector's counts yet.
 //