Kill stalled tier2 segments instead of slow ones

app/tier2.go

@@ -22,7 +22,8 @@ import (
 func NewDefaultTier2Config() *Tier2Config {
 	return &Tier2Config{
 		BlockExecutionTimeout:   3 * time.Minute,
-		SegmentExecutionTimeout: 60 * time.Minute,
+		SegmentExecutionTimeout: 4 * time.Hour,
+		SegmentStallTimeout:     10 * time.Minute,
 	}
 }
 
@@ -36,6 +37,7 @@ type Tier2Config struct {
 	WASMExtensions            wasm.WASMExtensioner
 	BlockExecutionTimeout     time.Duration
 	SegmentExecutionTimeout   time.Duration
+	SegmentStallTimeout       time.Duration
 	TmpDir                    string
 	StoresScratchSpace             string
 	StoresBackend                  string
@@ -100,6 +102,10 @@ func (a *Tier2App) Run() error {
 		opts = append(opts, service.WithSegmentExecutionTimeout(a.config.SegmentExecutionTimeout))
 	}
 
+	if a.config.SegmentStallTimeout != 0 {
+		opts = append(opts, service.WithSegmentStallTimeout(a.config.SegmentStallTimeout))
+	}
+
 	opts = append(opts, service.WithReadinessFunc(a.setIsReady))
 
 	if a.config.TmpDir != "" {

docs/release-notes/change-log.md

@@ -9,6 +9,20 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## Unreleased
+
+### Changed
+
+- Server: tier2 now aborts a segment when it **stops making progress** rather than when it exceeds a fixed time budget. A new stall timeout (10 minutes by default, `WithSegmentStallTimeout`) resets on every block processed, and the pre-existing segment execution timeout (`WithSegmentExecutionTimeout`) is kept only as an absolute backstop, its default raised from 60 minutes to 4 hours.
+
+  The old fixed budget was fatal to expensive-but-healthy workloads: a segment making thousands of `eth_call` per block could take slightly longer than 60 minutes while still advancing block by block, get killed, and — since a killed segment is never cached — have its retry redo the same work and hit the same wall. Such a request could never complete, no matter how many times it reconnected. A stalled segment is still killed promptly, and since a single block is already bounded by the block execution timeout (3 minutes by default), the stall timeout cannot be tripped by one legitimately slow block.
+
+  The `request active for a long time` log gained a `since_last_progress` field, and a segment killed for stalling now reports `request stalled, no block progress` instead of `request active for too long`.
+
+### Fixed
+
+- Server: a tier1 shutdown happening while a request was still in its parallel backprocessing phase was reported to the client as `Internal` instead of `Unavailable`. The `endpoint is shutting down, please reconnect` error is wrapped several times on its way up from the scheduler (`error during init_stores_and_backprocess: run_parallel_process failed: parallel processing run: scheduler run: ...`) and was matched with a pointer comparison, so it never took the `Unavailable` path. Clients now see the correct reconnect signal during a tier1 rollout.
+
 ## 1.20.3
 
 ### Added

metrics/stats.go

@@ -37,7 +37,9 @@ type Stats struct {
 	runningJobs             runningJobs
 	completedJobsStats      map[string]*pbssinternal.ModuleStats
 	uncompressedEgressBytes uint64
-	processedBlocks         uint64
+	// processedBlocks is written from the block-processing goroutine and read from tier2's
+	// segment watchdog goroutine (which uses it as a liveness signal), so it must be atomic.
+	processedBlocks atomic.Uint64
 
 	localProcessedBlockCount  uint64
 	remoteProcessedBlockCount uint64
@@ -278,11 +280,11 @@ func (s *Stats) RecordLastBlockSent(clock *pbsubstreams.Clock) {
 }
 
 func (s *Stats) RecordBlocksProcessed(count uint64) {
-	s.processedBlocks += count
+	s.processedBlocks.Add(count)
 }
 
 func (s *Stats) GetBlocksProcessed() uint64 {
-	return s.processedBlocks
+	return s.processedBlocks.Load()
 }
 
 func (s *Stats) RecordStages(stages []*pbsubstreamsrpc.Stage) {
@@ -774,7 +776,7 @@ func (s *Stats) getZapFields(meter dmetering.Meter) []zap.Field {
 		zap.Uint64("remote_jobs_retried", s.retriedJobs),
 		zap.Uint64("remote_jobs_delayed", s.delayedJobs),
 		zap.Uint64("remote_blocks_processed", s.remoteProcessedBlockCount), // "estimated" from remote ranges
-		zap.Uint64("total_blocks_processed", s.processedBlocks),            // includes remote and local blocks processed in this request, multiplied by execution stages, excludes blocks that were skipped from indexes
+		zap.Uint64("total_blocks_processed", s.processedBlocks.Load()),     // includes remote and local blocks processed in this request, multiplied by execution stages, excludes blocks that were skipped from indexes
 		zap.Uint64("uncompressed_egress_bytes", s.uncompressedEgressBytes),
 		zap.Duration("client_read_average_time_last_5_minutes", s.clientReadTime.Average()),
 		zap.Uint64("last_sent_block_num", s.lastSentBlockNum),

service/error.go

@@ -49,8 +49,10 @@ func toConnectError(ctx context.Context, err error) error {
 	if err, ok := dsession.ToConnectError(err); ok {
 		return err
 	}
-	// special case for context canceled when shutting down
-	if err == errShuttingDown {
+	// special case for context canceled when shutting down. The error is wrapped by the
+	// time it reaches here (ex: `error during init_stores_and_backprocess: ... scheduler
+	// run: %w`), so it must be matched with `errors.Is` and not a pointer comparison.
+	if errors.Is(err, errShuttingDown) {
 		return connect.NewError(connect.CodeUnavailable, err)
 	}
 
@@ -95,7 +97,7 @@ func toConnectError(ctx context.Context, err error) error {
 	}
 
 	// special case for "QuickSave" on shutdown
-	if err == pipeline.ErrShuttingDown {
+	if errors.Is(err, pipeline.ErrShuttingDown) {
 		return connect.NewError(connect.CodeUnavailable, err)
 	}
 

service/error_shutting_down_test.go

@@ -0,0 +1,56 @@
+package service
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"connectrpc.com/connect"
+	"github.com/streamingfast/substreams/pipeline"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestToConnectError_ShuttingDownIsUnavailable documents how a tier1 shutdown
+// reaches the client while a parallel backprocess is in flight.
+//
+// When the tier1 service starts terminating, the scheduler quits with
+// `errShuttingDown`, and that error is wrapped four times as it unwinds:
+//
+//	scheduler run -> parallel processing run -> run_parallel_process failed ->
+//	error during init_stores_and_backprocess
+//
+// It must still be reported as `CodeUnavailable` so that sinks treat it as a
+// "reconnect me" signal rather than a server-side fault.
+func TestToConnectError_ShuttingDownIsUnavailable(t *testing.T) {
+	ctx := context.Background()
+
+	cases := []struct {
+		name string
+		err  error
+	}{
+		{"unwrapped", errShuttingDown},
+		{
+			"wrapped by the backprocess call chain",
+			fmt.Errorf("error during init_stores_and_backprocess: %w",
+				fmt.Errorf("run_parallel_process failed: %w",
+					fmt.Errorf("parallel processing run: %w",
+						fmt.Errorf("scheduler run: %w", errShuttingDown)))),
+		},
+		{"quicksave on shutdown, unwrapped", pipeline.ErrShuttingDown},
+		{
+			"quicksave on shutdown, wrapped",
+			fmt.Errorf("stream terminated: %w", pipeline.ErrShuttingDown),
+		},
+	}
+
+	for _, tt := range cases {
+		t.Run(tt.name, func(t *testing.T) {
+			connectErr := toConnectError(ctx, tt.err)
+			require.Error(t, connectErr)
+
+			assert.Equal(t, connect.CodeUnavailable, connect.CodeOf(connectErr),
+				"shutdown errors must be Unavailable so clients reconnect, got: %s", connectErr)
+		})
+	}
+}

service/options.go

@@ -34,7 +34,9 @@ func WithBlockExecutionTimeout(timeout time.Duration) Option {
 	}
 }
 
-// Tier2 will completely bail out if a segment execution takes longer than the this.
+// Tier2 will completely bail out if a segment execution takes longer than the this. This is an
+// absolute backstop: a segment that keeps making progress is normally bounded by
+// [WithSegmentStallTimeout] instead, so this should stay generous.
 func WithSegmentExecutionTimeout(timeout time.Duration) Option {
 	return func(a anyTierService) {
 		switch s := a.(type) {
@@ -46,6 +48,21 @@ func WithSegmentExecutionTimeout(timeout time.Duration) Option {
 	}
 }
 
+// Tier2 will bail out if a segment stops processing blocks for longer than this. Unlike
+// [WithSegmentExecutionTimeout] the deadline resets on every block processed, so a slow but
+// advancing segment is left alone and only a wedged one is killed. Keep it well above
+// [WithBlockExecutionTimeout], which already bounds a single block.
+func WithSegmentStallTimeout(timeout time.Duration) Option {
+	return func(a anyTierService) {
+		switch s := a.(type) {
+		case *Tier1Service:
+			// not used
+		case *Tier2Service:
+			s.segmentStallTimeout = timeout
+		}
+	}
+}
+
 func WithWASMExtensioner(ext wasm.WASMExtensioner) Option {
 	return func(a anyTierService) {
 		switch s := a.(type) {

service/segment_watchdog.go

@@ -0,0 +1,65 @@
+package service
+
+import (
+	"time"
+
+	"connectrpc.com/connect"
+)
+
+// segmentWatchdog decides when a tier2 segment execution must be aborted. It enforces two
+// independent deadlines:
+//
+//   - a stall deadline, which resets every time the segment reports block progress. This is the
+//     one that normally fires, and it targets segments that are wedged rather than segments that
+//     are merely expensive.
+//   - an absolute deadline measured from the start of the segment, as a backstop for a segment
+//     that keeps inching forward but will never realistically complete.
+//
+// The distinction matters because killing a slow-but-advancing segment is expensive: the segment
+// is never cached, so the retry redoes all of it and races the same clock again. A workload that
+// needs slightly longer than the budget therefore never completes, no matter how many times it is
+// retried.
+//
+// A single block is already bounded by the block execution timeout, so a stall timeout kept well
+// above it cannot be tripped by one legitimately slow block.
+type segmentWatchdog struct {
+	startedAt        time.Time
+	stallTimeout     time.Duration
+	executionTimeout time.Duration
+
+	lastProgressAt  time.Time
+	processedBlocks uint64
+}
+
+func newSegmentWatchdog(startedAt time.Time, processedBlocks uint64, stallTimeout, executionTimeout time.Duration) *segmentWatchdog {
+	return &segmentWatchdog{
+		startedAt:        startedAt,
+		stallTimeout:     stallTimeout,
+		executionTimeout: executionTimeout,
+		lastProgressAt:   startedAt,
+		processedBlocks:  processedBlocks,
+	}
+}
+
+// check records the segment's current block count and returns the error the segment must be
+// canceled with, or nil if it may keep running. A zero timeout disables that deadline.
+func (w *segmentWatchdog) check(now time.Time, processedBlocks uint64) error {
+	if processedBlocks != w.processedBlocks {
+		w.processedBlocks = processedBlocks
+		w.lastProgressAt = now
+	}
+
+	if w.stallTimeout != 0 && now.Sub(w.lastProgressAt) > w.stallTimeout {
+		return connect.NewError(connect.CodeDeadlineExceeded, ErrRequestStalled)
+	}
+
+	if w.executionTimeout != 0 && now.Sub(w.startedAt) > w.executionTimeout {
+		return connect.NewError(connect.CodeDeadlineExceeded, ErrRequestActiveForTooLong)
+	}
+
+	return nil
+}
+
+func (w *segmentWatchdog) sinceLastProgress(now time.Time) time.Duration {
+	return now.Sub(w.lastProgressAt)
+}

service/segment_watchdog_test.go

@@ -0,0 +1,104 @@
+package service
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"connectrpc.com/connect"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+var watchdogStart = time.Date(2026, 7, 31, 19, 46, 57, 0, time.UTC)
+
+// TestSegmentWatchdog_SlowButProgressingSurvives is the case this watchdog exists for: a segment
+// making thousands of `eth_call` per block advances slowly but steadily, and used to be killed by
+// the fixed 60 minute budget. Since a killed segment is never cached, the retry redid the work and
+// hit the same wall, so the request could never complete.
+func TestSegmentWatchdog_SlowButProgressingSurvives(t *testing.T) {
+	w := newSegmentWatchdog(watchdogStart, 0, 10*time.Minute, 4*time.Hour)
+
+	// One block every 5 minutes for 3 hours: far past the old 60 minute budget, but never stalled.
+	var processed uint64
+	for elapsed := 5 * time.Minute; elapsed <= 3*time.Hour; elapsed += 5 * time.Minute {
+		processed++
+		require.NoError(t, w.check(watchdogStart.Add(elapsed), processed),
+			"a segment still making progress must not be killed (elapsed: %s)", elapsed)
+	}
+}
+
+func TestSegmentWatchdog_StalledIsKilled(t *testing.T) {
+	w := newSegmentWatchdog(watchdogStart, 0, 10*time.Minute, 4*time.Hour)
+
+	require.NoError(t, w.check(watchdogStart.Add(5*time.Minute), 12))
+	// No progress since, but still within the stall budget.
+	require.NoError(t, w.check(watchdogStart.Add(14*time.Minute), 12))
+
+	err := w.check(watchdogStart.Add(15*time.Minute+1*time.Second), 12)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ErrRequestStalled)
+	assert.Equal(t, connect.CodeDeadlineExceeded, connect.CodeOf(err))
+}
+
+// TestSegmentWatchdog_ProgressResetsStallDeadline verifies the deadline is measured from the last
+// progress and not from the segment start.
+func TestSegmentWatchdog_ProgressResetsStallDeadline(t *testing.T) {
+	w := newSegmentWatchdog(watchdogStart, 0, 10*time.Minute, 4*time.Hour)
+
+	// Nine minutes of silence, then a block: the stall clock restarts.
+	require.NoError(t, w.check(watchdogStart.Add(9*time.Minute), 0))
+	require.NoError(t, w.check(watchdogStart.Add(9*time.Minute+30*time.Second), 1))
+
+	// Another nine minutes of silence is 18m30s into the segment, but only 9m since progress.
+	require.NoError(t, w.check(watchdogStart.Add(18*time.Minute+30*time.Second), 1))
+	assert.Equal(t, 9*time.Minute, w.sinceLastProgress(watchdogStart.Add(18*time.Minute+30*time.Second)))
+
+	err := w.check(watchdogStart.Add(20*time.Minute), 1)
+	assert.ErrorIs(t, err, ErrRequestStalled)
+}
+
+// TestSegmentWatchdog_AbsoluteBackstop covers a segment that keeps inching forward forever: the
+// stall deadline never fires, so the absolute deadline must.
+func TestSegmentWatchdog_AbsoluteBackstop(t *testing.T) {
+	w := newSegmentWatchdog(watchdogStart, 0, 10*time.Minute, 1*time.Hour)
+
+	var processed uint64
+	var lastErr error
+	for elapsed := 5 * time.Minute; elapsed <= 2*time.Hour; elapsed += 5 * time.Minute {
+		processed++
+		if err := w.check(watchdogStart.Add(elapsed), processed); err != nil {
+			lastErr = err
+			assert.Greater(t, elapsed, 1*time.Hour, "backstop fired before the absolute deadline")
+			break
+		}
+	}
+
+	require.Error(t, lastErr)
+	assert.ErrorIs(t, lastErr, ErrRequestActiveForTooLong)
+	assert.Equal(t, connect.CodeDeadlineExceeded, connect.CodeOf(lastErr))
+}
+
+// TestSegmentWatchdog_StallTakesPrecedence documents that when both deadlines are blown, the stall
+// error is reported: it is the more actionable diagnosis.
+func TestSegmentWatchdog_StallTakesPrecedence(t *testing.T) {
+	w := newSegmentWatchdog(watchdogStart, 0, 10*time.Minute, 1*time.Hour)
+
+	err := w.check(watchdogStart.Add(2*time.Hour), 0)
+	require.Error(t, err)
+	assert.ErrorIs(t, err, ErrRequestStalled)
+}
+
+func TestSegmentWatchdog_ZeroTimeoutsDisableDeadlines(t *testing.T) {
+	w := newSegmentWatchdog(watchdogStart, 0, 0, 0)
+
+	assert.NoError(t, w.check(watchdogStart.Add(72*time.Hour), 0),
+		"both deadlines disabled means the segment is never killed by the watchdog")
+}
+
+// TestSegmentWatchdog_ErrorsAreDistinguishable guards the two abort reasons staying separable,
+// since they call for different operator responses.
+func TestSegmentWatchdog_ErrorsAreDistinguishable(t *testing.T) {
+	assert.False(t, errors.Is(ErrRequestStalled, ErrRequestActiveForTooLong))
+	assert.False(t, errors.Is(ErrRequestActiveForTooLong, ErrRequestStalled))
+}