From f89c4b1d15ae627b049732a51fec94b7727fbaf1 Mon Sep 17 00:00:00 2001 From: Michael Standen Date: Fri, 6 Nov 2026 10:19:24 +1300 Subject: [PATCH] ethmonitor: detect stalled newHeads streams and fall back to polling --- ethmonitor/README.md | 2 + ethmonitor/ethmonitor.go | 65 ++++++-- ethmonitor/prefetch_test.go | 36 ++++- ethmonitor/stream_health_test.go | 256 +++++++++++++++++++++++++++++++ 4 files changed, 349 insertions(+), 10 deletions(-) create mode 100644 ethmonitor/stream_health_test.go diff --git a/ethmonitor/README.md b/ethmonitor/README.md index bd2c2e59..0454231f 100644 --- a/ethmonitor/README.md +++ b/ethmonitor/README.md @@ -122,6 +122,8 @@ Start from `ethmonitor.DefaultOptions` and override what you need. | `StreamingErrNumToSwitchToPolling` | 3 | Streaming errors before switching to polling. | | `StreamingErrorResetInterval` | 2m | How often the streaming error count resets. | | `StreamingRetryAfter` | 5m | How long to poll before retrying streaming. | +| `StreamHealthCheckInterval` | 10s | How often to verify the last received head is still fresh while streaming. 0 disables the check. | +| `StreamHealthCheckStaleness` | 60s | Max age of the last head received from the stream before the subscription is dropped for polling. Set it well above the chain's block time. | | `Timeout` | 20s | Timeout for node requests. | | `StartBlockNumber` | nil (latest) | Block to start from. A negative value starts that many blocks behind the latest. | | `Bootstrap` | false | Start from a chain restored with `Chain().BootstrapFromBlocks` / `BootstrapFromBlocksJSON`. Takes precedence over `StartBlockNumber`. | diff --git a/ethmonitor/ethmonitor.go b/ethmonitor/ethmonitor.go index 662db229..b75c0622 100644 --- a/ethmonitor/ethmonitor.go +++ b/ethmonitor/ethmonitor.go @@ -32,6 +32,8 @@ var DefaultOptions = Options{ StreamingErrorResetInterval: 2 * time.Minute, StreamingRetryAfter: 5 * time.Minute, StreamingErrNumToSwitchToPolling: 3, + StreamHealthCheckInterval: 10 * time.Second, + StreamHealthCheckStaleness: 60 * time.Second, StreamingDisabled: false, UnsubscribeOnStop: false, Timeout: 20 * time.Second, @@ -70,6 +72,16 @@ type Options struct { // StreamingErrNumToSwitchToPolling is the number of errors before switching to polling StreamingErrNumToSwitchToPolling int + // StreamHealthCheckInterval is how often the streaming listener verifies + // the last received head is still fresh. 0 disables the check. + StreamHealthCheckInterval time.Duration + + // StreamHealthCheckStaleness is the max age of the head last received + // from the stream before the subscription is considered stalled, dropped, + // and counted toward the switch to polling. Set it well above the chain's + // block time. + StreamHealthCheckStaleness time.Duration + // StreamingDisabled flag to force disable streaming even if the provider supports it StreamingDisabled bool @@ -225,6 +237,10 @@ func NewMonitor(provider ethrpc.RawInterface, options ...Options) (*Monitor, err opts.MaxLogsAttempts = 10 } + if opts.StreamHealthCheckInterval > 0 && opts.StreamHealthCheckStaleness <= opts.StreamHealthCheckInterval { + return nil, fmt.Errorf("ethmonitor: StreamHealthCheckStaleness must be greater than StreamHealthCheckInterval") + } + if opts.PrefetchConcurrency < 0 { opts.PrefetchConcurrency = 0 } @@ -462,15 +478,6 @@ func (m *Monitor) listenNewHead(ctx context.Context, wg *sync.WaitGroup) <-chan streamingErrCount = 0 } - // TODO: even if streaming is enabled, and its running, we still need to add a - // "streamHealthCheck" that checks if the stream is still running, as perhaps the - // upstream service has a problem (which happens). - // - // The way to check this, is every 10 seconds we ask the node for the latest block - // and if the latest block from the stream is different from the latest block from the - // node, then we switch to polling mode. And after the reset interval we will try back - // again. We also need to call alerter.Alert() to notify the user that the stream is down. - // // TODO: maybe we should add to RawInterface() to "inform" / "notify" that the provider // is producing errors.. ie. for the Node or WS provider.. and so, we can tell the upstream // that the provider is behaving issues... ReportFault(err), ReportFaultWS(err) @@ -500,10 +507,32 @@ func (m *Monitor) listenNewHead(ctx context.Context, wg *sync.WaitGroup) <-chan goto reconnect } + // A subscription can silently serve stale heads: no error arrives + // on sub.Err(), while the node's http api keeps advancing. Track + // the chain timestamp of the last received head: a stream that + // keeps delivering while serving heads whose data is old is just + // as broken as one that goes silent. The subscription is dropped, + // counting it toward the switch to polling, when the head stops + // being fresh. Seeded on connect so a stream that never delivers + // a head is caught too. A zero StreamHealthCheckInterval disables + // the check, leaving the subscription loop active. + var lastHeadTime atomic.Int64 + lastHeadTime.Store(time.Now().Unix()) + + var streamHealthCheck *time.Ticker + var streamHealthCheckC <-chan time.Time + if m.options.StreamHealthCheckInterval > 0 { + streamHealthCheck = time.NewTicker(m.options.StreamHealthCheckInterval) + streamHealthCheckC = streamHealthCheck.C + } + for { select { case <-ctx.Done(): // if we're done, we'll unsubscribe and close the nextBlock channel + if streamHealthCheck != nil { + streamHealthCheck.Stop() + } sub.Unsubscribe() close(nextBlock) return @@ -513,6 +542,9 @@ func (m *Monitor) listenNewHead(ctx context.Context, wg *sync.WaitGroup) <-chan m.log.Warn(fmt.Sprintf("ethmonitor (chain %s): websocket subscription closed, error: %v", m.chainID.String(), err)) m.alert.Alert(context.Background(), "ethmonitor (chain %s): websocket subscription closed, error: %v", m.chainID.String(), err) sub.Unsubscribe() + if streamHealthCheck != nil { + streamHealthCheck.Stop() + } streamingErrLastTime = time.Now() goto reconnect @@ -520,6 +552,7 @@ func (m *Monitor) listenNewHead(ctx context.Context, wg *sync.WaitGroup) <-chan case newHead := <-newHeads: latestHeadBlock.Store(newHead.Number.Uint64()) m.latestHead.Store(newHead.Number.Uint64()) + lastHeadTime.Store(int64(newHead.Time)) if m.prefetch != nil { m.prefetch.notify() } @@ -528,6 +561,20 @@ func (m *Monitor) listenNewHead(ctx context.Context, wg *sync.WaitGroup) <-chan default: // non-blocking } + + case <-streamHealthCheckC: + last := lastHeadTime.Load() + if last == 0 || time.Since(time.Unix(last, 0)) <= m.options.StreamHealthCheckStaleness { + continue + } + age := time.Since(time.Unix(last, 0)).Round(time.Second) + m.log.Warn(fmt.Sprintf("ethmonitor (chain %s): newHeads stream stalled, last head block %d is %s old, reconnecting", m.chainID.String(), latestHeadBlock.Load(), age)) + m.alert.Alert(context.Background(), "ethmonitor (chain %s): newHeads stream stalled, last head block %d is %s old, reconnecting", m.chainID.String(), latestHeadBlock.Load(), age) + sub.Unsubscribe() + streamHealthCheck.Stop() + + streamingErrLastTime = time.Now() + goto reconnect } } } else { diff --git a/ethmonitor/prefetch_test.go b/ethmonitor/prefetch_test.go index 599463c5..bc58d847 100644 --- a/ethmonitor/prefetch_test.go +++ b/ethmonitor/prefetch_test.go @@ -660,6 +660,14 @@ type fakeChain struct { fork int heads []*fakeSubscription held map[uint64]chan struct{} // block numbers not served until released + + // muteHeads stops head notifications while the chain keeps producing, + // simulating a stalled newHeads stream against a live node. + muteHeads atomic.Bool + + // staleTimestamps makes head notifications carry old chain timestamps, + // simulating a stream that keeps delivering while serving stale data. + staleTimestamps atomic.Bool } type fakeBlock struct { @@ -701,9 +709,19 @@ func (c *fakeChain) produce(ctx context.Context, interval time.Duration) { c.mu.Lock() b := c.appendLocked() heads := append([]*fakeSubscription{}, c.heads...) + muted := c.muteHeads.Load() c.mu.Unlock() - header := &types.Header{Number: new(big.Int).SetUint64(b.num + c.announceAhead)} + if muted { + continue + } + + ts := time.Now() + if c.staleTimestamps.Load() { + ts = ts.Add(-30 * time.Minute) + } + + header := &types.Header{Number: new(big.Int).SetUint64(b.num + c.announceAhead), Time: uint64(ts.Unix())} for _, sub := range heads { sub.send(header) } @@ -711,6 +729,22 @@ func (c *fakeChain) produce(ctx context.Context, interval time.Duration) { } } +// mute stops head notifications; the chain keeps producing and serving +// blocks over http. unmute resumes them. +func (c *fakeChain) mute() { + c.muteHeads.Store(true) +} + +func (c *fakeChain) unmute() { + c.muteHeads.Store(false) +} + +// serveStaleTimestamps makes subsequent head notifications carry chain +// timestamps half an hour old, while the http api serves current blocks. +func (c *fakeChain) serveStaleTimestamps() { + c.staleTimestamps.Store(true) +} + // reorgFrom replaces the canonical chain from block num onwards with a new // fork of the same length. func (c *fakeChain) reorgFrom(num uint64) { diff --git a/ethmonitor/stream_health_test.go b/ethmonitor/stream_health_test.go new file mode 100644 index 00000000..cdc109bd --- /dev/null +++ b/ethmonitor/stream_health_test.go @@ -0,0 +1,256 @@ +package ethmonitor_test + +import ( + "context" + "fmt" + "math/big" + "strings" + "sync" + "testing" + "time" + + "github.com/0xsequence/ethkit/ethmonitor" + "github.com/0xsequence/ethkit/ethmonitor/internal/mocks" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/mock/gomock" +) + +// recordingAlerter captures alerts for inspection. +type recordingAlerter struct { + mu sync.Mutex + alerts []string +} + +func (a *recordingAlerter) Alert(_ context.Context, format string, v ...interface{}) { + a.mu.Lock() + defer a.mu.Unlock() + a.alerts = append(a.alerts, fmt.Sprintf(format, v...)) +} + +func (a *recordingAlerter) list() []string { + a.mu.Lock() + defer a.mu.Unlock() + return append([]string{}, a.alerts...) +} + +func newStreamHealthTestMonitor(t *testing.T, chain *fakeChain, alert *recordingAlerter, retryAfter, healthCheckInterval time.Duration) *ethmonitor.Monitor { + t.Helper() + + provider := &fakeProvider{ + MockRawInterface: mocks.NewMockRawInterface(gomock.NewController(t)), + chain: chain, + streaming: true, + } + + opts := ethmonitor.DefaultOptions + opts.PollingInterval = 20 * time.Millisecond + opts.Timeout = 2 * time.Second + opts.StartBlockNumber = new(big.Int).SetUint64(chain.base) + opts.StreamingErrNumToSwitchToPolling = 2 + opts.StreamingErrorResetInterval = time.Hour + opts.StreamingRetryAfter = retryAfter + // header timestamps are second-granular, so the staleness must be above + // the truncation error; keep it small but > 1s for a fast test. + opts.StreamHealthCheckInterval = healthCheckInterval + opts.StreamHealthCheckStaleness = 2 * time.Second + opts.Alerter = alert + + monitor, err := ethmonitor.NewMonitor(provider, opts) + require.NoError(t, err) + return monitor +} + +// waitFor polls cond until it holds or the deadline passes. +func waitFor(t *testing.T, d time.Duration, msg string, cond func() bool) { + t.Helper() + deadline := time.Now().Add(d) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("timed out waiting for %s", msg) +} + +// TestMonitorStreamHealthCheckFallsBackToPolling stops head notifications +// while the chain keeps producing blocks, reproducing a newHeads stream that +// silently serves nothing while the node's http api stays current. The +// stream health check must drop the subscription, alert, and switch the +// monitor to polling so its head keeps advancing. +func TestMonitorStreamHealthCheckFallsBackToPolling(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + chain := newFakeChain(1000, 1, 5*time.Millisecond) + alerts := &recordingAlerter{} + monitor := newStreamHealthTestMonitor(t, chain, alerts, time.Hour, 100*time.Millisecond) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go chain.produce(ctx, 20*time.Millisecond) + go monitor.Run(ctx) + defer monitor.Stop() + + waitFor(t, 5*time.Second, "streaming mode with an advancing head", func() bool { + return monitor.IsStreamingMode() && monitor.LatestBlockNum().Uint64() > chain.base+2 + }) + assert.Empty(t, alerts.list(), "healthy stream must not alert") + + // the stream stalls: no more heads arrive, the chain keeps producing + chain.mute() + headAtMute := monitor.LatestBlockNum().Uint64() + + waitFor(t, 10*time.Second, "switch to polling after the stream stalls", func() bool { + return !monitor.IsStreamingMode() + }) + + // polling keeps the monitor at the head, which is what the node-gateway + // serves: the incident this guards against left it 23 minutes behind. + waitFor(t, 5*time.Second, "monitor head to keep advancing in polling mode", func() bool { + return monitor.LatestBlockNum().Uint64() >= headAtMute+3 + }) + + stalled := 0 + for _, a := range alerts.list() { + if strings.Contains(a, "stalled") { + stalled++ + } + } + assert.Greater(t, stalled, 0, "a stalled stream must alert") +} + +// TestMonitorStreamHealthCheckZeroIntervalDisablesCheck runs streaming with +// the health check disabled: heads must keep flowing without a panic, and a +// silent stream must not trip a check that is off. +func TestMonitorStreamHealthCheckZeroIntervalDisablesCheck(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + chain := newFakeChain(1000, 1, 5*time.Millisecond) + alerts := &recordingAlerter{} + monitor := newStreamHealthTestMonitor(t, chain, alerts, time.Hour, 0) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go chain.produce(ctx, 20*time.Millisecond) + go monitor.Run(ctx) + defer monitor.Stop() + + waitFor(t, 5*time.Second, "streaming mode with an advancing head", func() bool { + return monitor.IsStreamingMode() && monitor.LatestBlockNum().Uint64() > chain.base+2 + }) + assert.Empty(t, alerts.list(), "the check is off, nothing may alert") + + // the stream goes silent; with the check disabled the monitor must stay + // in streaming mode rather than trip or panic + chain.mute() + time.Sleep(2500 * time.Millisecond) + + assert.True(t, monitor.IsStreamingMode()) + assert.Empty(t, alerts.list()) +} + +// TestMonitorStreamHealthCheckFreshStreamKeepsStreaming is the control: a +// live stream stays in streaming mode and raises no stall alerts. +func TestMonitorStreamHealthCheckFreshStreamKeepsStreaming(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + chain := newFakeChain(1000, 1, 5*time.Millisecond) + alerts := &recordingAlerter{} + monitor := newStreamHealthTestMonitor(t, chain, alerts, time.Hour, 100*time.Millisecond) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go chain.produce(ctx, 20*time.Millisecond) + go monitor.Run(ctx) + defer monitor.Stop() + + waitFor(t, 5*time.Second, "streaming mode with an advancing head", func() bool { + return monitor.IsStreamingMode() && monitor.LatestBlockNum().Uint64() > chain.base+2 + }) + + time.Sleep(1500 * time.Millisecond) + + assert.True(t, monitor.IsStreamingMode()) + for _, a := range alerts.list() { + assert.NotContains(t, a, "stalled") + } +} + +// TestMonitorStreamHealthCheckRecoversToStreaming stalls the stream until the +// monitor falls back to polling, then resumes heads: the polling listener +// must retry streaming and the monitor must return to streaming mode with an +// advancing head. +func TestMonitorStreamHealthCheckRecoversToStreaming(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + chain := newFakeChain(1000, 1, 5*time.Millisecond) + alerts := &recordingAlerter{} + monitor := newStreamHealthTestMonitor(t, chain, alerts, 500*time.Millisecond, 100*time.Millisecond) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go chain.produce(ctx, 20*time.Millisecond) + go monitor.Run(ctx) + defer monitor.Stop() + + waitFor(t, 5*time.Second, "streaming mode with an advancing head", func() bool { + return monitor.IsStreamingMode() && monitor.LatestBlockNum().Uint64() > chain.base+2 + }) + + chain.mute() + waitFor(t, 15*time.Second, "switch to polling after the stream stalls", func() bool { + return !monitor.IsStreamingMode() + }) + + chain.unmute() + waitFor(t, 10*time.Second, "streaming mode with an advancing head after recovery", func() bool { + return monitor.IsStreamingMode() && monitor.LatestBlockNum().Uint64() > chain.head()-50 + }) +} + +// TestMonitorStreamHealthCheckStaleHeadDataTrips runs a stream that keeps +// delivering heads, but every head carries a chain timestamp half an hour +// old while the node's http api serves current blocks: the incident shape. +// The check must drop the subscription and fall back to polling even though +// heads are still arriving. +func TestMonitorStreamHealthCheckStaleHeadDataTrips(t *testing.T) { + if testing.Short() { + t.Skip("Skipping in short mode") + } + + chain := newFakeChain(1000, 1, 5*time.Millisecond) + alerts := &recordingAlerter{} + monitor := newStreamHealthTestMonitor(t, chain, alerts, time.Hour, 100*time.Millisecond) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go chain.produce(ctx, 20*time.Millisecond) + go monitor.Run(ctx) + defer monitor.Stop() + + waitFor(t, 5*time.Second, "streaming mode with an advancing head", func() bool { + return monitor.IsStreamingMode() && monitor.LatestBlockNum().Uint64() > chain.base+2 + }) + + chain.serveStaleTimestamps() + waitFor(t, 10*time.Second, "switch to polling while the stream serves stale data", func() bool { + return !monitor.IsStreamingMode() + }) + + stalled := 0 + for _, a := range alerts.list() { + if strings.Contains(a, "stalled") { + stalled++ + } + } + assert.Greater(t, stalled, 0, "a stream serving stale heads must alert") +}