diff --git a/README.md b/README.md index e762f7b..653be04 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ Edit `my-config.json`: ```json { "endpoints": ["http://localhost:8545"], + "receiptEndpoint": "http://localhost:8546", "chainId": 713714, "scenarios": [ {"name": "EVMTransfer", "weight": 100} @@ -38,6 +39,21 @@ Edit `my-config.json`: } ``` +`endpoints` take the load. `receiptEndpoint` is the node the inclusion tracker +uses, and it should be a node taking no send load. The tracker takes both its +head subscription and its receipt reads from that one node: a head carries the +raw committed height while a status read resolves through a watermark behind it, +so heads from one node and reads from another disagree by that pair's peer lag. On seid, a +receipts read costs the serving node work that grows with the square of the +block's transaction count, so a node doing both degrades, and the tracker +reports that degradation as a chain result. Leave it out and the tracker reads +from `endpoints[0]`, which is fine for a local chain and not for a load run. + +Time `eth_getBlockReceipts` against the chain you are testing, on a block as +wide as the run will produce, before you turn `trackReceipts` on. A read slower +than the block interval makes the tracker blind, and a blind tracker reports +`status_unavailable` rather than a number. + ### 3. Run ```bash @@ -54,7 +70,7 @@ Edit `my-config.json`: | `--buffer-size, -b` | 1000 | Sender queue size | | `--dry-run` | false | Simulate without sending | | `--debug` | false | Log each transaction | -| `--track-receipts` | false | Enable the block-indexed tx→inclusion tracker (stamps InclusionTime; reports included/expired/inflight-at-shutdown) | +| `--track-receipts` | false | Read each block's receipts and record what the chain did with every transaction: committed, reverted, expired, status-unavailable, dropped-at-cap, or still in flight. The counts reach the `inclusion_outcome` metric; the closing log line carries the inclusion tally. The run report does not carry them yet. Set `receiptEndpoint` in the config with it | | `--inclusion-reap-after` | 30s | How long an un-included tx waits before reaping as expired (tune to expected inclusion time on congested chains) | | `--track-blocks` | false | Track block statistics | | `--track-user-latency` | false | Track user latency metrics | diff --git a/config/config.go b/config/config.go index 9370f7e..fc336e4 100644 --- a/config/config.go +++ b/config/config.go @@ -31,12 +31,28 @@ type LoadConfig struct { // operator reviews and commits. Empty writes nothing. ChainRecordPath string `json:"chainRecordPath,omitempty"` // SeiChainID is the textual chain ID used for tagging metric collection. - SeiChainID string `json:"seiChainID,omitempty"` - Endpoints []string `json:"endpoints"` - Accounts *AccountConfig `json:"accounts,omitempty"` - Scenarios []Scenario `json:"scenarios,omitempty"` - MockDeploy bool `json:"mockDeploy,omitempty"` - Settings *Settings `json:"settings,omitempty"` + SeiChainID string `json:"seiChainID,omitempty"` + Endpoints []string `json:"endpoints"` + // ReceiptEndpoint is the node the inclusion tracker uses. It takes both the + // head subscription and the receipt reads from it, because a head from one + // node and a status read from another disagree by that pair's peer lag. + // Empty uses Endpoints[0]. + // + // Name a node that takes no send load. On seid, eth_getBlockReceipts costs + // the node answering it work that grows with the square of the block's + // transaction count. A node that both takes the send load and answers that + // read degrades, and the tracker reports the degradation as a chain result. + // + // This bounds what the tracking node can keep up with. Time + // eth_getBlockReceipts against the target chain, on a block as wide as the + // run will produce, before enabling trackReceipts. A read slower than the + // chain's block interval makes the tracker blind, and a blind tracker + // reports status_unavailable rather than a number. + ReceiptEndpoint string `json:"receiptEndpoint,omitempty"` + Accounts *AccountConfig `json:"accounts,omitempty"` + Scenarios []Scenario `json:"scenarios,omitempty"` + MockDeploy bool `json:"mockDeploy,omitempty"` + Settings *Settings `json:"settings,omitempty"` // Funding, when set, funds the generated account pool from a root key at // startup so the run works against a real chain. See funding.go. Funding *FundingConfig `json:"funding,omitempty"` diff --git a/main.go b/main.go index 91d643d..6224541 100644 --- a/main.go +++ b/main.go @@ -288,8 +288,7 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command) error { }) } - // The --track-receipts flag now enables the block-indexed inclusion - // tracker (the lossy per-tx receipt path is retired). + // --track-receipts enables the block-indexed inclusion tracker. // Not wired under --dry-run: simulated sends never hit the chain, so they // would all reap as expired and pollute the inclusion stats. if len(cfg.Endpoints) > 0 && cfg.Settings.TrackReceipts && !cfg.Settings.DryRun { @@ -299,10 +298,26 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command) error { reapAfter, inclusionRegistryCap(cfg.Settings.MaxInFlight, cfg.Settings.TPS, reapAfter), cfg.Settings.ArrivalModel == config.ArrivalModelOpenLoop, + collector, ) inclusion = utils.Some(inclusionTracker) + // Heads and receipts come from the same node, always. A head + // notification carries the raw committed height while a status read + // resolves through a watermark behind it, so the two disagree even on + // one node; taking them from different nodes adds peer lag on top, + // 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] + 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 "+ + "transaction count.", trackingEndpoint) + } s.SpawnBgNamed("inclusion tracker", func() error { - return inclusionTracker.Run(ctx, cfg.Endpoints[0]) + return inclusionTracker.Run(ctx, trackingEndpoint) }) } @@ -419,10 +434,16 @@ func runLoadTest(ctx context.Context, cmd *cobra.Command) error { summary.InclusionTracked = true summary.Included = incl.Included summary.Expired = incl.Expired + summary.StatusUnavailable = incl.StatusUnavailable summary.DroppedAtCap = incl.DroppedAtCap summary.InflightAtShutdown = incl.InflightAtShutdown - log.Printf("📦 Inclusion: included=%d expired=%d dropped_at_cap=%d inflight_at_shutdown=%d", - incl.Included, incl.Expired, incl.DroppedAtCap, incl.InflightAtShutdown) + log.Printf("📦 Inclusion: included=%d expired=%d status_unavailable=%d dropped_at_cap=%d inflight_at_shutdown=%d", + incl.Included, incl.Expired, incl.StatusUnavailable, incl.DroppedAtCap, + incl.InflightAtShutdown) + if incl.StatusUnavailable > 0 { + log.Printf("⚠️ %d txs were in flight while a receipt read failed. "+ + "The run cannot say whether the chain took them.", incl.StatusUnavailable) + } } collector.EmitRunSummary(ctx, summary) if d := cfg.Settings.PostSummaryFlushDelay.ToDuration(); d > 0 { diff --git a/sender/doc.go b/sender/doc.go index b8fc3f0..bfdb8cc 100644 --- a/sender/doc.go +++ b/sender/doc.go @@ -90,29 +90,52 @@ // // When enabled (--track-receipts), the sender hands each successful send to the // [stats.InclusionTracker] at send-completion (after OnComplete, only on a nil -// send error). The tracker subscribes to new heads, fetches each arriving -// block's body once (O(blocks), not O(txs)), and stamps InclusionTime on every -// matched in-flight tx with the block's header-ARRIVAL time. Un-included txs are -// reaped as expired after reapAfter. inclusion_latency (arrival minus +// 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 +// header-ARRIVAL time. A receipt carries the execution status, so a matched tx +// resolves to committed or reverted rather than to one state covering both. +// Un-included txs are reaped after reapAfter. inclusion_latency (arrival minus // IntendedSendTime) is an open-loop-only measure; in closed-loop IntendedSendTime // is enqueue time, so the latency sample is omitted (counts are tracked in both). // -// Conservation. registered == included + expired + inflight_at_shutdown, and -// registered ⊆ succeeded (only successful sends are registered). The inclusion +// Conservation. Over [stats.Outcome]'s terminal states, +// +// accepted == committed + reverted + status_unavailable +// + expired + dropped_at_cap + dropped_at_handoff +// + inflight_at_shutdown +// +// 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 // 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. // // Accepted boundaries. (1) WS gaps degrade conservatively: a missed head is -// counted (block_gaps) but never backfilled, so its txs reap as expired — -// an undercount of inclusions, never a miscount. (2) Reorgs use -// first-observation-wins (stamp + delete); the inclusion-time error is bounded -// by reorg_depth × block_time, with no canonical reconciliation. (3) A single -// fetch endpoint (Endpoints[0], shared with the block collector) adds a small -// read load. (4) InclusionTime is the header-arrival wall clock, not fetch -// completion and not header.Time. (5) A failed block-body fetch is counted -// (block_fetch_errors) and not retried — that block's txs reap as expired, the -// same conservative undercount as a WS gap. (6) A tx registered after its +// counted (block_gaps) and never backfilled. The run never read those blocks, +// so a tx in flight across the gap reaches status_unavailable rather than +// expired: expired is a claim about the chain, and the run has no grounds for +// one about a block it did not read. (2) Reorgs use +// first-observation-wins (stamp + delete). This fixes an execution status as +// well as a time: a tx that reverted on an orphaned block and committed on the +// canonical one keeps the first answer. Sei's finality makes the window small, +// and there is no canonical reconciliation. (3) The tracker reads receipts from +// receiptEndpoint when a run names one, and from Endpoints[0] otherwise. A +// receipts read costs the serving node work that grows with the block's tx +// count, so a run at high TPS should name a second node. (4) InclusionTime is +// the header-arrival wall clock, not fetch completion and not header.Time. +// (5) A height the serving node has not reached is read again until a budget +// runs out, because a receipt node trailing the head node is ordinary. One head +// spends a bounded total on those re-reads, so a height the sweep does not reach +// waits for the next head rather than becoming a hole. Any other +// read failure is counted by reason (block_fetch_errors) and not retried, since +// a retry piles requests onto an endpoint already failing. Either way a tx in +// flight across an unread height reaches status_unavailable rather than expired. +// An empty array is not an unread height: a block carrying no EVM transaction +// answers that way, and most blocks on an idle chain do. (6) A tx registered after its // including block was already scanned is missed and reaps as expired — bounded // by the microsecond register window versus block time, a rare conservative // undercount, the same direction as a WS gap. diff --git a/sender/sharded_sender.go b/sender/sharded_sender.go index ca798ef..92214ac 100644 --- a/sender/sharded_sender.go +++ b/sender/sharded_sender.go @@ -124,7 +124,7 @@ func (ss *ShardedSender) Run(ctx context.Context) error { // Queue for inclusion check. if inclusion, ok := ss.inclusion.Get(); ok { - inclusion.Register(tx) + inclusion.Register(ctx, tx) } ss.queue.PopSent(addr) return nil diff --git a/stats/collector.go b/stats/collector.go index def47be..109b974 100644 --- a/stats/collector.go +++ b/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], + Reverted: samples.outcomes[OutcomeReverted], + 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,31 @@ type OperationStats struct { P99Latency time.Duration SampleCount int Window time.Duration + + // Three layers, not one. Committed and Reverted 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. + // + // Reverted is the execution status a receipt reported: the chain took the + // transaction and it did nothing. RunSummary.Failed is a send that returned + // an error, which is a different layer, and the two carry different names so + // an operator reading one series beside the other cannot conflate them. + // + // 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 for a run with --track-receipts off, because the + // inclusion tracker is the only producer. + Committed uint64 + Reverted uint64 + Expired uint64 + DroppedAtCap uint64 + DroppedAtHandoff uint64 + StatusUnavailable uint64 + Unrecorded uint64 } // operationSamples accumulates one operation's counts and its most recent @@ -369,6 +442,11 @@ type operationSamples struct { count uint64 successes uint64 samples []latencySample + + // outcomes counts terminal states by Outcome. 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. diff --git a/stats/collector_outcome_test.go b/stats/collector_outcome_test.go new file mode 100644 index 0000000..a701114 --- /dev/null +++ b/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.OutcomeReverted) + 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].Reverted) + require.Equal(t, uint64(1), got[rmw].Committed, "one operation's outcomes reached another's count") + require.Zero(t, got[rmw].Reverted) +} + +// 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 }}, + {"reverted", stats.OutcomeReverted, func(s stats.OperationStats) uint64 { return s.Reverted }}, + {"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, "reverted", stats.OutcomeReverted.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) +} diff --git a/stats/doc.go b/stats/doc.go new file mode 100644 index 0000000..1686b91 --- /dev/null +++ b/stats/doc.go @@ -0,0 +1,106 @@ +// 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 +// +// Four 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. +// - ErrEndpointUnusable marks a refusal to start. It is deliberately not a +// context error: main reads a cancelled or expired run context as the run +// ending normally and exits zero, so a refusal wrapping one would report a +// run that carried no load as a run that finished. +// +// # 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 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. +// +// 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. +// +// A third goroutine takes the registry lock: the inclusion_inflight gauge +// callback, on the metric SDK's collection goroutine. It takes no collector +// lock. Nothing may take both. +// +// # Lifecycle +// +// Collector outlives every producer. main builds it before the senders start +// 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 +// 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 +// numbers it did not measure. Otherwise Run subscribes to new heads, and then +// three goroutines drive it: a pump stamping each head's arrival, the head loop +// matching each arriving block once, and the reap loop evicting transactions +// that outlived reapAfter. All three end when the run context does, and a +// transaction still in the registry then is counted as inflight_at_shutdown, +// which is why the conservation identity is only readable after they have +// joined. +// +// A head stream that ends on its own is the exception. It stops the tracking and +// not the run, and everything in flight is settled as status_unavailable rather +// than left to fill the registry: nothing later is read, so nothing later is +// evidence about the chain. +// +// # Ownership boundaries +// +// The tracker owns every transaction's terminal state. It decides, and it +// reports once. The collector owns the counts and never decides. +// +// The reference runs one way. The tracker holds a *Collector; the collector +// knows nothing of the tracker. That is what keeps the lock rule above +// enforceable by structure rather than by convention. +// +// The tracker's registry holds a *types.LoadTx it does not own. The sender +// writes it once before the hand-off and never again, which is what makes it +// safe to read a scenario out of the registry and report on it after the lock +// is released. +// +// # Invariants +// +// sender/doc.go owns the conservation identity and states it there. +// TestInclusion_Conservation asserts the registry identity, and +// TestOutcomesPartitionEveryAcceptedTx asserts that Outcome's terminal states +// partition every accepted transaction. +// +// The watermarks that keep expired reachable are guarded through Run itself, by +// TestTheWatermarksKeepExpiredReachable: it drives one transaction accepted +// before the run read anything and one accepted after, because a single +// transaction moves both ways at once and cannot tell a working watermark from +// a dead one. +package stats diff --git a/stats/inclusion_outcome_test.go b/stats/inclusion_outcome_test.go new file mode 100644 index 0000000..3e1cd7d --- /dev/null +++ b/stats/inclusion_outcome_test.go @@ -0,0 +1,864 @@ +package stats + +import ( + "context" + "errors" + "math/big" + "net" + "strconv" + "syscall" + "testing" + "time" + + "github.com/ethereum/go-ethereum" + "github.com/ethereum/go-ethereum/rpc" + "github.com/sei-protocol/sei-load/types" + + ethtypes "github.com/ethereum/go-ethereum/core/types" + "github.com/stretchr/testify/require" +) + +// TestAllRevertedReceiptsYieldZeroCommitted fails when a reverted transaction and a +// committed one land in the same terminal state. +// Requirements: TOT-001 and SC-001. +func TestAllRevertedReceiptsYieldZeroCommitted(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Minute, 100, src) + key := OperationKey{Scenario: "storagerw", Operation: "write"} + + var receipts []blockReceipt + for i := uint64(1); i <= 5; i++ { + tx := loadTx(i, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + receipts = append(receipts, blockReceipt{ + Hash: tx.EthTx.Hash(), + Status: ethtypes.ReceiptStatusFailed, + HasStatus: true, + }) + } + src.SetReceipts(7, receipts...) + tr.matchBlock(context.Background(), 7, 0, time.Unix(1002, 0)) + + got := tr.collector.GetOperationStats()[key] + require.Zero(t, got.Committed, "a failed transaction counted as committed") + require.Equal(t, uint64(5), got.Reverted) + require.Equal(t, uint64(5), tr.Summary().Included, + "they were included; inclusion and execution are different questions") +} + +// TestReceiptsSeparateCommittedFromReverted covers the mixed block, which is the +// shape a real run produces. A block carrying both must split them. +// Requirements: TOT-001 and TOT-002. +func TestReceiptsSeparateCommittedFromReverted(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Minute, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + var receipts []blockReceipt + for i := uint64(1); i <= 4; i++ { + tx := loadTx(i, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + status := ethtypes.ReceiptStatusSuccessful + if i%2 == 0 { + status = ethtypes.ReceiptStatusFailed + } + receipts = append(receipts, blockReceipt{Hash: tx.EthTx.Hash(), Status: status, HasStatus: true}) + } + src.SetReceipts(9, receipts...) + tr.matchBlock(context.Background(), 9, 0, time.Unix(1002, 0)) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(2), got.Committed) + require.Equal(t, uint64(2), got.Reverted) + require.Zero(t, got.Unrecorded, "every matched transaction reached a real state") +} + +// TestOutcomesCarryTheOperation fails when the tracker labels an outcome by +// scenario alone. Two operations in one scenario would then share a count, and +// neither the report nor a dashboard could say which one degraded. +// Requirements: TOT-017 and TOT-006. +func TestOutcomesCarryTheOperation(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Minute, 100, src) + + read := loadTx(1, time.Unix(1000, 0)) + read.Scenario.Name, read.Scenario.Operation = "storagerw", "read" + write := loadTx(2, time.Unix(1000, 0)) + write.Scenario.Name, write.Scenario.Operation = "storagerw", "write" + tr.Register(context.Background(), read) + tr.Register(context.Background(), write) + + src.SetReceipts(3, + blockReceipt{Hash: read.EthTx.Hash(), Status: ethtypes.ReceiptStatusSuccessful, HasStatus: true}, + blockReceipt{Hash: write.EthTx.Hash(), Status: ethtypes.ReceiptStatusFailed, HasStatus: true}, + ) + tr.matchBlock(context.Background(), 3, 0, time.Unix(1002, 0)) + + stats := tr.collector.GetOperationStats() + require.Equal(t, uint64(1), stats[OperationKey{Scenario: "storagerw", Operation: "read"}].Committed) + require.Equal(t, uint64(1), stats[OperationKey{Scenario: "storagerw", Operation: "write"}].Reverted) + require.Zero(t, stats[OperationKey{Scenario: "storagerw", Operation: "read"}].Reverted, + "one operation's outcome reached another's count") +} + +// TestRequestsPerBlockDoNotTrackVolume fails when the request count rises with +// the transaction count. +// +// It measures requests this process issues. It says nothing about what one +// request costs the node that answers it, which is a separate constraint and a +// separate measurement. +// Requirements: TOT-009, TOT-015 and SC-004. +func TestRequestsPerBlockDoNotTrackVolume(t *testing.T) { + for _, volume := range []int{1, 50, 500} { + t.Run(strconv.Itoa(volume), func(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Minute, 10_000, src) + + var receipts []blockReceipt + for i := 1; i <= volume; i++ { + tx := loadTx(uint64(i), time.Unix(1000, 0)) + tr.Register(context.Background(), tx) + receipts = append(receipts, blockReceipt{ + Hash: tx.EthTx.Hash(), + Status: ethtypes.ReceiptStatusSuccessful, + HasStatus: true, + }) + } + src.SetReceipts(11, receipts...) + tr.matchBlock(context.Background(), 11, 0, time.Unix(1002, 0)) + + require.Equal(t, int64(1), src.FetchCount(), + "a block carrying %d transactions cost %d requests, not one", + volume, src.FetchCount()) + }) + } +} + +// TestReapedTransactionsReachTheCollector covers the path that does not go +// through a receipt. A transaction nothing named before the deadline is expired, +// and the collector has to hear about it or the states stop partitioning. +// Requirements: TOT-016. +func TestReapedTransactionsReachTheCollector(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + time.Sleep(time.Millisecond) + tr.reap(context.Background()) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.Expired) + require.Zero(t, got.Committed) +} + +// TestUnreadableBlockIsNotAChainVerdict fails when a block the run could not +// read is reported as a chain that left the transaction out. Expired is a claim +// about the chain; a read that failed supports no such claim. +// Requirements: TOT-004, TOT-020 and SC-011. +func TestUnreadableBlockIsNotAChainVerdict(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + + src.SetFetchErr(errors.New("connection refused")) + tr.matchBlock(context.Background(), 9, 0, time.Unix(1002, 0)) + tr.reap(context.Background()) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.StatusUnavailable, + "a block the run never read was reported as a chain result") + require.Zero(t, got.Expired) +} + +// TestAnIdleBlockIsNotAHole fails when a block carrying no EVM transaction is +// counted as one the run could not read. +// +// It is the most common answer on a live chain: 80 of 100 consecutive Sei +// mainnet blocks return an empty array. Counting those as holes makes expired +// unreachable, and a chain that stopped accepting work produces nothing but +// idle blocks, which is the one run where expired is the answer. +// Requirements: TOT-020. +func TestAnIdleBlockIsNotAHole(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + + src.SetReceipts(9) // the node holds the block and it carried nothing + tr.matchBlock(context.Background(), 9, 0, time.Unix(1002, 0)) + tr.reap(context.Background()) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.Expired, + "an idle block was read as a hole, so the chain's verdict was thrown away") + require.Zero(t, got.StatusUnavailable) +} + +// TestDuplicateRegistrationReachesATerminalState fails when two accepted +// transactions sharing one hash report one terminal state between them. The +// registry holds one slot per hash, and a resend after a send timeout produces +// exactly that collision. +// Requirements: TOT-003. +func TestDuplicateRegistrationReachesATerminalState(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Minute, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + tr.Register(context.Background(), tx) + + src.SetReceipts(3, blockReceipt{ + Hash: tx.EthTx.Hash(), Status: ethtypes.ReceiptStatusSuccessful, HasStatus: true, + }) + tr.matchBlock(context.Background(), 3, 0, time.Unix(1002, 0)) + + got := tr.collector.GetOperationStats()[key] + sum := got.Committed + got.Reverted + got.Expired + got.DroppedAtCap + + got.DroppedAtHandoff + got.StatusUnavailable + got.Unrecorded + require.Equal(t, uint64(2), sum+tr.Summary().InflightAtShutdown, + "two were accepted and %d reached a terminal state", sum) +} + +// TestTransactionsRegisteredAfterTheHoleStillExpire fails when one unreadable +// block turns every later transaction into status_unavailable. The hole covers +// the transactions in flight across it, and nothing after. +// Requirements: TOT-020 and SC-011. +func TestTransactionsRegisteredAfterTheHoleStillExpire(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + src.SetFetchErr(errors.New("connection refused")) + tr.matchBlock(context.Background(), 9, 0, time.Unix(1002, 0)) + + tx := loadTx(2, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + tr.reap(context.Background()) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.Expired, + "a transaction registered after the hole inherited it") + require.Zero(t, got.StatusUnavailable) +} + +// TestReceiptWithoutStatusIsNotAFailure fails when a receipt that carries a +// post-state root instead of a status is read as an execution failure. It says +// the transaction executed; it does not say how it ended. +// Requirements: TOT-004. +func TestReceiptWithoutStatusIsNotAFailure(t *testing.T) { + src := NewMockBlockSource() + tr := newTestTracker(t, time.Minute, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + src.SetReceipts(4, blockReceipt{Hash: tx.EthTx.Hash(), HasStatus: false}) + tr.matchBlock(context.Background(), 4, 0, time.Unix(1002, 0)) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.StatusUnavailable, + "an unreadable status was reported as a chain failure") + require.Zero(t, got.Reverted) +} + +// TestOutcomesPartitionEveryAcceptedTx is the guard stats/doc.go promises for +// the identity sender/doc.go states. It fails when an accepted transaction +// reaches no terminal state, or reaches two. +// +// Accepted rather than registered: a transaction refused at the cap never +// entered the registry, and the identity counts it all the same. +// +// Every reachable terminal state has a leg here. Leaving the reap out was worth +// a mutation test on its own, because dropping the whole reap report loop left +// a partition guard that still passed. +// Requirements: TOT-003 and SC-007. +func TestOutcomesPartitionEveryAcceptedTx(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + // The cap admits four; the last two of six are refused at it. + tr := newTestTracker(t, 30*time.Millisecond, 4, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + register := func(nonce uint64) *types.LoadTx { + tx := loadTx(nonce, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + return tx + } + + // Two commit and one reverts in block 5. The fourth is admitted and never + // named by a receipt, so it reaps as expired. + var receipts []blockReceipt + for i := uint64(1); i <= 3; i++ { + status := ethtypes.ReceiptStatusSuccessful + if i == 3 { + status = ethtypes.ReceiptStatusFailed + } + receipts = append(receipts, blockReceipt{ + Hash: register(i).EthTx.Hash(), Status: status, HasStatus: true, + }) + } + register(4) + // Two refused at the cap. + register(5) + register(6) + + src.SetReceipts(5, receipts...) + tr.matchBlock(ctx, 5, 0, time.Unix(1002, 0)) + + // A read that fails puts the run in a state it cannot attribute, so the + // fourth reaps as status_unavailable rather than expired. + src.SetFetchErr(errors.New("connection refused")) + tr.matchBlock(ctx, 6, 0, time.Unix(1003, 0)) + time.Sleep(40 * time.Millisecond) + tr.reap(ctx) + + got := tr.collector.GetOperationStats()[key] + sum := got.Committed + got.Reverted + got.Expired + got.DroppedAtCap + + got.DroppedAtHandoff + got.StatusUnavailable + got.Unrecorded + summary := tr.Summary() + require.Equal(t, uint64(6), sum+summary.InflightAtShutdown, + "six accepted, %d reached a terminal state and %d are still in flight", + sum, summary.InflightAtShutdown) + + // Each leg is reachable, so a later change cannot satisfy the sum by + // collapsing two states into one. + require.Equal(t, uint64(2), got.Committed) + require.Equal(t, uint64(1), got.Reverted) + require.Equal(t, uint64(2), got.DroppedAtCap) + require.Equal(t, uint64(1), got.StatusUnavailable) + require.Zero(t, got.Unrecorded, "an outcome no caller may report was recorded") + + // The registry counters and the outcome ledger describe the same + // transactions, so an operator reading the closing log line and the metric + // cannot see one transaction counted under two names. + require.Equal(t, got.Expired, summary.Expired) + require.Equal(t, got.StatusUnavailable, summary.StatusUnavailable) +} + +// TestPreflightFailsTheRunOnAnEndpointThatCannotAnswer fails when a run against +// a node that cannot serve receipts starts anyway. +// +// The probe asks for height 0, which sei-chain answers from a constant with no +// watermark, receipt store or sync state involved. A healthy Sei EVM RPC cannot +// fail it, so an error that survives every attempt disqualifies the endpoint +// whatever the error was, and the reason does not change the verdict. Such a run completes, reports +// every transaction un-included, and exits zero, which reads as a chain that +// accepted nothing. +// Requirements: TOT-021. +func TestPreflightFailsTheRunOnAnEndpointThatCannotAnswer(t *testing.T) { + cases := []struct { + name string + err error + wantErr bool + }{ + {"method_absent", errors.New("the method eth_getBlockReceipts does not exist/is not available"), true}, + {"not_listening", errors.New("dial tcp 10.0.0.1:8545: connect: connection refused"), true}, + {"no_such_host", errors.New("dial tcp: lookup rpc-0: no such host"), true}, + {"answers_nothing", context.DeadlineExceeded, true}, + {"answers_prose", errors.New("invalid character '<' looking for beginning of value"), true}, + // A Sei node serves genesis from a constant, ahead of its watermark and + // its receipt store, so not-found to height 0 is not a Sei EVM RPC. + {"not_found_at_genesis", ethereum.NotFound, true}, + {"reset_every_attempt", errors.New("read tcp 10.0.0.1:8545: read: connection reset by peer"), true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + src := NewMockBlockSource().SetFetchErr(tc.err) + tr := newTestTracker(t, time.Minute, 100, src) + + err := tr.preflight(context.Background(), "http://node:8545") + if !tc.wantErr { + require.NoError(t, err, "a busy endpoint failed the run instead of being retried") + return + } + require.ErrorContains(t, err, "receiptEndpoint", + "the error names no setting the operator can actually change") + require.NotContains(t, err.Error(), "--receipt-endpoint", + "the error names a flag that does not exist") + require.ErrorContains(t, err, "http://node:8545", "the error does not name the endpoint") + }) + } +} + +// TestNilReceiptDoesNotEndTheRun fails when an endpoint that returns a null +// array element panics the head loop. Nothing recovers there, so the run dies +// and loses every result it gathered. +// +// A null element is reported rather than refused. The run cannot see what that +// element was going to say, so the block is a hole for the transactions it did +// not name. It is not a hole for the ones it did: discarding a receipt the +// endpoint actually sent would throw away a known outcome to describe an +// unknown one. +// Requirements: TOT-020. +func TestNilReceiptDoesNotEndTheRun(t *testing.T) { + hash := loadTx(1, time.Unix(1000, 0)).EthTx.Hash() + + got, nulls := narrowReceipts([]*ethtypes.Receipt{ + nil, + {TxHash: hash, Status: ethtypes.ReceiptStatusSuccessful}, + nil, + }) + require.Equal(t, 2, nulls, "the nulls were not reported") + require.Equal(t, []blockReceipt{{Hash: hash, Status: 1, HasStatus: true}}, got, + "a receipt the endpoint did send was discarded") + + // An empty array is a block that carried no EVM transaction. It is the + // answer for most blocks on an idle chain, and it is not a loss. + got, nulls = narrowReceipts(nil) + require.Zero(t, nulls) + require.Empty(t, got) +} + +// TestAPartlyNullReadMatchesWhatArrivedAndCountsTheRest fails when a read that +// was partly unreadable either throws away the part that arrived or hides the +// part that did not. +// Requirements: TOT-020. +func TestAPartlyNullReadMatchesWhatArrivedAndCountsTheRest(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + named := loadTx(1, time.Unix(1000, 0)) + unnamed := loadTx(2, time.Unix(1000, 0)) + for _, tx := range []*types.LoadTx{named, unnamed} { + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + } + + // The endpoint sent one receipt and one null. + src.SetNulls(1).SetReceipts(4, blockReceipt{ + Hash: named.EthTx.Hash(), Status: ethtypes.ReceiptStatusSuccessful, HasStatus: true, + }) + tr.matchBlock(ctx, 4, 0, time.Unix(1002, 0)) + tr.reap(ctx) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.Committed, + "a receipt that arrived was discarded because another was null") + require.Equal(t, uint64(1), got.StatusUnavailable, + "the transaction the null would have named was blamed on the chain") + require.Zero(t, got.Expired) +} + +// TestASkippedHeadIsCountedAsAHole fails when a height the run never saw a head +// for goes uncounted. Its transactions would reap as expired, which is a claim +// about the chain for a block nothing read. +// Requirements: TOT-004. +func TestASkippedHeadIsCountedAsAHole(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + + last := tr.processHead(ctx, 10, 0, time.Unix(1002, 0), 0) + tr.processHead(ctx, 13, 0, time.Unix(1003, 0), last) // 11 and 12 missed + tr.reap(ctx) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.StatusUnavailable, + "a transaction that may have been in a skipped block was blamed on the chain") + require.Zero(t, got.Expired) +} + +// TestTheDuplicateLegReachesBothLedgers fails when a duplicate registration is +// counted on one operator-facing surface and not the other. +// Requirements: TOT-016. +func TestTheDuplicateLegReachesBothLedgers(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Minute, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + tr.Register(ctx, tx) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.StatusUnavailable) + require.Equal(t, got.StatusUnavailable, tr.Summary().StatusUnavailable, + "the closing log line and the metric disagree about the duplicate") +} + +// TestARefusedRunDoesNotLookLikeAFinishedOne fails when the preflight's refusal +// wraps a context sentinel. +// +// main treats a context.Canceled or a context.DeadlineExceeded as the run ending +// normally and exits zero. A refusal that wraps one is therefore a Complete run +// that carried no load, which anything downstream reads as a chain that included +// nothing: the false accusation this whole tracker exists to remove. +// Requirements: TOT-021. +func TestARefusedRunDoesNotLookLikeAFinishedOne(t *testing.T) { + for _, tc := range []struct { + name string + err error + }{ + {"answers_nothing", context.DeadlineExceeded}, + {"not_listening", &net.OpError{Op: "dial", Net: "tcp", Err: syscall.ECONNREFUSED}}, + {"not_json", errors.New("invalid character '<' looking for beginning of value")}, + } { + t.Run(tc.name, func(t *testing.T) { + src := NewMockBlockSource().SetFetchErr(tc.err) + tr := newTestTracker(t, time.Minute, 100, src) + + err := tr.preflight(context.Background(), "http://node:8545") + require.ErrorIs(t, err, ErrEndpointUnusable, "the refusal is not identifiable") + require.False(t, + errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded), + "the refusal reads as the run's own context ending, so the process exits zero") + }) + } +} + +// TestTheSummaryTermsAreDisjoint fails when one transaction lands in two terms +// of the closing log line, which invites an operator to add them up and get more +// than the run accepted. +// Requirements: TOT-003 and SC-010. +func TestTheSummaryTermsAreDisjoint(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Minute, 100, src) + + tx := loadTx(1, time.Unix(1000, 0)) + tr.Register(ctx, tx) + // A receipt carrying a post-state root instead of a status: it was in a + // block, and what it did there cannot be read. + src.SetReceipts(4, blockReceipt{Hash: tx.EthTx.Hash(), HasStatus: false}) + tr.matchBlock(ctx, 4, 0, time.Unix(1002, 0)) + + s := tr.Summary() + total := s.Included + s.Expired + s.StatusUnavailable + s.DroppedAtCap + s.InflightAtShutdown + require.Equal(t, uint64(1), total, + "one accepted transaction was counted %d times across the summary's terms", total) +} + +// TestAGapIsRecordedOnce fails when a missed-head range logs and counts per +// height. A long gap would then push the run's own summary out of any bounded log +// tail, which is the only diagnostic a failed run carries. +// Requirements: TOT-004. +func TestAGapIsRecordedOnce(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Minute, 100, src) + + last := tr.processHead(ctx, 10, 0, time.Unix(1002, 0), 0) + tr.processHead(ctx, 60, 0, time.Unix(1003, 0), last) // 49 heights missed + + for s := range tr.state.Lock() { + require.Equal(t, uint64(1), s.blindHeights, + "a 49-height gap was recorded %d times", s.blindHeights) + } +} + +// TestABacklogIsNotAChainVerdict fails when a transaction is called expired +// because the tracker had not got to its block yet. +// +// This is the case every other hole in this file misses. No read fails, no head +// is missed, no receipt is null. The chain includes the transaction, the node +// answers correctly, and the tracker is simply slower than the block rate, so +// the reap evicts the transaction before the block carrying it is opened. +// Reporting that as expired is a verdict about the chain drawn from a block the +// run had in hand and had not read. +// Requirements: TOT-004 and TOT-008. +func TestABacklogIsNotAChainVerdict(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + + // Heads 7 and 8 have arrived. Only 7 has been read. + tr.noteHeadReceived(7) + tr.noteHeadReceived(8) + tr.noteHeadResolved(7) + tr.reap(ctx) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.StatusUnavailable, + "a transaction was blamed on the chain while its block sat unread") + require.Zero(t, got.Expired) +} + +// TestACaughtUpTrackerStillReportsExpired fails when the backlog rule swallows +// the chain signal. A run that has read every head it received and still never +// saw the transaction is entitled to say the chain left it out, and that is the +// whole point of keeping the two states apart. +// Requirements: TOT-008 and SC-003. +func TestACaughtUpTrackerStillReportsExpired(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + + tr.noteHeadReceived(8) + tr.noteHeadResolved(8) + tr.reap(ctx) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.Expired, + "a caught-up run could not say the chain left a transaction out") + require.Zero(t, got.StatusUnavailable) +} + +// TestAStoppedHeadStreamIsNotAChainVerdict fails when a run whose head stream +// died keeps reporting chain verdicts. Nothing after that point is read at all. +// Requirements: TOT-004. +func TestAStoppedHeadStreamIsNotAChainVerdict(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + + tr.noteHeadReceived(8) + tr.noteHeadResolved(8) + tr.stopTracking(ctx) + tr.reap(ctx) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.StatusUnavailable, + "a run whose head stream died still claimed the chain left a tx out") + require.Zero(t, got.Expired) +} + +// TestAHeadIsStampedWhenItArrivesNotWhenItIsRead fails when the arrival stamp is +// taken at processing time. +// +// A tracker whose reads outrun the block interval builds a backlog. Stamping at +// processing would fold that backlog into InclusionTime and into every latency +// sample, so a run would report its own lateness as the chain's inclusion +// latency, and the error would grow for the length of the run. +// Requirements: TOT-022. +func TestAHeadIsStampedWhenItArrivesNotWhenItIsRead(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + tr := newTestTracker(t, time.Minute, 100, NewMockBlockSource()) + + headers := make(chan *ethtypes.Header, 4) + arrivals := make(chan headArrival, 4) + go func() { _ = tr.pumpHeads(ctx, headers, arrivals) }() + + sentAt := time.Now() + headers <- ðtypes.Header{Number: big.NewInt(7)} + + // The reader is busy. A stamp taken here would carry this delay. + time.Sleep(300 * time.Millisecond) + got := <-arrivals + + require.WithinDuration(t, sentAt, got.arrival, 150*time.Millisecond, + "the head was stamped %s after it arrived, so every latency sample "+ + "carries the tracker's own backlog", got.arrival.Sub(sentAt)) + + // And the head counts as received before anything reads its block. + for s := range tr.state.Lock() { + require.Equal(t, uint64(7), s.headsReceived) + require.Zero(t, s.headsResolved) + } +} + +// TestASingleMissedHeadIsCountedAsAHole fails when a gap of exactly one height +// goes unrecorded. One missed head is the common case, and its transactions +// would reap as expired for a block nothing read. +// Requirements: TOT-004. +func TestASingleMissedHeadIsCountedAsAHole(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + + last := tr.processHead(ctx, 10, 0, time.Unix(1002, 0), 0) + tr.processHead(ctx, 12, 0, time.Unix(1003, 0), last) // 11 alone was missed + tr.reap(ctx) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.StatusUnavailable, + "one missed head went unrecorded, so its txs were blamed on the chain") + require.Zero(t, got.Expired) +} + +// TestExpiredIsReachableInTheRecommendedTopology covers TOT-008 and SC-003, and +// guards the failure the reap's queue rule can cause. +// +// Refusing a chain verdict while a height sits queued is right, and it is only +// safe because the queue is normally empty. Taking heads and receipts from one +// node is what keeps it empty: a height cannot arrive as a head before the node +// that sent it holds the block. Split them, and the queue never empties, and +// expired stops being reachable at all. +// Requirements: TOT-008, TOT-022 and SC-003. +func TestExpiredIsReachableInTheRecommendedTopology(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + + // One node: every head the run sees is a block that node already holds, or + // holds within a retry, so every read resolves. + for h := uint64(7); h <= 10; h++ { + src.SetReceipts(h) + tr.noteHeadReceived(h) + tr.processHead(ctx, h, 0, time.Unix(1002, 0), h-1) + tr.noteHeadResolved(h) + } + tr.reap(ctx) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.Expired, + "a run that read every block it saw could not say the chain left a tx out") + require.Zero(t, got.StatusUnavailable) +} + +// Requirements: TOT-020 and TOT-004. +// TestTheWatermarkGapIsWaitedOutNotWrittenOff fails when a height the node has +// not finished writing is called a hole instead of waited for. +// +// seid publishes a head from Commit while the receipt store's writer is still +// async, so a height answers null for the gap between the two. That gap is one +// async write, not a chain fault: writing it off would raise the blind +// watermark, and because that watermark only grows, one such gap inside a +// transaction's reap window converts it. At any appreciable rate that makes +// expired unreachable and the run can never say the chain dropped anything. +func TestTheWatermarkGapIsWaitedOutNotWrittenOff(t *testing.T) { + ctx := context.Background() + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + src := NewMockBlockSource() + tr := newTestTracker(t, time.Nanosecond, 100, src) + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(ctx, tx) + + // The node has the head but not yet the receipts, then catches up. + src.FailTimes(2, ethereum.NotFound) + src.SetReceipts(7, blockReceipt{ + Hash: tx.EthTx.Hash(), Status: ethtypes.ReceiptStatusSuccessful, HasStatus: true, + }) + tr.matchBlock(ctx, 7, 0, time.Unix(1002, 0)) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.Committed, + "a height the node had not finished writing was written off") + for s := range tr.state.Lock() { + require.Zero(t, s.blindHeights, + "the watermark rose for a gap that resolved, which makes expired "+ + "unreachable for every transaction in the window") + } +} + +// Requirements: TOT-020 and TOT-023. +// TestARetryGivesUpAndBecomesAHole fails when a node that never catches up +// holds the head loop indefinitely. +// +// Head processing is serial, so a read that waits forever stops the run reading +// anything. Past its budget the height is a hole, which is honest and lets the +// loop move on. +func TestARetryGivesUpAndBecomesAHole(t *testing.T) { + ctx := context.Background() + src := NewMockBlockSource().SetFetchErr(ethereum.NotFound) + tr := newTestTracker(t, time.Minute, 100, src) + + start := time.Now() + tr.matchBlock(ctx, 7, 0, time.Unix(1002, 0)) + spent := time.Since(start) + + require.Less(t, spent, readTimeout*2, + "a node that never caught up held the head loop for %s", spent) + for s := range tr.state.Lock() { + require.Equal(t, uint64(1), s.blindHeights, + "a height the node never served was not counted as unread") + } +} + +// Requirements: TOT-020. +// TestOnlyAnUnreachedHeightIsRetried pins which errors the read retries. +// +// not_found is the one reason that loops, because it means the node has not +// finished writing that height and will answer differently in a moment. +// Anything else answered as well as it ever will, and retrying it spends the +// whole read budget on hundreds of requests against a node that is already +// wrong — pointed at a method whose cost grows with the block's transaction +// count. +func TestOnlyAnUnreachedHeightIsRetried(t *testing.T) { + cases := []struct { + name string + err error + want string + retried bool + }{ + {"node_has_not_reached_it", ethereum.NotFound, reasonNotFound, true}, + {"ingress_404_page", rpc.HTTPError{ + StatusCode: 404, Status: "404 Not Found", Body: []byte("404 page not found"), + }, reasonOther, false}, + {"gateway_filters_the_method", rpc.HTTPError{ + StatusCode: 403, Status: "403 Forbidden", + Body: []byte(`{"error":{"code":-32601,"message":"Method not found"}}`), + }, reasonOther, false}, + {"read_timed_out", context.DeadlineExceeded, reasonTimeout, false}, + {"connection_reset", errors.New("read tcp: connection reset by peer"), reasonOther, false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := fetchFailureReason(tc.err) + require.Equal(t, tc.want, got) + require.Equal(t, tc.retried, got == reasonNotFound, + "an error that answered as well as it ever will is retried") + }) + } +} + +// Requirements: TOT-020. +// TestAnErrorThatWillNotChangeCostsOneRequest fails when a permanent answer is +// retried for the whole read budget. +func TestAnErrorThatWillNotChangeCostsOneRequest(t *testing.T) { + src := NewMockBlockSource().SetFetchErr(rpc.HTTPError{ + StatusCode: 404, Status: "404 Not Found", Body: []byte("404 page not found"), + }) + tr := newTestTracker(t, time.Minute, 100, src) + + start := time.Now() + tr.matchBlock(context.Background(), 7, 0, time.Unix(1002, 0)) + spent := time.Since(start) + + require.Equal(t, int64(1), src.FetchCount(), + "a permanent answer cost %d requests", src.FetchCount()) + require.Less(t, spent, time.Second, "a permanent answer held the head loop for %s", spent) +} diff --git a/stats/inclusion_run_test.go b/stats/inclusion_run_test.go new file mode 100644 index 0000000..957da32 --- /dev/null +++ b/stats/inclusion_run_test.go @@ -0,0 +1,377 @@ +package stats + +import ( + "context" + "math/big" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + ethtypes "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/rpc" + "github.com/stretchr/testify/require" +) + +// headServer serves eth_subscribe("newHeads") over WebSocket, so a test can +// drive InclusionTracker.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 +// subscription ends the tracking. None of that is reachable from a helper, and +// the defects that reached round seven were all in it. +type headServer struct { + srv *httptest.Server + heads chan *ethtypes.Header + + mu sync.Mutex + conns []net.Conn +} + +type headSubscription struct{ heads chan *ethtypes.Header } + +// NewHeads backs eth_subscribe("newHeads"): go-ethereum takes the subscription +// name from the method name. +func (h *headSubscription) NewHeads(ctx context.Context) (*rpc.Subscription, error) { + notifier, ok := rpc.NotifierFromContext(ctx) + if !ok { + return nil, rpc.ErrNotificationsUnsupported + } + sub := notifier.CreateSubscription() + go func() { + for { + select { + case hdr := <-h.heads: + _ = notifier.Notify(sub.ID, hdr) + case <-sub.Err(): + return + } + } + }() + return sub, nil +} + +func newHeadServer(t *testing.T) *headServer { + t.Helper() + heads := make(chan *ethtypes.Header, 64) + rpcSrv := rpc.NewServer() + require.NoError(t, rpcSrv.RegisterName("eth", &headSubscription{heads: heads})) + + hs := &headServer{heads: heads} + hs.srv = httptest.NewUnstartedServer(rpcSrv.WebsocketHandler([]string{"*"})) + // A WebSocket is a hijacked connection, which CloseClientConnections does + // not reach. Hold the raw conns so a test can drop one the way a node + // restart or a proxy eviction does. + hs.srv.Config.ConnState = func(c net.Conn, state http.ConnState) { + if state != http.StateNew { + return + } + hs.mu.Lock() + defer hs.mu.Unlock() + hs.conns = append(hs.conns, c) + } + hs.srv.Start() + t.Cleanup(hs.srv.Close) + return hs +} + +// endpoint is what a run would be configured with. GetWSEndpoint rewrites the +// scheme and leaves this port alone, so the tracker dials this same server. +func (h *headServer) endpoint() string { + return strings.Replace(h.srv.URL, "ws://", "http://", 1) +} + +func (h *headServer) send(t *testing.T, num uint64) { + t.Helper() + // go-ethereum's decoder requires difficulty and the base fee, so a bare + // header with only a number is rejected by the client before the tracker + // ever sees it. + h.heads <- ðtypes.Header{ + Number: new(big.Int).SetUint64(num), + Difficulty: new(big.Int), + BaseFee: new(big.Int), + Extra: []byte{}, + } +} + +// kill drops the WebSocket, which is what a node restart or a proxy eviction +// looks like to the run. +func (h *headServer) kill() { + h.mu.Lock() + defer h.mu.Unlock() + for _, c := range h.conns { + _ = c.Close() + } + h.conns = nil +} + +// runTracker starts Run against the head server with an injected receipt source +// 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) { + t.Helper() + tr := newInclusionTrackerWithSource( + NewInclusionTracker("test-chain", reapAfter, 1000, true, NewCollector()), src) + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- tr.Run(ctx, hs.endpoint()) }() + t.Cleanup(cancel) + return tr, cancel, done +} + +// Requirements: TOT-001, TOT-002 and TOT-022. +// TestRunReadsAHeadAndResolvesIt is the harness's own proof: a head delivered +// over a real subscription reaches the reader and its block's receipts resolve. +func TestRunReadsAHeadAndResolvesIt(t *testing.T) { + hs := newHeadServer(t) + src := NewMockBlockSource() + tr, cancel, done := runTracker(t, hs, src, time.Minute) + defer cancel() + + tx := loadTx(1, time.Unix(1000, 0)) + tr.Register(context.Background(), tx) + src.SetReceipts(7, blockReceipt{ + Hash: tx.EthTx.Hash(), Status: ethtypes.ReceiptStatusSuccessful, HasStatus: true, + }) + + hs.send(t, 7) + require.Eventually(t, func() bool { return tr.Summary().Included == 1 }, + 3*time.Second, 10*time.Millisecond, + "a head delivered over a real subscription never reached the reader") + + cancel() + <-done +} + +// Requirements: TOT-004 and TOT-008. +// TestATxAcceptedBeforeTheFirstReadIsNotAChainVerdict fails when a transaction +// accepted before the tracker read anything reaps as expired. +// +// The senders and the tracker start together, and the tracker dials and +// preflights before it reads a block, so every run accepts transactions during +// a window in which it is looking at nothing. Those transactions land in blocks +// the run never opened. Calling that expired is a claim about the chain drawn +// from a period the run did not observe, and on the shipped profiles it happens +// on every run. +func TestATxAcceptedBeforeTheFirstReadIsNotAChainVerdict(t *testing.T) { + hs := newHeadServer(t) + src := NewMockBlockSource() + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tr, cancel, done := runTracker(t, hs, src, 30*time.Millisecond) + defer cancel() + + // Accepted while the tracker is still starting: its block is already behind + // the first head the run will ever see. + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + + // The run starts looking from here. Block 99, which held the transaction, is + // never read. + for h := uint64(100); h <= 103; h++ { + src.SetReceipts(h) + hs.send(t, h) + } + + require.Eventually(t, func() bool { + g := tr.collector.GetOperationStats()[key] + return g.Expired+g.StatusUnavailable == 1 + }, 5*time.Second, 10*time.Millisecond, "the transaction never reached a terminal state") + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.StatusUnavailable, + "a transaction accepted before the run read anything was blamed on the chain") + require.Zero(t, got.Expired) + + cancel() + <-done +} + +// Requirements: TOT-004. +// TestADeadSubscriptionStopsTheRunSayingAnything fails when a run whose head +// stream dies keeps producing chain verdicts, or reports the failure as +// something else. +// +// Nothing after the stream dies is read, so nothing after it is evidence about +// the chain. The run continues, because the senders are still working and a +// read-only observer must not fail them. +func TestADeadSubscriptionStopsTheRunSayingAnything(t *testing.T) { + hs := newHeadServer(t) + src := NewMockBlockSource() + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + tr, cancel, done := runTracker(t, hs, src, 30*time.Millisecond) + defer cancel() + + src.SetReceipts(7) + hs.send(t, 7) + require.Eventually(t, func() bool { return tr.Summary().Included == 0 && headsSeen(tr) >= 7 }, + 5*time.Second, 10*time.Millisecond, "the first head never landed") + + // Accepted while the run is healthy, then the socket drops. + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + hs.kill() + + // The run does not fail: a read-only observer must not kill the senders. + select { + case err := <-done: + require.NoError(t, err, "a dropped subscription failed the whole run") + case <-time.After(5 * time.Second): + t.Fatal("Run did not return after the head stream died") + } + + time.Sleep(60 * time.Millisecond) // past reapAfter, so the tx is due + tr.reap(context.Background()) + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.Expired+got.StatusUnavailable, + "the transaction never reached a terminal state") + require.Equal(t, uint64(1), got.StatusUnavailable, + "a run whose head stream died still claimed the chain left a tx out") + require.Zero(t, got.Expired) +} + +// headsSeen reports the highest head the tracker has taken off the wire. +func headsSeen(tr *InclusionTracker) uint64 { + for s := range tr.state.Lock() { + return s.headsReceived + } + panic("unreachable") +} + +// Requirements: TOT-003 and TOT-004. +// TestADeadSubscriptionSettlesEverythingInFlight fails when a run whose head +// stream died leaves transactions to pile up. +// +// The reap loop ends with the head loop, so nothing reaps afterwards. Without a +// drain the registry fills from senders that are still working and every later +// transaction reports dropped_at_cap, which tells an operator to raise a cap +// when the real cause was a dead subscription. +func TestADeadSubscriptionSettlesEverythingInFlight(t *testing.T) { + hs := newHeadServer(t) + src := NewMockBlockSource() + key := OperationKey{Scenario: "erc20", Operation: "erc20_transfer"} + + // A reap window longer than the test: nothing may depend on a reap running. + tr, cancel, done := runTracker(t, hs, src, time.Hour) + defer cancel() + + src.SetReceipts(7) + hs.send(t, 7) + require.Eventually(t, func() bool { return headsSeen(tr) >= 7 }, + 5*time.Second, 10*time.Millisecond, "the first head never landed") + + inFlight := loadTx(1, time.Unix(1000, 0)) + inFlight.Scenario.Name, inFlight.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), inFlight) + + hs.kill() + require.NoError(t, <-done) + + // A sender that has not noticed yet keeps handing transactions over. + after := loadTx(2, time.Unix(1000, 0)) + after.Scenario.Name, after.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), after) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(2), got.StatusUnavailable, + "a dead subscription left transactions unsettled") + require.Equal(t, got.StatusUnavailable, tr.Summary().StatusUnavailable, + "the drain reached the metric ledger and not the closing log line, so "+ + "one transaction is counted under two names") + require.Zero(t, got.DroppedAtCap, "the cause was reported as a full registry") + require.Zero(t, tr.Summary().InflightAtShutdown) +} + +// Requirements: TOT-004 and TOT-008. +// TestTheWatermarksKeepExpiredReachable drives two transactions through Run: one +// accepted before the run read anything, one after. +// +// One transaction cannot tell these apart. Both mechanisms that make expired +// reachable — the first-read watermark and marking a head resolved — can be +// disabled without a single-transaction test noticing, because every verdict +// moves the same way. Then a run reports status_unavailable for everything it +// ever reaps, which is the over-reporting direction, and it is silent. +func TestTheWatermarksKeepExpiredReachable(t *testing.T) { + hs := newHeadServer(t) + src := NewMockBlockSource() + tr, cancel, done := runTracker(t, hs, src, 40*time.Millisecond) + defer cancel() + + before := OperationKey{Scenario: "s", Operation: "before"} + after := OperationKey{Scenario: "s", Operation: "after"} + + // Accepted while the run is still starting. Its block is behind the first + // head the run will ever see. + early := loadTx(1, time.Unix(1000, 0)) + early.Scenario.Name, early.Scenario.Operation = before.Scenario, before.Operation + tr.Register(context.Background(), early) + + // The run starts looking here. + src.SetReceipts(100) + hs.send(t, 100) + require.Eventually(t, func() bool { return headsSeen(tr) >= 100 }, + 5*time.Second, 5*time.Millisecond, "the first head never landed") + + // Accepted with the run underway, and never included. + late := loadTx(2, time.Unix(1000, 0)) + late.Scenario.Name, late.Scenario.Operation = after.Scenario, after.Operation + tr.Register(context.Background(), late) + + for h := uint64(101); h <= 104; h++ { + src.SetReceipts(h) + hs.send(t, h) + } + require.Eventually(t, func() bool { return headsSeen(tr) >= 104 }, + 5*time.Second, 5*time.Millisecond, "the later heads never landed") + + time.Sleep(80 * time.Millisecond) + tr.reap(context.Background()) + + stats := tr.collector.GetOperationStats() + require.Equal(t, uint64(1), stats[before].StatusUnavailable, + "a transaction accepted before the run read anything was blamed on the chain") + require.Equal(t, uint64(1), stats[after].Expired, + "a run that read every head it saw could not say the chain left a tx out; "+ + "expired is unreachable") + + cancel() + <-done +} + +// Requirements: TOT-004. +// TestARunThatReadNothingCannotSpeakForTheChain fails when a run whose +// subscription is up but has delivered no head calls a transaction expired. +// +// It is reachable: the socket connects, the chain is quiet or the node is not +// publishing, and reapAfter elapses. Every other watermark reads as healthy — +// nothing failed, no head was missed, and the run is trivially caught up with +// the nothing it has seen. +func TestARunThatReadNothingCannotSpeakForTheChain(t *testing.T) { + hs := newHeadServer(t) + tr, cancel, done := runTracker(t, hs, NewMockBlockSource(), 30*time.Millisecond) + defer cancel() + + key := OperationKey{Scenario: "s", Operation: "o"} + tx := loadTx(1, time.Unix(1000, 0)) + tx.Scenario.Name, tx.Scenario.Operation = key.Scenario, key.Operation + tr.Register(context.Background(), tx) + + // No head is ever sent. + time.Sleep(60 * time.Millisecond) + tr.reap(context.Background()) + + got := tr.collector.GetOperationStats()[key] + require.Equal(t, uint64(1), got.StatusUnavailable, + "a run that read no block at all claimed the chain left a tx out") + require.Zero(t, got.Expired) + + cancel() + <-done +} diff --git a/stats/inclusion_tracker.go b/stats/inclusion_tracker.go index f5ac769..6e21400 100644 --- a/stats/inclusion_tracker.go +++ b/stats/inclusion_tracker.go @@ -2,14 +2,18 @@ package stats import ( "context" + "errors" "fmt" "log" - "math/big" + "net" + "strings" "time" + "github.com/ethereum/go-ethereum" "github.com/ethereum/go-ethereum/common" ethtypes "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/ethclient" + "github.com/ethereum/go-ethereum/rpc" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/metric" @@ -18,34 +22,153 @@ import ( "github.com/sei-protocol/sei-load/utils/scope" ) -// blockSource yields the tx hashes of a single block by number. Consumer-side -// interface so tests can drive matching without a live chain. -type blockSource interface { - BlockTxHashes(ctx context.Context, n uint64) ([]common.Hash, error) +// blockReceipt is what the tracker reads out of one transaction's receipt: the +// hash it joins against the registry, and the status that says what execution +// did. +// +// This package's own type rather than *ethtypes.Receipt. A receipt carries a +// dozen more fields the tracker has no business reading, and a test supplies a +// pair without constructing one. +// +// HasStatus is false for a receipt that carries a post-state root instead of a +// status, which is a receipt whose execution result cannot be read. Status is +// meaningless when it is false. +// +// It detects that one shape and no other. go-ethereum makes both the status and +// the root optional when it decodes, so a receipt carrying neither arrives here +// indistinguishable from one reporting a failure, and this run reads it as a +// failure. Sei always writes a status, so the undetectable shape does not come +// from the chain this tracks. +type blockReceipt struct { + Hash common.Hash + Status uint64 + HasStatus bool } -// ethBlockSource is the production blockSource backed by an ethclient. -type ethBlockSource struct{ client *ethclient.Client } +// receiptSource yields one block's receipts by number. Consumer-side interface +// so tests can drive matching without a live chain. One call per block, +// whatever the block's transaction count. +// +// One call is what this process issues. It is not what the call costs the node +// that answers it. seid resolves each receipt's index by walking the whole +// block and recovering every sender again, so its work grows with the square of +// the block's transaction count. Upstream go-ethereum does not do this; the +// cost belongs to the implementation being measured, not to the method. Point +// the tracker at a node that takes no send load, and see +// config.ReceiptEndpoint. +type receiptSource interface { + // BlockReceipts returns the block's receipts, and how many the endpoint + // sent as null. A null carries no hash, so those transactions are a hole + // even though the read succeeded. + BlockReceipts(ctx context.Context, n uint64) (receipts []blockReceipt, nulls int, err error) +} + +// ethReceiptSource is the production receiptSource backed by an ethclient. +type ethReceiptSource struct{ client *ethclient.Client } -func (s ethBlockSource) BlockTxHashes(ctx context.Context, n uint64) ([]common.Hash, error) { - block, err := s.client.BlockByNumber(ctx, new(big.Int).SetUint64(n)) +func (s ethReceiptSource) BlockReceipts(ctx context.Context, n uint64) ([]blockReceipt, int, error) { + // A block height never approaches MaxInt64, where the conversion would land + // on rpc.BlockNumber's negative sentinels. + number := rpc.BlockNumberOrHashWithNumber(rpc.BlockNumber(n)) + receipts, err := s.client.BlockReceipts(ctx, number) if err != nil { - return nil, err + return nil, 0, err } - txs := block.Transactions() - hashes := make([]common.Hash, len(txs)) - for i, tx := range txs { - hashes[i] = tx.Hash() + out, nulls := narrowReceipts(receipts) + return out, nulls, nil +} + +// narrowReceipts keeps what the tracker reads and drops the rest. +// +// An endpoint decides what it puts in this array. A nil element would panic the +// head loop, which ends the run and loses every result it gathered, so drop it +// rather than trust the shape. +// +// It reports how many elements were nil rather than refusing the whole array. +// The run cannot see what a nil element was going to say, so the block is a hole +// for the transactions it did not name. It is not a hole for the ones it did: +// discarding a receipt the endpoint actually sent would throw away a known +// outcome to describe an unknown one. +func narrowReceipts(receipts []*ethtypes.Receipt) (out []blockReceipt, nulls int) { + out = make([]blockReceipt, 0, len(receipts)) + for _, r := range receipts { + if r == nil { + nulls++ + continue + } + out = append(out, blockReceipt{ + Hash: r.TxHash, + Status: r.Status, + // A root alone means no status. A root beside a status means the + // status is there and is the more specific of the two, so it wins: + // reading it as unavailable would report a whole run as unreadable + // against any endpoint that sends both. + HasStatus: len(r.PostState) == 0 || r.Status != 0, + }) } - return hashes, nil + return out, nulls +} + +// headArrival is one head and the moment it reached this process. +type headArrival struct { + num uint64 + gasUsed uint64 + arrival time.Time } type entry struct { tx *types.LoadTx registeredAt time.Time + // blindHeightsAtRegistration is inclusionState.blindHeights at the moment + // this tx was registered. A reap compares it against the count now: a + // higher count means a height went unread while this tx was in flight, so + // the run cannot say the chain left it out. See reap. + blindHeightsAtRegistration uint64 } type inclusionState struct { + // blindHeights counts heights the run could not read, whether a read failed + // or a head never arrived. It only ever grows, so an entry's terminal state + // follows from comparing this against the value it recorded at registration. + // + // The count carries no height, so it marks a tx in flight across any unread + // height rather than a tx in the block that went unread. The tracker does + // not know which registered txs a block it never read was holding, and + // claiming to would be the invention this state exists to avoid. + blindHeights uint64 + // statusUnavailable counts reaped txs that spanned a failed read, plus + // duplicate registrations. It is the part the run cannot attribute to the + // chain. + statusUnavailable uint64 + // headsReceived and headsResolved are the highest head this process has taken + // off the wire and the highest whose block it has finished reading. They + // differ whenever the tracker is behind the chain. + // + // A reap needs them. Every hole this file records is a read that failed or a + // head that never came, and none of that covers the case where every read + // succeeds and the tracker is simply late: the block carrying a transaction + // is read after the reap already evicted it. Without this the run reports + // expired, which is a claim about the chain, for a block it had in hand and + // had not opened yet. + headsReceived uint64 + headsResolved uint64 + // firstReadAt is when the run first read a block. Before it the tracker is + // dialling and probing while the senders are already sending, so every run + // accepts transactions during a window in which it observes nothing. + // + // A reap compares it against the transaction's own registration. A + // transaction accepted before the run started looking may have landed in a + // block the run never opened, so the chain cannot be blamed for it. + firstReadAt time.Time + // trackingStopped means the head stream ended and no later height will be + // read at all. Register reads it; a reap does not need to, because + // stopTracking settles everything in flight and Register admits nothing + // after it, so no reap can find a transaction from that period. + trackingStopped bool + // duplicates counts registrations of a hash already in flight. The registry + // holds one slot per hash, so the second one has no place to go and the run + // can say nothing about it. + duplicates uint64 inflight map[common.Hash]*entry included uint64 expired uint64 @@ -53,9 +176,11 @@ type inclusionState struct { inflightAtShutdown uint64 } -// InclusionTracker matches arriving blocks against in-flight txs to stamp -// InclusionTime. Conservation: registered == included + expired + -// inflight_at_shutdown, and registered ⊆ succeeded (see sender/doc.go). +// InclusionTracker matches arriving blocks against in-flight txs, stamps +// InclusionTime, and resolves each one to a terminal [Outcome]. +// +// sender/doc.go owns the conservation identity these outcomes satisfy, and it is +// not restated here. type InclusionTracker struct { seiChainID string reapAfter time.Duration @@ -65,8 +190,19 @@ type InclusionTracker struct { // enqueue time, so arrival-IntendedSendTime would be an enqueue→inclusion // latency that must not be mixed into the histogram. openLoop bool - source blockSource + source receiptSource state utils.Mutex[*inclusionState] + + // collector receives every terminal outcome. The reference runs one way: the + // tracker may take the collector's lock, and the collector must never take + // this tracker's state lock. Nothing takes both today, and this is where the + // rule is written down so nothing starts. A third holder exists: the + // inclusion_inflight gauge callback in metrics.go takes the state lock on + // the metric SDK's collection goroutine. It takes no collector lock, and + // the rule above is why it must not start. + // + // Never nil. NewInclusionTracker takes it because every run has one. + collector *Collector } // defaultMaxInflight bounds the registry when the caller passes a non-positive @@ -83,7 +219,7 @@ const defaultInclusionReapAfter = 30 * time.Second // reaps un-included txs after reapAfter. openLoop gates the inclusion_latency // sample (included/expired counts are tracked in both models). The block source // is the production ethclient impl; tests inject via newInclusionTrackerWithSource. -func NewInclusionTracker(seiChainID string, reapAfter time.Duration, maxInflight int, openLoop bool) *InclusionTracker { +func NewInclusionTracker(seiChainID string, reapAfter time.Duration, maxInflight int, openLoop bool, collector *Collector) *InclusionTracker { if maxInflight <= 0 { maxInflight = defaultMaxInflight } @@ -95,6 +231,7 @@ func NewInclusionTracker(seiChainID string, reapAfter time.Duration, maxInflight reapAfter: reapAfter, maxInflight: maxInflight, openLoop: openLoop, + collector: collector, state: utils.NewMutex(&inclusionState{ inflight: make(map[common.Hash]*entry), }), @@ -103,7 +240,7 @@ func NewInclusionTracker(seiChainID string, reapAfter time.Duration, maxInflight return t } -func newInclusionTrackerWithSource(t *InclusionTracker, source blockSource) *InclusionTracker { +func newInclusionTrackerWithSource(t *InclusionTracker, source receiptSource) *InclusionTracker { t.source = source return t } @@ -111,53 +248,102 @@ func newInclusionTrackerWithSource(t *InclusionTracker, source blockSource) *Inc // Register hands ownership of tx's InclusionTime to the tracker. Caller must // invoke it only for successful sends, at send-completion, so // registered ⊆ succeeded holds. At cap the tx is dropped and counted. -func (t *InclusionTracker) Register(tx *types.LoadTx) { +func (t *InclusionTracker) Register(ctx context.Context, tx *types.LoadTx) { hash := tx.EthTx.Hash() - var droppedAtCap bool + var outcome Outcome for s := range t.state.Lock() { + // Nothing further is read, so holding this would only fill the registry + // and report dropped_at_cap, which names the wrong cause. + if s.trackingStopped { + s.statusUnavailable++ + outcome = OutcomeStatusUnavailable + break + } // Cap check and insert share one critical section: race-free admission. if len(s.inflight) >= t.maxInflight { s.droppedAtCap++ - droppedAtCap = true + outcome = OutcomeDroppedAtCap + break + } + // One map slot per hash, so a second registration of the same hash + // would overwrite the first and the run would report one terminal state + // for two accepted transactions. A resend after a send timeout produces + // exactly that. Count the duplicate rather than lose it. + // + // status_unavailable is the closest of the states the spec defines, and + // it is not a clean fit: the run could read this transaction's status + // perfectly well, it just has nowhere to put a second copy. The state + // exists for a status the run could not read. Naming this collision + // properly needs a state the spec does not have. + if _, dup := s.inflight[hash]; dup { + s.duplicates++ + outcome = OutcomeStatusUnavailable break } - s.inflight[hash] = &entry{tx: tx, registeredAt: time.Now()} + s.inflight[hash] = &entry{tx: tx, registeredAt: time.Now(), blindHeightsAtRegistration: s.blindHeights} } - if droppedAtCap { - t.recordOutcome("dropped_at_cap", tx.Scenario) + if outcome != outcomeUnset { + t.report(ctx, outcome, 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) { - inclusionOutcome.Add(context.Background(), 1, metric.WithAttributes( +// meterOutcome emits one terminal outcome to the inclusion_outcome counter. +// 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. +// +// 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) meterOutcome(ctx context.Context, outcome Outcome, scenario *types.TxScenario) { + inclusionOutcome.Add(ctx, 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()), )) } -// Run subscribes to new heads and matches each arriving block once. -func (t *InclusionTracker) Run(ctx context.Context, firstEndpoint string) error { - wsEndpoint := utils.GetWSEndpoint(firstEndpoint) +// Run subscribes to new heads on endpoint and reads each arriving block's +// receipts from the same endpoint. +// +// One node for both, and that is a correctness rule rather than a convenience. +// A head notification carries the raw committed height; a status read resolves +// through a watermark that also waits on the receipt store. The two disagree +// even on one node, and taking them from separate nodes adds peer lag on top of +// that, which a node inside its readiness threshold can carry for minutes +// without reporting unhealthy. Every height would then arrive before the reading +// node held it. +// +// The tracker reads the height it just received as a head, waiting in place if +// the node has not finished writing it. It never asks for an older height, so +// the serving node's receipt retention does not bound it. A change that reaches +// further back does. +func (t *InclusionTracker) Run(ctx context.Context, endpoint string) error { + wsEndpoint := utils.GetWSEndpoint(endpoint) if t.source == nil { - client, err := ethclient.Dial(firstEndpoint) + client, err := ethclient.Dial(endpoint) if err != nil { - return fmt.Errorf("inclusion tracker: dial %s: %w", firstEndpoint, err) + return fmt.Errorf("inclusion tracker: dial %s: %w", endpoint, err) } defer client.Close() - t.source = ethBlockSource{client: client} + t.source = ethReceiptSource{client: client} + if err := t.preflight(ctx, endpoint); err != nil { + return err + } } - return scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { + err := scope.Run(ctx, func(ctx context.Context, s scope.Scope) error { client, err := ethclient.Dial(wsEndpoint) if err != nil { return fmt.Errorf("inclusion tracker: connect WebSocket %s: %w", wsEndpoint, err) } - headers := make(chan *ethtypes.Header) + defer client.Close() + // Buffered: head processing is serial, and a read that takes its whole + // budget must not cost a head. go-ethereum buffers far more than this + // on its own side and fails the subscription rather than dropping + // quietly, so this is headroom for the handoff, not the real bound. + headers := make(chan *ethtypes.Header, 32) sub, err := client.SubscribeNewHead(ctx, headers) if err != nil { return fmt.Errorf("inclusion tracker: subscribe new heads: %w", err) @@ -168,63 +354,307 @@ func (t *InclusionTracker) Run(ctx context.Context, firstEndpoint string) error if err != nil { return err } - return subErr + // Not the run's error. See the sentinel's own doc. + return fmt.Errorf("%w: %v", errHeadStreamEnded, subErr) }) s.Spawn(func() error { return t.reapLoop(ctx) }) + // One goroutine takes heads off the wire and stamps each with the moment + // it arrived; another reads their blocks. Stamping at the point of + // processing instead would fold this tracker's own backlog into + // InclusionTime, so a run whose reads outrun the block interval would + // report its own lateness as the chain's inclusion latency. + arrivals := make(chan headArrival, cap(headers)) + s.SpawnBg(func() error { return t.pumpHeads(ctx, headers, arrivals) }) + var lastSeen uint64 // 0 = unset; first head seeds it (no backfill). for ctx.Err() == nil { - header, err := utils.Recv(ctx, headers) + a, err := utils.Recv(ctx, arrivals) if err != nil { return err } - lastSeen = t.processHead(ctx, header.Number.Uint64(), time.Now(), lastSeen) + // How long this head waited to be read. It is the tracker's own lag, + // and it is the number that separates a chain that took nothing from + // a run that could not keep up: both report un-included txs, and only + // this one says which. + inclusionHeadLag.Record(ctx, time.Since(a.arrival).Seconds(), + metric.WithAttributes(attribute.String("chain_id", t.seiChainID))) + lastSeen = t.processHead(ctx, a.num, a.gasUsed, a.arrival, lastSeen) + t.noteHeadResolved(a.num) } return ctx.Err() }) + // A head stream that ends mid-run stops the tracking, not the run. The + // senders, the generator and the report are all still doing their job, and + // failing the whole run over a read-only observer's transport would turn a + // good run red. Every later height is unread, so nothing after this can reap + // as a verdict about the chain. + if errors.Is(err, errHeadStreamEnded) { + log.Printf("⚠️ inclusion tracker: %v. Tracking stops here; "+ + "outcomes after this point are status_unavailable", err) + t.stopTracking(ctx) + return nil + } + return err } // processHead handles one arriving head: counts any gap (no backfill), matches // the block, and returns the new lastSeen. lastSeen==0 seeds on the first head. -func (t *InclusionTracker) processHead(ctx context.Context, num uint64, arrival time.Time, lastSeen uint64) uint64 { +func (t *InclusionTracker) processHead(ctx context.Context, num, gasUsed uint64, arrival time.Time, lastSeen uint64) uint64 { if lastSeen != 0 && num <= lastSeen { return lastSeen // duplicate or out-of-order head: no re-fetch, no spurious gap. } if lastSeen != 0 && num > lastSeen+1 { inclusionBlockGaps.Add(ctx, int64(num-lastSeen-1), metric.WithAttributes( attribute.String("chain_id", t.seiChainID))) + // A height nothing read is a hole, however it went missing. Leaving the + // gap uncounted let a tx in one of those blocks reap as expired, which + // is a claim about the chain the run has no grounds for. + // + // One record for the whole gap. The reap only asks whether the count + // rose, so recording each height would change nothing about attribution + // and would emit a log line per height. A long gap would then push the + // run's own summary out of any bounded log tail, which is the only + // diagnostic a failed run carries. + t.recordBlindGap(ctx, lastSeen+1, num-1) } - t.matchBlock(ctx, num, arrival) + t.matchBlock(ctx, num, gasUsed, arrival) return num } -// matchBlock fetches block num once and stamps every in-flight tx it includes -// with the header-arrival time. -func (t *InclusionTracker) matchBlock(ctx context.Context, num uint64, arrival time.Time) { - // Explicit per-iteration cancel (not deferred-in-loop): bound the fetch. - fetchCtx, cancel := context.WithTimeout(ctx, 10*time.Second) - hashes, err := t.source.BlockTxHashes(fetchCtx, num) - cancel() +// pumpHeads takes heads off the wire and stamps each with the moment it arrived. +// +// It is its own step so the stamp cannot drift to the moment of processing. A +// tracker whose reads outrun the block interval builds a backlog, and a stamp +// taken at processing would carry that backlog into InclusionTime and report +// this run's own lateness as the chain's inclusion latency. +func (t *InclusionTracker) pumpHeads( + ctx context.Context, headers <-chan *ethtypes.Header, arrivals chan<- headArrival, +) error { + for ctx.Err() == nil { + header, err := utils.Recv(ctx, headers) + if err != nil { + return err + } + a := headArrival{ + num: header.Number.Uint64(), + gasUsed: header.GasUsed, + arrival: time.Now(), + } + t.noteHeadReceived(a.num) + if err := utils.Send(ctx, arrivals, a); err != nil { + return err + } + } + return ctx.Err() +} + +// noteHeadReceived records that a head reached this process, before anything +// reads its block. +func (t *InclusionTracker) noteHeadReceived(num uint64) { + for s := range t.state.Lock() { + if num > s.headsReceived { + s.headsReceived = num + } + } +} + +// noteHeadResolved records that a head's block has been read. +func (t *InclusionTracker) noteHeadResolved(num uint64) { + for s := range t.state.Lock() { + if num > s.headsResolved { + s.headsResolved = num + } + } +} + +// stopTracking records that no further height will be read, and settles +// everything still in flight. +// +// The draining is the point. The reap loop ends with the head loop, so nothing +// reaps after this: without a drain the registry keeps filling from senders that +// are still working, and every later transaction reports dropped_at_cap. An +// operator reads that as "raise the cap" rather than "the subscription died". +func (t *InclusionTracker) stopTracking(ctx context.Context) { + var stranded []resolvedOutcome + for s := range t.state.Lock() { + s.trackingStopped = true + for h, e := range s.inflight { + delete(s.inflight, h) + s.statusUnavailable++ + stranded = append(stranded, resolvedOutcome{ + outcome: OutcomeStatusUnavailable, scenario: e.tx.Scenario, + }) + } + } + for _, r := range stranded { + t.report(ctx, r.outcome, r.scenario) + } + inclusionBlockFetchErrors.Add(ctx, 1, metric.WithAttributes( + attribute.String("chain_id", t.seiChainID), + attribute.String("reason", reasonTrackingStopped))) +} + +// preflight proves the endpoint serves eth_getBlockReceipts before the run +// spends gas. Without it the run completes, reports every transaction +// status_unavailable, and exits zero — a blind run that reads as a chain that +// accepted nothing. +// +// It asks for height 0. sei-chain answers that from a constant, ahead of its +// watermark and its receipt store, so the probe tests the endpoint and says +// nothing about the node's sync state: a node merely behind still passes. +// Nothing but a broken endpoint fails it, which is why any error that survives +// every attempt refuses the run and no error classifier is needed here. +func (t *InclusionTracker) preflight(ctx context.Context, endpoint string) error { + // Several attempts, because one verdict stands for the whole run. A single + // reset or timeout from a node that is merely busy must not end a run before + // it starts. + var err error + for attempt := range preflightAttempts { + if attempt > 0 { + if _, waitErr := utils.Recv(ctx, time.After(preflightBackoff)); waitErr != nil { + return nil + } + } + probeCtx, cancel := context.WithTimeout(ctx, preflightTimeout) + _, _, err = t.source.BlockReceipts(probeCtx, 0) + cancel() + if err == nil { + return nil + } + log.Printf("inclusion tracker: preflight against %s: %v", endpoint, err) + } + return fmt.Errorf( + "inclusion tracker: %w: %s failed %d eth_getBlockReceipts probes (%v). "+ + "receiptEndpoint must name the EVM HTTP port of a node in fullNode or "+ + "archive mode: validator and seed modes serve no EVM HTTP, and the "+ + "metrics port, the Cosmos RPC port and an ingress error page all answer "+ + "without serving the method", + ErrEndpointUnusable, endpoint, preflightAttempts, err) +} + +// readReceipts reads one block's receipts, waiting out a height the node has +// not yet made readable, and reports how long it waited. +// +// A not-found is the node's watermark trailing the head it just published, not +// a missing block. Retrying in place costs a phase shift and no backlog: the +// wait for height N ends when N becomes readable, which on a healthy node is +// long before head N+1 arrives. +func (t *InclusionTracker) readReceipts( + ctx context.Context, num uint64, +) (receipts []blockReceipt, nulls int, waited time.Duration, err error) { + readCtx, cancel := context.WithTimeout(ctx, readTimeout) + defer cancel() + start := time.Now() + for { + receipts, nulls, err = t.source.BlockReceipts(readCtx, num) + if err == nil || fetchFailureReason(err) != reasonNotFound { + return receipts, nulls, time.Since(start), err + } + if _, waitErr := utils.Recv(readCtx, time.After(notFoundBackoff)); waitErr != nil { + return nil, 0, time.Since(start), err + } + } +} + +// matchBlock reads block num's receipts and stamps every in-flight tx the block +// includes with arrival, the moment the head reached this process. +// +// seid publishes a head from App.Commit, and eth_getBlockReceipts resolves +// through a watermark that also waits on the receipt store's async writer, so a +// height can answer null for the moment between the two. readReceipts waits that +// out; anything readTimeout does not resolve is a hole. +func (t *InclusionTracker) matchBlock(ctx context.Context, num, gasUsed uint64, arrival time.Time) { + receipts, nulls, waited, err := t.readReceipts(ctx, num) if err != nil { - // No retry (avoids piling RPC onto a struggling SUT): the block's txs go - // unmatched and reap as expired. Surfaced so the undercount is visible. - log.Printf("inclusion tracker: fetch block %d: %v", num, err) - inclusionBlockFetchErrors.Add(ctx, 1, metric.WithAttributes( - attribute.String("chain_id", t.seiChainID))) + // The block goes unmatched, and blindHeights records that this run has + // a hole, so a tx in flight across it reaps as status_unavailable + // rather than as a verdict about the chain. + // + // An empty slice is not this case. A node that holds a block carrying + // no EVM transaction answers it truthfully with an empty array, and a + // node that holds nothing answers null, which arrives here as + // ethereum.NotFound. Reading an empty array as a hole would mark every + // idle block, and a chain that stopped accepting work produces nothing + // but idle blocks, which is the one run where expired is the answer. + t.recordBlindFetch(ctx, num, fetchFailureReason(err), err) return } - matched := make([]inclusionSample, 0, len(hashes)) + if waited > 0 { + // How long the node trailed its own head on this height. The + // distribution is the one signal that says whether the serving node + // keeps up with the stream it publishes. + inclusionReadWait.Record(ctx, waited.Seconds(), + metric.WithAttributes(attribute.String("chain_id", t.seiChainID))) + } + if nulls > 0 { + // The read succeeded and part of it is unreadable. Match what arrived, + // and record the hole so a transaction the missing part would have named + // is not blamed on the chain. + t.recordBlindFetch(ctx, num, reasonNullReceipt, nil) + } + if len(receipts) == 0 && gasUsed > 0 { + // An empty array usually means the block carried no EVM transaction, + // and it is the answer for most blocks on an idle chain. It has one + // other cause: sei-chain drops a receipt its store cannot find and + // compacts it out of the array, so a block whose receipts went missing + // answers the same way. + // + // This is the gas on the head from the newHeads subscription, which + // sei-chain sums over every transaction's consensus result rather than + // over receipts. That independence is the whole reason it can witness a + // missing receipt. Note which header: eth_getBlockByNumber sums + // receipt.GasUsed instead, which would be circular here, and + // sei-chain's evmrpc/subscribe.go carries a TODO to make newHeads exact + // the same way. If that lands, this counter stops being able to fire. + // + // Counted and not acted on. The gas covers non-EVM transactions and an + // EVM transaction that failed the ante handler, both of which burn gas + // and produce no receipt legitimately, so treating this as a hole would + // invent holes, and holes reported too often make expired unreachable. + inclusionEmptyWithGas.Add(ctx, 1, metric.WithAttributes( + attribute.String("chain_id", t.seiChainID))) + } + matched := make([]inclusionSample, 0, len(receipts)) + resolved := make([]resolvedOutcome, 0, len(receipts)) for s := range t.state.Lock() { - for _, h := range hashes { - e, ok := s.inflight[h] + if s.firstReadAt.IsZero() { + // The run is looking from here. Anything accepted earlier could be + // in a block it never opened. + s.firstReadAt = time.Now() + } + for _, r := range receipts { + e, ok := s.inflight[r.Hash] if !ok { continue } // Single writer of InclusionTime, under the lock; first observation // wins (delete-on-touch) — see reorg note in sender/doc.go. e.tx.InclusionTime = arrival - delete(s.inflight, h) - s.included++ + delete(s.inflight, r.Hash) + // A receipt carries one status bit, so the outcome names what the run + // observed. Every failure cause shares the failed status. + outcome := OutcomeStatusUnavailable + switch { + case !r.HasStatus: + // A receipt with a post-state root instead of a status says the + // tx executed and does not say how it ended. Reading that as a + // failure would invent a chain result. + // + // It counts here and not in included, so the summary's terms + // stay disjoint and an operator can add them up. + s.statusUnavailable++ + case r.Status == ethtypes.ReceiptStatusSuccessful: + outcome = OutcomeCommitted + s.included++ + default: + outcome = OutcomeReverted + s.included++ + } + resolved = append(resolved, resolvedOutcome{ + outcome: outcome, + scenario: e.tx.Scenario, + }) // Open-loop only: IntendedSendTime is a true arrival schedule there, so // arrival-IntendedSendTime is a real inclusion latency. In closed-loop // it is enqueue time and the sample is omitted. A zero IntendedSendTime @@ -238,6 +668,9 @@ func (t *InclusionTracker) matchBlock(ctx context.Context, num uint64, arrival t } } } + // Both loops run outside the registry lock. The sender blocks on that lock at + // every send completion, so work held under it lands in the latency this + // package reports. for _, m := range matched { inclusionLatency.Record(ctx, m.latency, metric.WithAttributes( attribute.String("chain_id", t.seiChainID), @@ -245,6 +678,29 @@ func (t *InclusionTracker) matchBlock(ctx context.Context, num uint64, arrival t attribute.String("operation", m.scenario.Operation), )) } + for _, r := range resolved { + t.report(ctx, r.outcome, r.scenario) + } +} + +// resolvedOutcome is one matched tx and what its receipt said, carried out of +// the registry lock so the report lands outside the critical section. +type resolvedOutcome struct { + outcome Outcome + scenario *types.TxScenario +} + +// report sends one terminal outcome to both ledgers: the inclusion_outcome +// metric an operator watches live, and the collector's per-operation counts. +// The run report does not read those counts yet. +// +// Call it outside the registry lock. +func (t *InclusionTracker) report(ctx context.Context, outcome Outcome, scenario *types.TxScenario) { + t.meterOutcome(ctx, outcome, scenario) + t.collector.RecordOutcome(OperationKey{ + Scenario: scenario.Name, + Operation: scenario.Operation, + }, outcome) } // inclusionSample is one matched tx, carried out of the registry lock so its @@ -265,7 +721,7 @@ func (t *InclusionTracker) reapLoop(ctx context.Context) error { log.Printf("inclusion tracker: reap loop: %v", err) continue } - t.reap() + t.reap(ctx) } return ctx.Err() } @@ -273,31 +729,192 @@ func (t *InclusionTracker) reapLoop(ctx context.Context) error { // reap evicts txs in-flight longer than reapAfter as expired. Delete-on-touch // under the lock races safely against matchBlock: whoever holds the lock first // wins, no double count. -func (t *InclusionTracker) reap() { +func (t *InclusionTracker) reap(ctx context.Context) { cutoff := time.Now().Add(-t.reapAfter) - var expired []*types.TxScenario + var expired []resolvedOutcome for s := range t.state.Lock() { for h, e := range s.inflight { if e.registeredAt.After(cutoff) { continue } delete(s.inflight, h) - s.expired++ - expired = append(expired, e.tx.Scenario) + // A fetch failed while this tx was in flight, so the block that + // would have carried it was never read. The chain may well have + // included it. Expired would report a chain problem where the truth + // is a measurement problem, which is what Outcome's two states are + // for. + // + // The registry counters split the same way the outcomes do. They + // reach the operator through InclusionSummary and the closing log + // line, so letting them stay merged would print one transaction as + // expired on the surface an operator reads first and as + // status_unavailable on the one they read second. + outcome := OutcomeExpired + switch { + case s.blindHeights > e.blindHeightsAtRegistration, + s.headsResolved < s.headsReceived, + s.firstReadAt.IsZero(), + s.firstReadAt.After(e.registeredAt): + // Either a height went unread, or the run is holding one it has + // not opened yet. A transaction may be sitting in either, so the + // run cannot say the chain left it out. + outcome = OutcomeStatusUnavailable + s.statusUnavailable++ + default: + s.expired++ + } + expired = append(expired, resolvedOutcome{outcome: outcome, scenario: e.tx.Scenario}) } } // 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) + for _, r := range expired { + t.report(ctx, r.outcome, r.scenario) + } +} + +// recordBlindGap marks that the run never saw the heads from first to last, so +// it never read those blocks. +func (t *InclusionTracker) recordBlindGap(ctx context.Context, first, last uint64) { + for s := range t.state.Lock() { + s.blindHeights++ + } + log.Printf("inclusion tracker: missed heads %d..%d (%d blocks), never read", + first, last, last-first+1) + // One height, one count, the same as every other reason on this counter. + // The watermark above rises once, because the reap only asks whether it + // rose, but a counter of heights that reported a gap as one would undercount + // by the length of every gap. + inclusionBlockFetchErrors.Add(ctx, int64(last-first+1), metric.WithAttributes( + attribute.String("chain_id", t.seiChainID), + attribute.String("reason", reasonMissedHead))) +} + +// recordBlindFetch marks that the run could not read one block's receipts. Every +// tx in flight now reaps as status_unavailable rather than expired, because the +// run cannot tell the two apart for a block it never read. +func (t *InclusionTracker) recordBlindFetch(ctx context.Context, num uint64, reason string, err error) { + for s := range t.state.Lock() { + s.blindHeights++ + } + if err != nil { + log.Printf("inclusion tracker: fetch block %d (%s): %v", num, reason, err) + } else { + log.Printf("inclusion tracker: block %d returned no receipts (%s)", num, reason) + } + inclusionBlockFetchErrors.Add(ctx, 1, metric.WithAttributes( + attribute.String("chain_id", t.seiChainID), + attribute.String("reason", reason))) +} + +// Reasons a receipt read failed. They are metric label values and preflight +// branches on two of them, so one constant owns each string rather than a +// literal at every site. +const ( + reasonNotFound = "not_found" + reasonTimeout = "timeout" + reasonOther = "other" + + // These name why a height went unread rather than why a call failed. They + // share the counter and not the cause, because the operator's next move + // differs: a node behind the head is a topology problem, a head the run + // never saw is a subscription problem, and a receipt the node could not + // produce is a chain problem. + // reasonTrackingStopped is the head stream ending mid-run. Every later + // height is unread, so nothing after it is evidence about the chain. + reasonTrackingStopped = "tracking_stopped" + reasonMissedHead = "missed_head" + reasonNullReceipt = "null_receipt" +) + +// readTimeout bounds one height's whole read, retries included. Head processing +// is serial, so this is what one head costs at worst. +// +// It should stay below the run's reap deadline. A read that returns after the +// reap has already evicted a transaction answers a question the run has closed, +// and converts a real inclusion into status_unavailable. +const readTimeout = 10 * time.Second + +// notFoundBackoff paces the retry of a height the node has not made readable. +// It is short relative to the block interval, because the wait it paces is the +// node's own commit-to-watermark gap and not a block. +const notFoundBackoff = 25 * time.Millisecond + +// errHeadStreamEnded marks the head subscription ending on its own, which stops +// the tracking and not the run. It is separate from ErrEndpointUnusable because +// the two want opposite outcomes: an endpoint that cannot serve the run should +// stop it before it wastes gas, and a transport that drops mid-run should not +// fail a run whose senders are still working. +var errHeadStreamEnded = errors.New("head stream ended") + +// ErrEndpointUnusable means the receipt endpoint cannot serve this run, and the +// run stops rather than reporting numbers it did not measure. +// +// It is a sentinel because main treats a context.Canceled or a +// context.DeadlineExceeded as the run ending normally. Wrapping the probe's own +// error would have made a refusal exit zero, which is a Complete run with no +// load, read by anything downstream as a chain that included nothing. +var ErrEndpointUnusable = errors.New("receipt endpoint unusable") + +// How hard the preflight tries before it speaks for the whole run. +const ( + preflightAttempts = 3 + preflightTimeout = 5 * time.Second + preflightBackoff = 500 * time.Millisecond +) + +// fetchFailureReason buckets a receipt-read error so an operator reads the cause +// off a dashboard instead of the pod log. The strings are label values: keep +// them few, and keep them stable. +// +// The typed checks come first because they hold across servers. The substring +// checks below them do not, and they are here because no typed error exists for +// what they catch. Sei writes "receipts have been pruned; earliest available is +// N" for one and "not yet available" for the other, which differ by one word, so +// pruning is tested before anything matching on availability. +func fetchFailureReason(err error) string { + // not_found is the only reason the read retries, so anything that reaches it + // by accident is retried for the whole read budget against a node that will + // never answer differently. + if errors.Is(err, ethereum.NotFound) { + return reasonNotFound + } + if errors.Is(err, context.DeadlineExceeded) { + return reasonTimeout + } + var netErr net.Error + if errors.As(err, &netErr) && netErr.Timeout() { + return reasonTimeout + } + // An answer with a status is the server refusing, whatever prose it chose. + // rpc.HTTPError.Error() renders as "404 Not Found: 404 page not found", so + // the substring below would read an ingress error page as a height the node + // has not reached, and retry it hundreds of times. + var httpErr rpc.HTTPError + if errors.As(err, &httpErr) { + return reasonOther + } + var rpcErr rpc.Error + if errors.As(err, &rpcErr) { + return reasonOther + } + if strings.Contains(strings.ToLower(err.Error()), "not found") { + return reasonNotFound } + return reasonOther } // InclusionSummary is the conservation tally. Read only after both sender and // the tracker have joined, so inflightAtShutdown is final. type InclusionSummary struct { - Included uint64 - Expired uint64 + Included uint64 + // Expired is a tx the run read every block for and never saw. It is a claim + // about the chain. + Expired uint64 + // StatusUnavailable is a tx that was in flight while a receipt read failed. + // It is a claim about the run, and the two are counted apart so the closing + // log line and the outcome ledger cannot disagree about one transaction. + StatusUnavailable uint64 DroppedAtCap uint64 InflightAtShutdown uint64 } @@ -309,6 +926,7 @@ func (t *InclusionTracker) Summary() InclusionSummary { return InclusionSummary{ Included: s.included, Expired: s.expired, + StatusUnavailable: s.statusUnavailable + s.duplicates, DroppedAtCap: s.droppedAtCap, InflightAtShutdown: s.inflightAtShutdown, } diff --git a/stats/inclusion_tracker_test.go b/stats/inclusion_tracker_test.go index 7efbc13..7a9ffb9 100644 --- a/stats/inclusion_tracker_test.go +++ b/stats/inclusion_tracker_test.go @@ -2,6 +2,7 @@ package stats import ( "context" + "github.com/ethereum/go-ethereum" "math/big" "sync" "sync/atomic" @@ -15,34 +16,98 @@ import ( "github.com/sei-protocol/sei-load/types" ) -// MockBlockSource is a deterministic blockSource for tests. Setter style mirrors -// MockBlockStats: SetBlock seeds a block's tx hashes; fetches are counted. +// MockBlockSource is a deterministic receiptSource for tests. +// +// Three knobs inject failure and they do not compose: a sequence set by +// SetErrSequence wins while it has entries, and SetFetchErr and FailTimes share +// one slot, so the later call replaces the earlier. Setter style +// mirrors MockBlockStats: SetBlock seeds a block's receipts; fetches are counted. +// +// SetBlock takes hashes and marks each committed, which is what most tests want. +// SetReceipts takes the pairs, for the tests that care about status. type MockBlockSource struct { mu sync.Mutex - blocks map[uint64][]common.Hash + blocks map[uint64][]blockReceipt fetches atomic.Int64 fetchErr error + failures int + nulls int + errSeq []error } func NewMockBlockSource() *MockBlockSource { - return &MockBlockSource{blocks: make(map[uint64][]common.Hash)} + return &MockBlockSource{blocks: make(map[uint64][]blockReceipt)} } func (m *MockBlockSource) SetBlock(n uint64, hashes ...common.Hash) *MockBlockSource { + rs := make([]blockReceipt, len(hashes)) + for i, h := range hashes { + rs[i] = blockReceipt{Hash: h, Status: ethtypes.ReceiptStatusSuccessful, HasStatus: true} + } + return m.SetReceipts(n, rs...) +} + +func (m *MockBlockSource) SetReceipts(n uint64, rs ...blockReceipt) *MockBlockSource { + m.mu.Lock() + defer m.mu.Unlock() + m.blocks[n] = rs + return m +} + +// SetFetchErr makes every later fetch fail, so a test can drive the branch where +// the run cannot read a block at all. +func (m *MockBlockSource) SetFetchErr(err error) *MockBlockSource { + m.mu.Lock() + defer m.mu.Unlock() + m.fetchErr = err + m.failures = -1 // every fetch + return m +} + +// SetNulls makes every later read report n receipts the endpoint sent as null, +// so a test can drive a read that partly succeeded. +func (m *MockBlockSource) SetNulls(n int) *MockBlockSource { m.mu.Lock() defer m.mu.Unlock() - m.blocks[n] = hashes + m.nulls = n return m } -func (m *MockBlockSource) BlockTxHashes(_ context.Context, n uint64) ([]common.Hash, error) { +// SetErrSequence makes each later read fail with the next error in turn, so a +// test can drive an endpoint whose cause changes between attempts. +func (m *MockBlockSource) SetErrSequence(errs ...error) *MockBlockSource { + m.mu.Lock() + defer m.mu.Unlock() + m.errSeq = errs + return m +} + +// FailTimes makes the next n fetches fail and the rest succeed, so a test can +// drive a node that recovers rather than one that stays broken. +func (m *MockBlockSource) FailTimes(n int, err error) *MockBlockSource { + m.mu.Lock() + defer m.mu.Unlock() + m.fetchErr = err + m.failures = n + return m +} + +func (m *MockBlockSource) BlockReceipts(_ context.Context, n uint64) ([]blockReceipt, int, error) { m.fetches.Add(1) - if m.fetchErr != nil { - return nil, m.fetchErr - } m.mu.Lock() defer m.mu.Unlock() - return m.blocks[n], nil + if len(m.errSeq) > 0 { + err := m.errSeq[0] + m.errSeq = m.errSeq[1:] + return nil, 0, err + } + if m.fetchErr != nil && m.failures != 0 { + if m.failures > 0 { + m.failures-- + } + return nil, 0, m.fetchErr + } + return m.blocks[n], m.nulls, nil } func (m *MockBlockSource) FetchCount() int64 { return m.fetches.Load() } @@ -50,15 +115,27 @@ func (m *MockBlockSource) FetchCount() int64 { return m.fetches.Load() } // newTestTracker builds an open-loop tracker wired to a mock source, skipping // the live dial. Open-loop is the default so latency-bearing tests exercise the // inclusion_latency path; closed-loop is covered explicitly via newTestTrackerLoop. -func newTestTracker(t *testing.T, reapAfter time.Duration, maxInflight int, src blockSource) *InclusionTracker { +func newTestTracker(t *testing.T, reapAfter time.Duration, maxInflight int, src receiptSource) *InclusionTracker { t.Helper() return newTestTrackerLoop(t, reapAfter, maxInflight, src, true) } -func newTestTrackerLoop(t *testing.T, reapAfter time.Duration, maxInflight int, src blockSource, openLoop bool) *InclusionTracker { +// newTestTrackerLoop builds a tracker that has already read a block. +// +// A run that has not read anything yet cannot call a transaction expired: it may +// have landed in a block the run never opened, which is what happens on every +// real run while the tracker is still dialling. These tests are about what +// happens once the run is underway, so they start there. +// TestATxAcceptedBeforeTheFirstReadIsNotAChainVerdict covers the other case +// through Run itself. +func newTestTrackerLoop(t *testing.T, reapAfter time.Duration, maxInflight int, src receiptSource, openLoop bool) *InclusionTracker { t.Helper() - return newInclusionTrackerWithSource( - NewInclusionTracker("test-chain", reapAfter, maxInflight, openLoop), src) + tr := newInclusionTrackerWithSource( + NewInclusionTracker("test-chain", reapAfter, maxInflight, openLoop, NewCollector()), src) + for s := range tr.state.Lock() { + s.firstReadAt = time.Now().Add(-time.Hour) + } + return tr } // loadTx builds a LoadTx with a deterministic hash from nonce and an intended @@ -100,12 +177,12 @@ func TestInclusion_MatchStamps(t *testing.T) { intended := time.Unix(1000, 0) tx := loadTx(1, intended) - tr.Register(tx) + tr.Register(context.Background(), tx) require.Equal(t, 1, inflightLen(t, tr)) arrival := time.Unix(1002, 0) src.SetBlock(5, tx.EthTx.Hash()) - tr.matchBlock(context.Background(), 5, arrival) + tr.matchBlock(context.Background(), 5, 0, arrival) require.Equal(t, arrival, tx.InclusionTime, "InclusionTime is the header-arrival time") require.Equal(t, 0, inflightLen(t, tr), "matched tx leaves the registry") @@ -123,10 +200,10 @@ func TestInclusion_ClosedLoopCountsNoLatency(t *testing.T) { require.False(t, tr.openLoop, "tracker built closed-loop: latency sample is gated") tx := loadTx(1, time.Unix(1000, 0)) - tr.Register(tx) + tr.Register(context.Background(), tx) arrival := time.Unix(1002, 0) src.SetBlock(5, tx.EthTx.Hash()) - tr.matchBlock(context.Background(), 5, arrival) + tr.matchBlock(context.Background(), 5, 0, arrival) require.Equal(t, arrival, tx.InclusionTime, "InclusionTime still stamped in closed-loop") require.Equal(t, uint64(1), tr.Summary().Included, "included count tracked in closed-loop") @@ -137,11 +214,11 @@ func TestInclusion_ReapExpires(t *testing.T) { tr := newTestTracker(t, 10*time.Millisecond, 100, NewMockBlockSource()) tx := loadTx(1, time.Now()) - tr.Register(tx) + tr.Register(context.Background(), tx) require.Equal(t, 1, inflightLen(t, tr)) time.Sleep(20 * time.Millisecond) - tr.reap() + tr.reap(context.Background()) require.Equal(t, 0, inflightLen(t, tr), "reaped tx leaves the registry (no leak)") require.True(t, tx.InclusionTime.IsZero(), "reaped tx is never stamped") @@ -156,11 +233,11 @@ func TestInclusion_ReapVsLateInclusion(t *testing.T) { src := NewMockBlockSource() tr := newTestTracker(t, time.Nanosecond, 100, src) tx := loadTx(1, time.Unix(1000, 0)) - tr.Register(tx) + tr.Register(context.Background(), tx) time.Sleep(time.Millisecond) - tr.reap() // wins: expired + tr.reap(context.Background()) // wins: expired src.SetBlock(5, tx.EthTx.Hash()) - tr.matchBlock(context.Background(), 5, time.Unix(1002, 0)) // no-op + tr.matchBlock(context.Background(), 5, 0, time.Unix(1002, 0)) // no-op s := tr.Summary() require.Equal(t, uint64(1), s.Expired) require.Equal(t, uint64(0), s.Included) @@ -170,11 +247,11 @@ func TestInclusion_ReapVsLateInclusion(t *testing.T) { src := NewMockBlockSource() tr := newTestTracker(t, time.Nanosecond, 100, src) tx := loadTx(1, time.Unix(1000, 0)) - tr.Register(tx) + tr.Register(context.Background(), tx) src.SetBlock(5, tx.EthTx.Hash()) - tr.matchBlock(context.Background(), 5, time.Unix(1002, 0)) // wins: included + tr.matchBlock(context.Background(), 5, 0, time.Unix(1002, 0)) // wins: included time.Sleep(time.Millisecond) - tr.reap() // no-op + tr.reap(context.Background()) // no-op s := tr.Summary() require.Equal(t, uint64(1), s.Included) require.Equal(t, uint64(0), s.Expired) @@ -188,7 +265,7 @@ func TestInclusion_BoundedCap(t *testing.T) { tr := newTestTracker(t, time.Minute, cap, NewMockBlockSource()) for i := range uint64(10) { - tr.Register(loadTx(i, time.Now())) + tr.Register(context.Background(), loadTx(i, time.Now())) require.LessOrEqual(t, inflightLen(t, tr), cap, "map never exceeds cap") } s := tr.Summary() @@ -201,7 +278,7 @@ func TestInclusion_BoundedCap(t *testing.T) { func TestInclusion_NonPositiveCapFallsBack(t *testing.T) { tr := newTestTracker(t, time.Minute, 0, NewMockBlockSource()) for i := range uint64(5) { - tr.Register(loadTx(i, time.Now())) + tr.Register(context.Background(), loadTx(i, time.Now())) } require.Equal(t, 5, inflightLen(t, tr), "registrations are admitted, not all dropped") require.Equal(t, uint64(0), tr.Summary().DroppedAtCap) @@ -249,11 +326,11 @@ func TestInclusion_Conservation(t *testing.T) { txs := make([]*types.LoadTx, tc.attempts) for i := range txs { txs[i] = loadTx(uint64(i), time.Unix(1000, 0)) - tr.Register(txs[i]) + tr.Register(context.Background(), txs[i]) } for i := 0; i < tc.matched; i++ { src.SetBlock(uint64(i), txs[i].EthTx.Hash()) - tr.matchBlock(context.Background(), uint64(i), time.Unix(1001, 0)) + tr.matchBlock(context.Background(), uint64(i), 0, time.Unix(1001, 0)) } // Reap the next `reaped` txs by forcing their registeredAt past cutoff. for s := range tr.state.Lock() { @@ -266,14 +343,17 @@ func TestInclusion_Conservation(t *testing.T) { } } } - tr.reap() + tr.reap(context.Background()) s := tr.Summary() - // dropped_at_cap is excluded from the registered set, so every Register - // attempt is accounted by exactly one of the four buckets. + // Every Register attempt is accounted by exactly one bucket. + // status_unavailable is one of them: a reap reaches it whenever the + // run could not see, and an identity that omits it cannot notice a + // leg migrating there. require.Equal(t, uint64(tc.attempts), - s.Included+s.Expired+s.InflightAtShutdown+s.DroppedAtCap, - "attempts == included + expired + inflight_at_shutdown + dropped_at_cap") + s.Included+s.Expired+s.StatusUnavailable+s.InflightAtShutdown+s.DroppedAtCap, + "attempts == included + expired + status_unavailable + "+ + "inflight_at_shutdown + dropped_at_cap") }) } } @@ -285,9 +365,9 @@ func TestInclusion_GapNoBackfill(t *testing.T) { tr := newTestTracker(t, time.Minute, 100, src) ctx := context.Background() - last := tr.processHead(ctx, 10, time.Now(), 0) // seeds, no gap - last = tr.processHead(ctx, 11, time.Now(), last) - last = tr.processHead(ctx, 15, time.Now(), last) // gap 12,13,14 + last := tr.processHead(ctx, 10, 0, time.Now(), 0) // seeds, no gap + last = tr.processHead(ctx, 11, 0, time.Now(), last) + last = tr.processHead(ctx, 15, 0, time.Now(), last) // gap 12,13,14 _ = last require.Equal(t, int64(3), src.FetchCount(), @@ -310,26 +390,98 @@ func TestInclusion_ConcurrentRaceSafe(t *testing.T) { go func() { defer wg.Done() for i := range txs { - tr.Register(txs[i]) + tr.Register(context.Background(), txs[i]) } }() go func() { defer wg.Done() for i := range txs { - tr.matchBlock(context.Background(), uint64(i), time.Unix(1001, 0)) + tr.matchBlock(context.Background(), uint64(i), 0, time.Unix(1001, 0)) } }() go func() { defer wg.Done() for range 50 { - tr.reap() + tr.reap(context.Background()) time.Sleep(100 * time.Microsecond) } }() wg.Wait() - tr.reap() + tr.reap(context.Background()) s := tr.Summary() require.Equal(t, uint64(n), s.Included+s.Expired+s.InflightAtShutdown, "conservation holds under concurrency") } + +// CountingSlowSource records which heights were asked for, and answers each +// after a delay. heldFrom names the first height it actually holds; anything +// below that answers not-found, so a test can drive a node whose availability +// is not monotone with the queue's age order. +type CountingSlowSource struct { + mu sync.Mutex + delay time.Duration + heldFrom uint64 + asked map[uint64]int + receipts map[uint64][]blockReceipt +} + +func NewCountingSlowSource(delay time.Duration, heldFrom uint64) *CountingSlowSource { + return &CountingSlowSource{ + delay: delay, heldFrom: heldFrom, + asked: map[uint64]int{}, receipts: map[uint64][]blockReceipt{}, + } +} + +func (c *CountingSlowSource) Hold(n uint64, rs ...blockReceipt) *CountingSlowSource { + c.mu.Lock() + defer c.mu.Unlock() + c.receipts[n] = rs + return c +} + +func (c *CountingSlowSource) Asked() map[uint64]int { + c.mu.Lock() + defer c.mu.Unlock() + out := make(map[uint64]int, len(c.asked)) + for k, v := range c.asked { + out[k] = v + } + return out +} + +func (c *CountingSlowSource) BlockReceipts(ctx context.Context, n uint64) ([]blockReceipt, int, error) { + c.mu.Lock() + c.asked[n]++ + delay, held, rs := c.delay, n >= c.heldFrom, c.receipts[n] + c.mu.Unlock() + select { + case <-time.After(delay): + case <-ctx.Done(): + return nil, 0, ctx.Err() + } + if !held { + return nil, 0, ethereum.NotFound + } + return rs, 0, nil +} + +// SlowBlockSource fails every read after a delay, so a test can measure what a +// hanging endpoint costs a head. +type SlowBlockSource struct { + err error + delay time.Duration +} + +func NewSlowBlockSource(err error, delay time.Duration) *SlowBlockSource { + return &SlowBlockSource{err: err, delay: delay} +} + +func (s *SlowBlockSource) BlockReceipts(ctx context.Context, _ uint64) ([]blockReceipt, int, error) { + select { + case <-time.After(s.delay): + return nil, 0, s.err + case <-ctx.Done(): + return nil, 0, ctx.Err() + } +} diff --git a/stats/metrics.go b/stats/metrics.go index 70ffd46..6b6c51e 100644 --- a/stats/metrics.go +++ b/stats/metrics.go @@ -57,9 +57,11 @@ var ( metric.WithUnit("{transactions}"))) // Inclusion tracker. inclusion_latency is open-loop only (closed-loop - // IntendedSendTime is enqueue time, not a schedule); its _count is the - // included count only there. Denominator for inclusion rate is the existing - // succeeded/txs_accepted series, never a new "registered" series. + // IntendedSendTime is enqueue time, not a schedule). Its _count is every tx + // a receipt named there, which is InclusionSummary.Included plus any whose + // status could not be read, so the two are not interchangeable. Denominator + // for inclusion rate is the existing succeeded/txs_accepted series, never a + // new "registered" series. inclusionLatency = must(meter.Float64Histogram( "inclusion_latency", metric.WithDescription("Latency from intended send to observed on-chain inclusion in seconds"), @@ -68,17 +70,34 @@ 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("Terminal outcome of every tx the inclusion tracker registered. See stats.Outcome for the values."), metric.WithUnit("{transactions}"))) inclusionBlockGaps = must(meter.Int64Counter( "block_gaps", - metric.WithDescription("Missed block heights observed by the inclusion tracker (no backfill)"), + metric.WithDescription("Block heights the head subscription never delivered, counted per height. Not backfilled; the gap is recorded once as a hole on block_fetch_errors{reason=missed_head}"), + metric.WithUnit("{blocks}"))) + + inclusionReadWait = must(meter.Float64Histogram( + "block_read_wait", + metric.WithDescription("How long a block's receipts took to become readable after its head arrived. seid publishes a head from Commit while the receipt store's writer is still async, so this is that gap. A rising distribution is the node's watermark falling behind its own head"), + metric.WithUnit("s"), + metric.WithExplicitBucketBoundaries(0.005, 0.025, 0.1, 0.25, 0.5, 1, 2, 5, 10))) + + inclusionHeadLag = must(meter.Float64Histogram( + "head_lag", + metric.WithDescription("How long a head waited between reaching this process and its block being read. This is the tracker's own lag: un-included txs with this near zero are the chain's doing, and with this climbing they are the run's"), + metric.WithUnit("s"), + metric.WithExplicitBucketBoundaries(0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10, 30))) + + inclusionEmptyWithGas = must(meter.Int64Counter( + "block_empty_with_gas", + metric.WithDescription("Blocks that returned no receipts while the head reported gas burned. Causes: non-EVM traffic, an EVM tx that failed the ante handler and got no receipt, or receipts going missing from the store. Only the last is a defect, and this counter cannot tell them apart"), metric.WithUnit("{blocks}"))) inclusionBlockFetchErrors = must(meter.Int64Counter( "block_fetch_errors", - metric.WithDescription("Block-body fetches that failed; the block's txs go unmatched and reap as expired (no retry)"), + metric.WithDescription("Heights the run could not read, by reason; a tx in flight across one reaches status_unavailable rather than expired"), metric.WithUnit("{blocks}"))) // Run-summary: the only inclusion tally with no live series, since it is the diff --git a/stats/outcome.go b/stats/outcome.go new file mode 100644 index 0000000..c45a483 --- /dev/null +++ b/stats/outcome.go @@ -0,0 +1,111 @@ +package stats + +// Outcome is what became of one transaction the endpoint accepted. +// +// InclusionTracker 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. +// +// Two distinctions carry the point of the type. Committed and Reverted separate a +// transaction that did what the workload asked from one that burned its gas +// doing nothing, which an inclusion count cannot tell apart. And +// StatusUnavailable separates "the run did not see" from "the chain did not take +// it", which decides whether a low goodput ratio is a finding about the chain or +// a finding about the run. +// +// Six states, and a residual. Every accepted transaction reaches exactly one of +// them, or is still in the registry when the run ends: +// +// accepted = committed + reverted + status_unavailable +// + expired + dropped_at_cap + dropped_at_handoff +// + inflight_at_shutdown +type Outcome uint8 + +const ( + // outcomeUnset is the zero value, and no caller may report it. It exists so + // that a transaction nobody classified cannot read as a success. + // + // Committed at index 0 would mean an unassigned variable, a switch that + // matched no case, or an early return counts as a commit. That is the + // failure this whole type exists to remove, and it would arrive silently. + // + // RecordOutcome folds this and any out-of-range value into one visible + // count. A non-zero Unrecorded means sei-load has a bug: nothing else + // produces one. + outcomeUnset Outcome = iota + + // OutcomeCommitted is a receipt reporting a successful status. + OutcomeCommitted + + // OutcomeReverted is a receipt reporting a failed status. Named for what the + // EVM calls it, and named apart from RunSummary.Failed, which is a send that + // returned an error rather than a transaction the chain took and ran. + // + // A receipt carries one status bit. An explicit revert, an out-of-gas and an + // invalid opcode all arrive here. Separating them needs a trace call for + // every transaction, and this tracker reads status once per block however + // many transactions that block carries. So the state names what the run + // observed, not a cause it cannot see. + OutcomeReverted + + // OutcomeExpired is a transaction no receipt named within reapAfter. + // + // The run cannot say why, and one state covers every reason: no peer + // gossiped it, a mempool evicted it, nobody proposed it, or it reached a + // block whose receipt was never written. A reader of receipts cannot + // separate those. + OutcomeExpired + + // OutcomeDroppedAtCap is a transaction the registry could not admit, because + // it already held maxInflight entries. + OutcomeDroppedAtCap + + // OutcomeDroppedAtHandoff is a transaction the tracker could not take from + // the sender. + // + // Distinct from the dispatcher's own load shed, which RunSummary.Dropped + // counts: that transaction never reached the chain. This one did, and the + // run then lost track of it. + OutcomeDroppedAtHandoff + + // OutcomeStatusUnavailable is a transaction whose execution status the run + // could not read. + // + // It is not Expired. The chain may well have included it, and the run simply + // did not see. Counting it as Expired would report a chain problem where the + // truth is a measurement problem. + OutcomeStatusUnavailable + + // outcomeCount bounds the array indexed by Outcome, and MUST stay last. + // + // Add a new state above it. A state added below takes the value outcomeCount + // already holds, so RecordOutcome folds every call carrying it into + // Unrecorded: the state's own count reads zero for the life of the run, and + // no compiler check catches it. + outcomeCount +) + +// outcomeNames are FROZEN wire values, on the same footing as the operation +// names in config/operation.go. Once a run emits one, a dashboard query and a +// saved report both match it by value, and a later rename orphans every one. +// Add a state rather than repurpose one. +// +// The array is indexed by Outcome, so a state added without a name here reads +// as the empty string rather than silently taking another state's name. +var outcomeNames = [outcomeCount]string{ + outcomeUnset: "unrecorded", + OutcomeCommitted: "committed", + OutcomeReverted: "reverted", + OutcomeExpired: "expired", + OutcomeDroppedAtCap: "dropped_at_cap", + OutcomeDroppedAtHandoff: "dropped_at_handoff", + OutcomeStatusUnavailable: "status_unavailable", +} + +// String names the outcome for a report and a metric label. +func (o Outcome) String() string { + if o >= outcomeCount { + return outcomeNames[outcomeUnset] + } + return outcomeNames[o] +} diff --git a/stats/run_summary.go b/stats/run_summary.go index 9b46633..7a94f6e 100644 --- a/stats/run_summary.go +++ b/stats/run_summary.go @@ -26,16 +26,21 @@ type RunSummary struct { // scheduled = dropped + succeeded + failed) is auditable from the run summary. Failed uint64 - // Inclusion-stage tally (see sender/doc.go). The conservation identity is - // registered == Included + Expired + InflightAtShutdown, with - // registered ⊆ succeeded. InclusionTracked disambiguates a not-tracked run - // (all zero, flag false) from a tracked run with no inclusions yet. + // Inclusion-stage tally. sender/doc.go states the conservation identity these + // satisfy; do not restate it here, because it has already drifted once. + // InclusionTracked disambiguates a not-tracked run (all zero, flag false) + // from a tracked run with no inclusions yet. // TODO(PLT-467): owns run-summary schema versioning for these fields. InclusionTracked bool // Included is the count of txs the tracker observed on-chain (stamped). Included uint64 - // Expired is the count of registered txs reaped un-included after reapAfter. + // Expired is the count of registered txs the run read every block for and + // never saw. It is a claim about the chain. Expired uint64 + // StatusUnavailable is the count of registered txs that were in flight while + // a receipt read failed. It is a claim about the run, and it is counted + // apart from Expired so the two cannot be read as one number. + StatusUnavailable uint64 // DroppedAtCap is the count of successful sends rejected at the in-flight cap; // excluded from the inclusion denominator (they were never registered). DroppedAtCap uint64