diff --git a/.github/workflows/test-go-e2e.yml b/.github/workflows/test-go-e2e.yml index 02e64aa8e..d72610ee9 100644 --- a/.github/workflows/test-go-e2e.yml +++ b/.github/workflows/test-go-e2e.yml @@ -75,7 +75,7 @@ jobs: # a primary failure cascade into otherwise unrelated bridge tests. - env: anvil-2chains group: autoclaim - run: "^(TestAutoClaimL1ToL2AllowAll|TestAutoClaimL1ToL2APIApprove|TestAutoClaimL1ToL2BasicFilter|TestAutoClaimL2ToL1AllowAll|TestAutoClaimL2ToL2AllowAll)$" + run: "^(TestAutoClaimL1ToL2AllowAll|TestAutoClaimL1ToL2APIApprove|TestAutoClaimL1ToL2BasicFilter|TestAutoClaimL2ToL1AllowAll|TestAutoClaimL2ToL2AllowAll|TestAutoClaimL1ToL2FailingDestinationIsolated)$" - env: anvil-2chains group: bridge run: "^(TestJustBridge|TestBridgeL2ToL2|TestBridgeTrackerL1ToL2)$" diff --git a/autoclaim/bridgedetector/destination_backoff.go b/autoclaim/bridgedetector/destination_backoff.go new file mode 100644 index 000000000..1281c30b3 --- /dev/null +++ b/autoclaim/bridgedetector/destination_backoff.go @@ -0,0 +1,134 @@ +package bridgedetector + +import ( + "errors" + "net/http" + "time" + + "github.com/agglayer/aggkit/autoclaim/metrics" + aggkitcommon "github.com/agglayer/aggkit/common" + "github.com/ethereum/go-ethereum/rpc" +) + +const ( + destinationBackoffFactor = 2.0 + destinationBackoffMax = 2 * time.Minute + destinationBackoffJitter = 0.2 +) + +// destinationBackoff tracks, per destination network, a capped exponential back-off so that a failing +// destination is retried after a growing pause instead of on every poll. The first window equals the +// initial delay (the detector poll period) and doubles up to the max; a success resets it. +// +// The window is purely computed from the number of consecutive failures: a server-provided Retry-After +// hint is NOT honoured, because go-ethereum's rpc.HTTPError does not expose the response headers. +// +// It is not goroutine-safe: each detector owns one instance and only touches it from its poll goroutine. +// The type is shared by the L1-to-L2 and L2-to-Lx detectors; the instances are not. +type destinationBackoff struct { + initial, max time.Duration + jitter float64 + randFloat func() float64 // nil = default + entries map[uint32]*destinationBackoffEntry +} + +type destinationBackoffEntry struct { + backoff *aggkitcommon.ExponentialBackoff + until time.Time + lastErr error +} + +// newDestinationBackoff creates a tracker whose first window is initial, grows by destinationBackoffFactor +// and is capped at maxWindow, with the given symmetric relative jitter. Out-of-range arguments are sanitised +// (initial <= 0 becomes one second, maxWindow < initial becomes initial, jitter outside [0, 1) becomes 0). +func newDestinationBackoff(initial, maxWindow time.Duration, jitter float64) *destinationBackoff { + if initial <= 0 { + initial = time.Second + } + if maxWindow < initial { + maxWindow = initial + } + if jitter < 0 || jitter >= 1 { + jitter = 0 + } + return &destinationBackoff{ + initial: initial, + max: maxWindow, + jitter: jitter, + entries: make(map[uint32]*destinationBackoffEntry), + } +} + +// shouldSkip reports whether dest is inside its back-off window at now. +func (b *destinationBackoff) shouldSkip(dest uint32, now time.Time) bool { + entry, ok := b.entries[dest] + return ok && now.Before(entry.until) +} + +// skipUntil returns the end of dest's current window (zero if none), for the Debug log. +func (b *destinationBackoff) skipUntil(dest uint32) time.Time { + if entry, ok := b.entries[dest]; ok { + return entry.until + } + return time.Time{} +} + +// recordFailure opens the next window: window = entry.backoff.Next(); until = now + window. +// It returns the window and the number of consecutive failures (the attempt) recorded for dest. +func (b *destinationBackoff) recordFailure( + dest uint32, err error, now time.Time, +) (window time.Duration, attempt int) { + entry, ok := b.entries[dest] + if !ok { + var opts []aggkitcommon.ExponentialBackoffOption + if b.randFloat != nil { + opts = append(opts, aggkitcommon.WithBackoffRandFloat(b.randFloat)) + } + // The configuration is sanitised by newDestinationBackoff, so it is always valid. + backoff, _ := aggkitcommon.NewExponentialBackoff(aggkitcommon.ExponentialBackoffConfig{ + Initial: b.initial, + Max: b.max, + Factor: destinationBackoffFactor, + Jitter: b.jitter, + }, opts...) + entry = &destinationBackoffEntry{backoff: backoff} + b.entries[dest] = entry + } + window = entry.backoff.Next() + entry.until = now.Add(window) + entry.lastErr = err + return window, entry.backoff.Attempts() +} + +// recordSuccess deletes dest's entry (full reset). +func (b *destinationBackoff) recordSuccess(dest uint32) { + delete(b.entries, dest) +} + +// stalledCount returns the number of destinations with an entry (failing and not yet recovered). +func (b *destinationBackoff) stalledCount() int { + return len(b.entries) +} + +// recordFailureAndWarn records a destination failure, counts it in the destination error metric and logs +// the one Warn per back-off window (an HTTP 429 gets its own wording). detectorName is the detector name +// used in the log text ("l1-to-l2" or "l2-to-lx"). +func (b *destinationBackoff) recordFailureAndWarn( + detector metrics.Detector, + detectorName string, + dest uint32, + err error, + now time.Time, + warnf func(format string, args ...interface{}), +) { + window, attempt := b.recordFailure(dest, err, now) + metrics.IncDestinationError(detector, dest) + var httpErr rpc.HTTPError + if errors.As(err, &httpErr) && httpErr.StatusCode == http.StatusTooManyRequests { + warnf("autoclaim %s bridge detector: detection for destination %d paused for %s "+ + "after rate limiting (HTTP 429, attempt %d): %v", detectorName, dest, window, attempt, err) + return + } + warnf("autoclaim %s bridge detector: detection for destination %d paused for %s "+ + "after error (attempt %d): %v", detectorName, dest, window, attempt, err) +} diff --git a/autoclaim/bridgedetector/destination_backoff_test.go b/autoclaim/bridgedetector/destination_backoff_test.go new file mode 100644 index 000000000..35f084959 --- /dev/null +++ b/autoclaim/bridgedetector/destination_backoff_test.go @@ -0,0 +1,302 @@ +package bridgedetector + +import ( + "context" + "errors" + "fmt" + "net/http" + "strings" + "testing" + "time" + + autoclaimmetrics "github.com/agglayer/aggkit/autoclaim/metrics" + autoclaimtypes "github.com/agglayer/aggkit/autoclaim/types" + "github.com/agglayer/aggkit/bridgesync" + "github.com/agglayer/aggkit/prometheus" + "github.com/ethereum/go-ethereum/rpc" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" +) + +// capturingLogger records formatted Warn and Debug lines; other levels are discarded. +type capturingLogger struct { + warns []string + debugs []string +} + +func (l *capturingLogger) Panicf(string, ...interface{}) {} +func (l *capturingLogger) Fatalf(string, ...interface{}) {} +func (l *capturingLogger) Info(...interface{}) {} +func (l *capturingLogger) Infof(string, ...interface{}) {} +func (l *capturingLogger) Error(...interface{}) {} +func (l *capturingLogger) Errorf(string, ...interface{}) {} +func (l *capturingLogger) Warn(args ...interface{}) { l.warns = append(l.warns, fmt.Sprint(args...)) } +func (l *capturingLogger) Warnf(format string, args ...interface{}) { + l.warns = append(l.warns, fmt.Sprintf(format, args...)) +} +func (l *capturingLogger) Debug(args ...interface{}) { + l.debugs = append(l.debugs, fmt.Sprint(args...)) +} +func (l *capturingLogger) Debugf(format string, args ...interface{}) { + l.debugs = append(l.debugs, fmt.Sprintf(format, args...)) +} + +func TestDestinationBackoff_Curve(t *testing.T) { + b := newDestinationBackoff(time.Second, 2*time.Minute, 0) + now := testNow + failure := errors.New("boom") + + require.False(t, b.shouldSkip(7, now)) + require.True(t, b.skipUntil(7).IsZero()) + + expected := []time.Duration{ + time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second, + 32 * time.Second, 64 * time.Second, 2 * time.Minute, 2 * time.Minute, + } + for i, want := range expected { + window, attempt := b.recordFailure(7, failure, now) + require.Equal(t, want, window) + require.Equal(t, i+1, attempt) + require.Equal(t, now.Add(want), b.skipUntil(7)) + require.True(t, b.shouldSkip(7, now)) + require.True(t, b.shouldSkip(7, now.Add(want-time.Nanosecond))) + require.False(t, b.shouldSkip(7, now.Add(want))) + } + require.Equal(t, 1, b.stalledCount()) + + b.recordSuccess(7) + require.Equal(t, 0, b.stalledCount()) + window, attempt := b.recordFailure(7, failure, now) + require.Equal(t, time.Second, window) + require.Equal(t, 1, attempt) +} + +func TestDestinationBackoff_Sanitises(t *testing.T) { + b := newDestinationBackoff(5*time.Minute, time.Minute, 2) + window, _ := b.recordFailure(1, errors.New("x"), testNow) + require.Equal(t, 5*time.Minute, window) +} + +func TestNewL1ToL2_PollPeriodDrivesInitialBackoff(t *testing.T) { + failing := &fakeClaimer{ + target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer10ID, DestinationNetwork: 10}, + claimErr: errors.New("boom"), + } + source := &fakeBridgeSource{ + lastProcessedBlock: 10, + found: true, + bridgesByRange: map[blockRange][]bridgesync.Bridge{ + {from: 0, to: 10}: {makeSyncBridge(1, autoclaimtypes.L1OriginNetwork, 10, 1, 0)}, + }, + } + detector := newTestDetector(t, source, newMemoryCursorStore(), newFakeRegistry(failing), + WithBlockWindow(11), WithPollPeriod(5*time.Second)) + + _, err := detector.PollOnce(context.Background()) + require.Error(t, err) + // Default construction: initial = poll period, jitter +-20%. + window := detector.backoff.skipUntil(10).Sub(testNow) + require.GreaterOrEqual(t, window, 4*time.Second) + require.LessOrEqual(t, window, 6*time.Second) + require.Equal(t, 5*time.Second, detector.backoff.initial) +} + +type backoffFixture struct { + detector *L1ToL2 + failing *fakeClaimer + healthyY *fakeClaimer + healthyZ *fakeClaimer + log *capturingLogger + clock *time.Time +} + +func newBackoffFixture(t *testing.T, failure error) *backoffFixture { + t.Helper() + bridges := []bridgesync.Bridge{ + makeSyncBridge(1, autoclaimtypes.L1OriginNetwork, 10, 2, 0), + makeSyncBridge(2, autoclaimtypes.L1OriginNetwork, 11, 3, 0), + makeSyncBridge(3, autoclaimtypes.L1OriginNetwork, 12, 4, 0), + } + source := &fakeBridgeSource{ + lastProcessedBlock: 10, + found: true, + bridgesByRange: map[blockRange][]bridgesync.Bridge{{from: 1, to: 10}: bridges}, + } + f := &backoffFixture{ + failing: &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer10ID, DestinationNetwork: 10}, claimErr: failure}, + healthyY: &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer11ID, DestinationNetwork: 11}}, + healthyZ: &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: "claimer-12", DestinationNetwork: 12}}, + log: &capturingLogger{}, + } + clock := testNow + f.clock = &clock + store := newMemoryCursorStore() + f.detector = newTestDetector(t, source, store, newFakeRegistry(f.failing, f.healthyY, f.healthyZ), + WithStartBlock(0), WithBlockWindow(10), WithOverlapBlocks(0), WithLogger(f.log), + WithNow(func() time.Time { return *f.clock })) + f.detector.backoff = newDestinationBackoff(time.Second, 2*time.Minute, 0) + seeded := autoclaimtypes.BridgeCursor{FromBlock: 0, ToBlock: 0, BlockNum: 0} + for _, dest := range []uint32{10, 11, 12} { + store.cursors[f.detector.cursorNameForDestination(dest)] = seeded + } + return f +} + +func TestL1ToL2_BackoffSkipsFailingDestination(t *testing.T) { + ctx := context.Background() + f := newBackoffFixture(t, errors.New("boom")) + + result, err := f.detector.PollOnce(ctx) + require.Error(t, err) + require.Equal(t, []uint32{10}, result.FailedDestinations) + require.Empty(t, result.BackedOffDestinations) + require.Len(t, f.failing.claimChecks, 1) + require.Len(t, f.log.warns, 1) + + // Same clock: destination 10 is skipped, the others are unaffected. + result, err = f.detector.PollOnce(ctx) + require.NoError(t, err) + require.Equal(t, []uint32{10}, result.BackedOffDestinations) + require.Empty(t, result.FailedDestinations) + require.Len(t, f.failing.claimChecks, 1, "no IsClaimed call while backed off") + require.Len(t, f.healthyY.claimChecks, 1, "healthy destination is not re-evaluated once its cursor advanced") + require.Len(t, f.log.warns, 1) + require.Len(t, f.log.debugs, 1) + require.Contains(t, f.log.debugs[0], "skipping destination 10, backed off until") + + // Just before the window ends it is still skipped; at the end it is retried. + *f.clock = testNow.Add(time.Second - time.Nanosecond) + result, err = f.detector.PollOnce(ctx) + require.NoError(t, err) + require.Equal(t, []uint32{10}, result.BackedOffDestinations) + + *f.clock = testNow.Add(time.Second) + result, err = f.detector.PollOnce(ctx) + require.Error(t, err) + require.Empty(t, result.BackedOffDestinations) + require.Equal(t, []uint32{10}, result.FailedDestinations) + require.Len(t, f.failing.claimChecks, 2) + require.Len(t, f.log.warns, 2) +} + +func TestL1ToL2_BackoffWindowsGrowAndWarnOncePerWindow(t *testing.T) { + ctx := context.Background() + f := newBackoffFixture(t, errors.New("boom")) + + windows := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second} + for i, window := range windows { + _, err := f.detector.PollOnce(ctx) + require.Error(t, err) + require.Len(t, f.log.warns, i+1) + require.Contains(t, f.log.warns[i], "paused for "+window.String()) + require.Contains(t, f.log.warns[i], fmt.Sprintf("after error (attempt %d)", i+1)) + require.Equal(t, f.clock.Add(window), f.detector.backoff.skipUntil(10)) + + // Several polls inside the window add no Warn. + for range 3 { + result, err := f.detector.PollOnce(ctx) + require.NoError(t, err) + require.Equal(t, []uint32{10}, result.BackedOffDestinations) + } + require.Len(t, f.log.warns, i+1) + *f.clock = f.clock.Add(window) + } +} + +func TestL1ToL2_BackoffSuccessResets(t *testing.T) { + ctx := context.Background() + f := newBackoffFixture(t, errors.New("boom")) + + _, err := f.detector.PollOnce(ctx) + require.Error(t, err) + *f.clock = f.clock.Add(time.Second) + _, err = f.detector.PollOnce(ctx) + require.Error(t, err) + require.Equal(t, 1, f.detector.backoff.stalledCount()) + + f.failing.claimErr = nil + *f.clock = f.clock.Add(2 * time.Second) + result, err := f.detector.PollOnce(ctx) + require.NoError(t, err) + require.Empty(t, result.FailedDestinations) + require.Equal(t, 0, f.detector.backoff.stalledCount()) + require.Len(t, f.log.warns, 2) + + // A new failure starts again from the initial window. + f.failing.claimErr = errors.New("boom again") + f.detector.backoff.recordFailure(11, nil, *f.clock) // unrelated entry must not matter + f.detector.backoff.recordSuccess(11) + window, attempt := f.detector.backoff.recordFailure(10, f.failing.claimErr, *f.clock) + require.Equal(t, time.Second, window) + require.Equal(t, 1, attempt) +} + +func TestL1ToL2_BackoffRateLimitedUsesHTTP429Wording(t *testing.T) { + ctx := context.Background() + f := newBackoffFixture(t, rpc.HTTPError{StatusCode: http.StatusTooManyRequests, Status: "429 Too Many Requests"}) + + for i, window := range []time.Duration{time.Second, 2 * time.Second, 4 * time.Second} { + _, err := f.detector.PollOnce(ctx) + require.Error(t, err) + require.Len(t, f.log.warns, i+1) + require.Contains(t, f.log.warns[i], "paused for "+window.String()) + require.Contains(t, f.log.warns[i], fmt.Sprintf("after rate limiting (HTTP 429, attempt %d)", i+1)) + *f.clock = f.clock.Add(window) + } +} + +func TestL1ToL2_BackoffOtherHTTPErrorUsesGenericWording(t *testing.T) { + ctx := context.Background() + f := newBackoffFixture(t, rpc.HTTPError{StatusCode: http.StatusInternalServerError, Status: "500"}) + + _, err := f.detector.PollOnce(ctx) + require.Error(t, err) + require.Len(t, f.log.warns, 1) + require.True(t, strings.Contains(f.log.warns[0], "paused for 1s after error (attempt 1)")) +} + +func TestL1ToL2_BackoffNotUpdatedWhenContextCancelled(t *testing.T) { + f := newBackoffFixture(t, errors.New("boom")) + ctx, cancel := context.WithCancel(context.Background()) + f.failing.claimErr = errors.New("boom") + cancel() + + result, err := f.detector.PollOnce(ctx) + require.Error(t, err) + require.Equal(t, []uint32{10}, result.FailedDestinations) + require.Equal(t, 0, f.detector.backoff.stalledCount()) + require.Empty(t, f.log.warns) +} + +func TestL1ToL2_BackoffMetrics(t *testing.T) { + prometheus.Init() + autoclaimmetrics.Register() + ctx := context.Background() + f := newBackoffFixture(t, errors.New("boom")) + + gauge, ok := prometheus.Gauge("l1_to_l2_stalled_destinations") // the wrapper keys series by bare name + require.True(t, ok) + vec, ok := prometheus.CounterVec("detector_destination_errors_total") + require.True(t, ok) + errCounter := vec.WithLabelValues(string(autoclaimmetrics.DetectorL1ToL2), "10") + before := testutil.ToFloat64(errCounter) + + _, err := f.detector.PollOnce(ctx) + require.Error(t, err) + require.Equal(t, 1.0, testutil.ToFloat64(gauge)) + require.Equal(t, before+1, testutil.ToFloat64(errCounter)) + + // Polls inside the window neither add errors nor change the gauge. + _, err = f.detector.PollOnce(ctx) + require.NoError(t, err) + require.Equal(t, 1.0, testutil.ToFloat64(gauge)) + require.Equal(t, before+1, testutil.ToFloat64(errCounter)) + + // Recovery brings the gauge back to 0. + f.failing.claimErr = nil + *f.clock = f.clock.Add(time.Second) + _, err = f.detector.PollOnce(ctx) + require.NoError(t, err) + require.Equal(t, 0.0, testutil.ToFloat64(gauge)) + require.Equal(t, before+1, testutil.ToFloat64(errCounter)) +} diff --git a/autoclaim/bridgedetector/l1_to_l2.go b/autoclaim/bridgedetector/l1_to_l2.go index 9b3ebe5cb..5c73896a1 100644 --- a/autoclaim/bridgedetector/l1_to_l2.go +++ b/autoclaim/bridgedetector/l1_to_l2.go @@ -2,10 +2,12 @@ package bridgedetector import ( "context" + "errors" "fmt" "sort" "time" + "github.com/agglayer/aggkit/autoclaim/metrics" autoclaimtypes "github.com/agglayer/aggkit/autoclaim/types" aggkitcommon "github.com/agglayer/aggkit/common" ) @@ -109,8 +111,20 @@ type PollResult struct { MatchedBridgeCount int EnqueuedBridgeCount int IgnoredBridgeCount int + // AlreadyClaimedCount counts bridges skipped because the target network already claimed them. They + // are not part of IgnoredBridgeCount, which only counts bridges to unknown destinations. + AlreadyClaimedCount int SkippedBridgeCount int - CursorAdvanced bool + // DeferredBridgeCount counts bridges of a destination that failed earlier in this poll: they were + // not evaluated and will be re-evaluated on the next poll. + DeferredBridgeCount int + // FailedDestinations lists, in ascending order, the destinations whose IsClaimed or Enqueue failed + // this poll. + FailedDestinations []uint32 + // BackedOffDestinations lists, in ascending order, the destinations excluded from this poll because + // they are inside a back-off window after earlier failures. + BackedOffDestinations []uint32 + CursorAdvanced bool } // L1ToL2 is the bridge detector that discovers L1-initiated bridge exits and routes them to destination claimers. @@ -127,6 +141,7 @@ type L1ToL2 struct { etrogL1UpgradeBlock uint64 now func() time.Time log aggkitcommon.Logger + backoff *destinationBackoff } type destinationCursorState struct { @@ -136,6 +151,8 @@ type destinationCursorState struct { cursorFound bool fromBlock uint64 nextCursor autoclaimtypes.BridgeCursor + // err is the first IsClaimed/Enqueue failure for this destination in the current poll (nil = healthy). + err error } // NewL1ToL2 creates an L1-to-L2 Auto Claim bridge detector. @@ -172,6 +189,7 @@ func NewL1ToL2( for _, option := range options { option(detector) } + detector.backoff = newDestinationBackoff(detector.pollPeriod, destinationBackoffMax, destinationBackoffJitter) return detector, nil } @@ -208,6 +226,12 @@ func (w *L1ToL2) Start(ctx context.Context) { // cursor does not roll back or gate a different group's already-persisted cursor. A single map[RequestKey] // deduplication set spans the whole poll (not per group), which stays correct because every RequestKey // embeds its destination and each destination belongs to exactly one group. +// +// Failures are isolated per destination: when IsClaimed or Enqueue fails for a destination, that +// destination's remaining bridges in the window are deferred (not evaluated, not marked seen) and its +// cursor is not saved, so the next poll re-evaluates them from the persisted cursor. The other +// destinations of the same group keep processing and persist their cursors. All errors are returned +// joined, and PollResult.FailedDestinations lists the failed destinations. func (w *L1ToL2) PollOnce(ctx context.Context) (*PollResult, error) { if !w.enabled { return &PollResult{}, nil @@ -221,11 +245,14 @@ func (w *L1ToL2) PollOnce(ctx context.Context) (*PollResult, error) { return &PollResult{}, nil } - states, err := w.destinationCursorStates(ctx, lastProcessedBlock) + states, backedOff, err := w.destinationCursorStates(ctx, lastProcessedBlock) if err != nil { return nil, err } - result := &PollResult{LastProcessedBlock: lastProcessedBlock} + result := &PollResult{LastProcessedBlock: lastProcessedBlock, BackedOffDestinations: backedOff} + // queried holds the destinations of the groups whose GetBridges succeeded. + queried := make(map[uint32]*destinationCursorState) + defer func() { w.updateBackoff(ctx, states, queried) }() if len(states) == 0 { return result, nil } @@ -233,7 +260,7 @@ func (w *L1ToL2) PollOnce(ctx context.Context) (*PollResult, error) { groups := groupStatesByFromBlock(states) seen := make(map[autoclaimtypes.RequestKey]struct{}) - var firstErr error + var errs []error firstWindow := true for _, fromBlock := range orderedFromBlocks(groups) { // A fromBlock beyond lastProcessedBlock means that group's destination(s) are already @@ -261,12 +288,13 @@ func (w *L1ToL2) PollOnce(ctx context.Context) (*PollResult, error) { bridges, err := w.bridgeSource.GetBridges(ctx, fromBlock, toBlock) if err != nil { - if firstErr == nil { - firstErr = fmt.Errorf("get l1 bridges from %d to %d: %w", fromBlock, toBlock, err) - } + errs = append(errs, fmt.Errorf("get l1 bridges from %d to %d: %w", fromBlock, toBlock, err)) continue } result.BridgeCount += len(bridges) + for destination, state := range groupStates { + queried[destination] = state + } // Every bridge exit returned by l1bridgesync was initiated on L1, so there is no // bridge-origin filter to apply here. The claim identity (and the request key) is keyed on @@ -274,30 +302,25 @@ func (w *L1ToL2) PollOnce(ctx context.Context) (*PollResult, error) { // exit.OriginNetwork (the bridged token's origin network, which can be any network for a // wrapped token). A bridge whose destination is not part of this group (e.g. its window // overlaps another group's) is ignored here and left for the group that owns it. - groupFailed := false for _, bridge := range bridges { exit := autoclaimtypes.NewBridgeExitFromSyncWithEtrog(bridge, w.etrogL1UpgradeBlock) if err := w.processBridge(ctx, exit, groupStates, seen, result); err != nil { - if firstErr == nil { - firstErr = err - } - groupFailed = true - break + errs = append(errs, fmt.Errorf("destination %d: %w", exit.DestinationNetwork, err)) } } - if groupFailed { - continue - } if err := w.saveCursors(ctx, groupStates, result); err != nil { - if firstErr == nil { - firstErr = err - } - continue + errs = append(errs, err) + } + } + + for _, state := range orderedStates(states) { + if state.err != nil { + result.FailedDestinations = append(result.FailedDestinations, state.claimer.Target().DestinationNetwork) } } - return result, firstErr + return result, errors.Join(errs...) } // processBridge evaluates a single bridge exit and enqueues it for claiming when appropriate. @@ -305,6 +328,10 @@ func (w *L1ToL2) PollOnce(ctx context.Context) (*PollResult, error) { // bridge whose destination is owned by a different group's window (e.g. two windows overlap) is // tallied via the same unknown-destination path as a truly unknown destination: it is ignored here // and left for the group that owns it to enqueue exactly once. +// A destination that already failed in this poll (state.err != nil) defers its later bridges: they are +// counted in result.DeferredBridgeCount, neither marked seen nor used to advance the cursor, and +// nil is returned. The first IsClaimed/Enqueue failure of a destination is recorded in state.err and +// returned; it never affects other destinations. // It updates result counters and the seen deduplication map in place. func (w *L1ToL2) processBridge( ctx context.Context, @@ -323,6 +350,11 @@ func (w *L1ToL2) processBridge( return nil } + if state.err != nil { + result.DeferredBridgeCount++ + return nil + } + state.nextCursor = maxCursorPosition(state.nextCursor, exit.BlockNum, exit.BlockPos) // The source network is always L1 (network 0) for L1-initiated exits, independent of the @@ -337,16 +369,21 @@ func (w *L1ToL2) processBridge( claimed, err := state.claimer.IsClaimed(ctx, exit) if err != nil { - return fmt.Errorf("check l1 bridge %s target claim state: %w", key, err) + state.err = fmt.Errorf("check l1 bridge %s target claim state: %w", key, err) + return state.err } if claimed { - result.IgnoredBridgeCount++ + result.AlreadyClaimedCount++ + metrics.IncSkippedAlreadyClaimed(metrics.DetectorL1ToL2, exit.DestinationNetwork) + w.logDebugf("autoclaim %s bridge detector: skipped bridge %s (destination %d, global index %s): "+ + "already claimed on target", metrics.DetectorL1ToL2, key, exit.DestinationNetwork, globalIndexString(exit)) return nil } result.MatchedBridgeCount++ if err := state.claimer.Enqueue(ctx, exit); err != nil { - return fmt.Errorf("enqueue l1 bridge %s to claimer %s: %w", key, state.claimer.Target().ID, err) + state.err = fmt.Errorf("enqueue l1 bridge %s to claimer %s: %w", key, state.claimer.Target().ID, err) + return state.err } result.EnqueuedBridgeCount++ return nil @@ -370,13 +407,18 @@ func bridgeFilteredByPosition(exit autoclaimtypes.BridgeExit, state *destination // saveCursors persists the advanced cursor for every destination state in states, which is scoped to // a single window group by the caller. A destination whose cursor is saved here keeps that write even // if a save for a different group fails later in the same poll: persistence is per-group, never rolled -// back or gated by another group's outcome. +// back or gated by another group's outcome. Destinations that failed during this poll (state.err != nil) +// are skipped, so their persisted cursor stays unchanged and their bridges are re-evaluated next poll. +// A SaveBridgeCursor error stops the loop and is returned. func (w *L1ToL2) saveCursors( ctx context.Context, states map[uint32]*destinationCursorState, result *PollResult, ) error { for _, state := range orderedStates(states) { + if state.err != nil { + continue + } if err := w.cursorStore.SaveBridgeCursor(ctx, state.cursorName, state.nextCursor, w.now()); err != nil { return fmt.Errorf("save autoclaim l1-to-l2 bridge detector cursor %s: %w", state.cursorName, err) } @@ -385,24 +427,58 @@ func (w *L1ToL2) saveCursors( return nil } +// updateBackoff feeds the outcome of a poll into the per-destination back-off tracker: a destination of a +// group whose GetBridges succeeded and that did not fail is reset, and a destination that failed opens its +// next back-off window (one Warn per window). When ctx is done the failure may be a cancellation artefact, +// so the tracker is left untouched. +func (w *L1ToL2) updateBackoff( + ctx context.Context, + states map[uint32]*destinationCursorState, + queried map[uint32]*destinationCursorState, +) { + defer func() { metrics.SetStalledDestinations(metrics.DetectorL1ToL2, w.backoff.stalledCount()) }() + for _, state := range orderedStates(states) { + dest := state.claimer.Target().DestinationNetwork + if state.err == nil { + if _, ok := queried[dest]; ok { + w.backoff.recordSuccess(dest) + } + continue + } + if ctx.Err() != nil { + continue + } + w.backoff.recordFailureAndWarn(metrics.DetectorL1ToL2, "l1-to-l2", dest, state.err, w.now(), w.logWarnf) + } +} + +// destinationCursorStates builds the cursor state of every registered destination that is not inside a +// back-off window. The skipped destinations are returned, in ascending order, as the second value. func (w *L1ToL2) destinationCursorStates( ctx context.Context, lastProcessedBlock uint64, -) (map[uint32]*destinationCursorState, error) { +) (map[uint32]*destinationCursorState, []uint32, error) { claimers, err := w.registry.Claimers(ctx) if err != nil { - return nil, fmt.Errorf("list autoclaim l1-to-l2 bridge detector claimers: %w", err) + return nil, nil, fmt.Errorf("list autoclaim l1-to-l2 bridge detector claimers: %w", err) } states := make(map[uint32]*destinationCursorState, len(claimers)) + var backedOff []uint32 for _, runtimeClaimer := range claimers { if runtimeClaimer == nil { - return nil, fmt.Errorf("autoclaim l1-to-l2 bridge detector registry returned nil claimer") + return nil, nil, fmt.Errorf("autoclaim l1-to-l2 bridge detector registry returned nil claimer") } target := runtimeClaimer.Target() + if w.backoff.shouldSkip(target.DestinationNetwork, w.now()) { + w.logDebugf("autoclaim l1-to-l2 bridge detector: skipping destination %d, backed off until %s", + target.DestinationNetwork, w.backoff.skipUntil(target.DestinationNetwork)) + backedOff = append(backedOff, target.DestinationNetwork) + continue + } cursorName := w.cursorNameForDestination(target.DestinationNetwork) cursor, cursorFound, err := w.cursorStore.GetBridgeCursor(ctx, cursorName) if err != nil { - return nil, fmt.Errorf("get autoclaim l1-to-l2 bridge detector cursor %s: %w", cursorName, err) + return nil, nil, fmt.Errorf("get autoclaim l1-to-l2 bridge detector cursor %s: %w", cursorName, err) } states[target.DestinationNetwork] = &destinationCursorState{ claimer: runtimeClaimer, @@ -412,7 +488,8 @@ func (w *L1ToL2) destinationCursorStates( fromBlock: w.nextFromBlock(cursor, cursorFound, lastProcessedBlock), } } - return states, nil + sort.Slice(backedOff, func(i, j int) bool { return backedOff[i] < backedOff[j] }) + return states, backedOff, nil } func (w *L1ToL2) cursorNameForDestination(destinationNetwork uint32) string { @@ -497,6 +574,18 @@ func (w *L1ToL2) logErrorf(format string, args ...interface{}) { } } +func (w *L1ToL2) logWarnf(format string, args ...interface{}) { + if w.log != nil { + w.log.Warnf(format, args...) + } +} + +func (w *L1ToL2) logDebugf(format string, args ...interface{}) { + if w.log != nil { + w.log.Debugf(format, args...) + } +} + func minUint64(a, b uint64) uint64 { if a < b { return a @@ -522,3 +611,11 @@ func maxCursorPosition( } return cursor } + +// globalIndexString renders the exit's global index for logs, or "unknown" when it is not set. +func globalIndexString(exit autoclaimtypes.BridgeExit) string { + if exit.GlobalIndex == nil { + return "unknown" + } + return exit.GlobalIndex.String() +} diff --git a/autoclaim/bridgedetector/l1_to_l2_test.go b/autoclaim/bridgedetector/l1_to_l2_test.go index 65d24d16e..55933e06a 100644 --- a/autoclaim/bridgedetector/l1_to_l2_test.go +++ b/autoclaim/bridgedetector/l1_to_l2_test.go @@ -9,11 +9,14 @@ import ( "testing" "time" + autoclaimmetrics "github.com/agglayer/aggkit/autoclaim/metrics" autoclaimtypes "github.com/agglayer/aggkit/autoclaim/types" "github.com/agglayer/aggkit/bridgesync" bridgesynctypes "github.com/agglayer/aggkit/bridgesync/types" aggkitcommon "github.com/agglayer/aggkit/common" + "github.com/agglayer/aggkit/prometheus" "github.com/ethereum/go-ethereum/common" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/require" ) @@ -277,12 +280,27 @@ func TestPollOnceIgnoresAlreadyClaimedBridgeBeforeEnqueue(t *testing.T) { target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer10ID, DestinationNetwork: 10}, claimed: true, } - detector := newTestDetector(t, source, store, newFakeRegistry(claimer), WithBlockWindow(11)) + log := &capturingLogger{} + detector := newTestDetector(t, source, store, newFakeRegistry(claimer), WithBlockWindow(11), WithLogger(log)) + + prometheus.Init() + autoclaimmetrics.Register() + vec, ok := prometheus.CounterVec("detector_skipped_already_claimed_total") + require.True(t, ok) + skipped := vec.WithLabelValues(string(autoclaimmetrics.DetectorL1ToL2), "10") + before := testutil.ToFloat64(skipped) result, err := detector.PollOnce(ctx) require.NoError(t, err) require.Equal(t, 0, result.EnqueuedBridgeCount) - require.Equal(t, 1, result.IgnoredBridgeCount) + require.Equal(t, 0, result.IgnoredBridgeCount) + require.Equal(t, 1, result.AlreadyClaimedCount) + require.Equal(t, before+1, testutil.ToFloat64(skipped)) + require.Len(t, claimer.claimChecks, 1) + require.Contains(t, log.debugs, fmt.Sprintf( + "autoclaim l1_to_l2 bridge detector: skipped bridge 0:10:1 (destination 10, global index %s): "+ + "already claimed on target", claimer.claimChecks[0].GlobalIndex.String())) + require.Empty(t, log.warns) require.True(t, result.CursorAdvanced) require.Empty(t, claimer.enqueued) require.Len(t, claimer.claimChecks, 1) @@ -656,6 +674,154 @@ func TestL1ToL2_IgnoresBridgeOutsideOwnWindowWithoutDoubleEnqueue(t *testing.T) require.Len(t, claimerB.enqueued, 1, "destination B must receive exactly one enqueue, not one per overlapping window") } +// TestL1ToL2_FailingDestinationDoesNotBlockOthersInGroup proves that an IsClaimed or Enqueue failure +// for one destination of a window group neither blocks the other destinations of that group nor lets +// the failed destination's cursor advance past a bridge that was not enqueued. +func TestL1ToL2_FailingDestinationDoesNotBlockOthersInGroup(t *testing.T) { + failure := errors.New("claimer failure") + tests := []struct { + name string + setup func(c *fakeClaimer) + }{ + {name: "IsClaimed error", setup: func(c *fakeClaimer) { c.claimErr = failure }}, + {name: "Enqueue error", setup: func(c *fakeClaimer) { c.err = failure }}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + bridges := []bridgesync.Bridge{ + makeSyncBridge(1, autoclaimtypes.L1OriginNetwork, 10, 2, 0), + makeSyncBridge(2, autoclaimtypes.L1OriginNetwork, 11, 3, 0), + makeSyncBridge(3, autoclaimtypes.L1OriginNetwork, 12, 4, 0), + makeSyncBridge(4, autoclaimtypes.L1OriginNetwork, 10, 5, 0), + makeSyncBridge(5, autoclaimtypes.L1OriginNetwork, 11, 6, 0), + makeSyncBridge(6, autoclaimtypes.L1OriginNetwork, 10, 7, 0), + makeSyncBridge(7, autoclaimtypes.L1OriginNetwork, 12, 8, 0), + } + source := &fakeBridgeSource{ + lastProcessedBlock: 10, + found: true, + bridgesByRange: map[blockRange][]bridgesync.Bridge{{from: 1, to: 10}: bridges}, + } + store := newMemoryCursorStore() + failing := &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer10ID, DestinationNetwork: 10}} + tt.setup(failing) + healthyY := &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer11ID, DestinationNetwork: 11}} + healthyZ := &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: "claimer-12", DestinationNetwork: 12}} + detector := newTestDetector( + t, source, store, newFakeRegistry(failing, healthyY, healthyZ), + WithStartBlock(0), WithBlockWindow(10), WithOverlapBlocks(0), + ) + seeded := autoclaimtypes.BridgeCursor{FromBlock: 0, ToBlock: 0, BlockNum: 0} + for _, dest := range []uint32{10, 11, 12} { + store.cursors[detector.cursorNameForDestination(dest)] = seeded + } + + result, err := detector.PollOnce(ctx) + require.ErrorIs(t, err, failure) + require.ErrorContains(t, err, "destination 10") + require.Equal(t, []uint32{10}, result.FailedDestinations) + require.Equal(t, 2, result.DeferredBridgeCount, "bridges 4 and 6 of the failed destination are deferred") + require.True(t, result.CursorAdvanced) + + require.Len(t, healthyY.enqueued, 2) + require.Len(t, healthyZ.enqueued, 2) + require.Equal(t, uint64(10), store.cursors[detector.cursorNameForDestination(11)].ToBlock) + require.Equal(t, uint64(10), store.cursors[detector.cursorNameForDestination(12)].ToBlock) + + require.Empty(t, failing.enqueued) + require.Len(t, failing.claimChecks, 1, "later bridges of the failed destination must not be evaluated") + require.Equal(t, seeded, store.cursors[detector.cursorNameForDestination(10)]) + + claimers := map[uint32]*fakeClaimer{10: failing, 11: healthyY, 12: healthyZ} + for dest, claimer := range claimers { + cursor := store.cursors[detector.cursorNameForDestination(dest)] + for _, bridge := range bridges { + if bridge.DestinationNetwork != dest || bridge.BlockNum > cursor.BlockNum { + continue + } + enqueued := false + for _, exit := range claimer.enqueued { + enqueued = enqueued || exit.DepositCount == bridge.DepositCount + } + require.True(t, enqueued, + "cursor of destination %d advanced past non-enqueued bridge %d", dest, bridge.DepositCount) + } + } + }) + } +} + +// TestL1ToL2_DestinationFailingAfterSomeBridgesEnqueued covers a destination that fails after part of its +// bridges were enqueued: its cursor must not move, and once it recovers the earlier bridges are +// re-enqueued idempotently before the cursor advances. +func TestL1ToL2_DestinationFailingAfterSomeBridgesEnqueued(t *testing.T) { + failure := errors.New("claimer failure") + tests := []struct { + name string + setup func(c *fakeClaimer) + clear func(c *fakeClaimer) + }{ + { + name: "IsClaimed error on the second bridge", + setup: func(c *fakeClaimer) { c.claimErr, c.claimErrFrom = failure, 2 }, + clear: func(c *fakeClaimer) { c.claimErr = nil }, + }, + { + name: "Enqueue error on the second bridge", + setup: func(c *fakeClaimer) { c.err, c.enqueueErrFrom = failure, 2 }, + clear: func(c *fakeClaimer) { c.err = nil }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + ctx := context.Background() + bridges := []bridgesync.Bridge{ + makeSyncBridge(1, autoclaimtypes.L1OriginNetwork, 10, 2, 0), + makeSyncBridge(2, autoclaimtypes.L1OriginNetwork, 11, 3, 0), + makeSyncBridge(4, autoclaimtypes.L1OriginNetwork, 10, 5, 0), + makeSyncBridge(6, autoclaimtypes.L1OriginNetwork, 10, 7, 0), + } + source := &fakeBridgeSource{ + lastProcessedBlock: 10, + found: true, + bridgesByRange: map[blockRange][]bridgesync.Bridge{{from: 1, to: 10}: bridges}, + } + store := newMemoryCursorStore() + failing := &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer10ID, DestinationNetwork: 10}} + tt.setup(failing) + healthy := &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer11ID, DestinationNetwork: 11}} + clock := testNow + detector := newTestDetector(t, source, store, newFakeRegistry(failing, healthy), + WithStartBlock(0), WithBlockWindow(10), WithOverlapBlocks(0), + WithNow(func() time.Time { return clock })) + detector.backoff = newDestinationBackoff(time.Second, 2*time.Minute, 0) + seeded := autoclaimtypes.BridgeCursor{FromBlock: 0, ToBlock: 0, BlockNum: 0} + for _, dest := range []uint32{10, 11} { + store.cursors[detector.cursorNameForDestination(dest)] = seeded + } + + result, err := detector.PollOnce(ctx) + require.ErrorIs(t, err, failure) + require.Equal(t, []uint32{10}, result.FailedDestinations) + require.Len(t, failing.enqueued, 1, "the first bridge was enqueued before the failure") + require.Equal(t, uint32(1), failing.enqueued[0].DepositCount) + require.Equal(t, seeded, store.cursors[detector.cursorNameForDestination(10)], + "the cursor of the failed destination does not move") + require.Equal(t, uint64(10), store.cursors[detector.cursorNameForDestination(11)].ToBlock) + + // Recovered and past the back-off window: earlier bridges are re-enqueued idempotently. + tt.clear(failing) + clock = clock.Add(time.Second) + result, err = detector.PollOnce(ctx) + require.NoError(t, err) + require.Empty(t, result.FailedDestinations) + require.Len(t, failing.enqueued, 3, "bridges 1, 4 and 6 are enqueued exactly once each") + require.Equal(t, uint64(10), store.cursors[detector.cursorNameForDestination(10)].ToBlock) + }) + } +} + func newTestDetector( t *testing.T, source autoclaimtypes.BridgeSource, @@ -824,13 +990,18 @@ func (r *fakeRegistry) Claimers(_ context.Context) ([]autoclaimtypes.Claimer, er } type fakeClaimer struct { - target autoclaimtypes.ClaimerTarget - err error - claimErr error - claimed bool - claimChecks []autoclaimtypes.BridgeExit - enqueued []autoclaimtypes.BridgeExit - seen map[autoclaimtypes.RequestKey]struct{} + target autoclaimtypes.ClaimerTarget + err error + claimErr error + // claimErrFrom, when > 0, makes claimErr apply only from the N-th IsClaimed call (1-based); + // enqueueErrFrom does the same for err and Enqueue. Zero means every call fails. + claimErrFrom int + enqueueErrFrom int + enqueueCalls int + claimed bool + claimChecks []autoclaimtypes.BridgeExit + enqueued []autoclaimtypes.BridgeExit + seen map[autoclaimtypes.RequestKey]struct{} } func (c *fakeClaimer) Target() autoclaimtypes.ClaimerTarget { @@ -839,14 +1010,15 @@ func (c *fakeClaimer) Target() autoclaimtypes.ClaimerTarget { func (c *fakeClaimer) IsClaimed(_ context.Context, bridge autoclaimtypes.BridgeExit) (bool, error) { c.claimChecks = append(c.claimChecks, bridge) - if c.claimErr != nil { + if c.claimErr != nil && len(c.claimChecks) >= c.claimErrFrom { return false, c.claimErr } return c.claimed, nil } func (c *fakeClaimer) Enqueue(_ context.Context, bridge autoclaimtypes.BridgeExit) error { - if c.err != nil { + c.enqueueCalls++ + if c.err != nil && c.enqueueCalls >= c.enqueueErrFrom { return c.err } if c.seen == nil { diff --git a/autoclaim/bridgedetector/l2_to_lx.go b/autoclaim/bridgedetector/l2_to_lx.go index 5d414f36f..ba9588dfa 100644 --- a/autoclaim/bridgedetector/l2_to_lx.go +++ b/autoclaim/bridgedetector/l2_to_lx.go @@ -4,9 +4,12 @@ import ( "context" "errors" "fmt" + "maps" + "slices" "sort" "time" + "github.com/agglayer/aggkit/autoclaim/metrics" autoclaimtypes "github.com/agglayer/aggkit/autoclaim/types" "github.com/agglayer/aggkit/bridgeservice" aggkitcommon "github.com/agglayer/aggkit/common" @@ -220,10 +223,11 @@ type L2ToLxPollResult struct { ProcessedSourceCount int // SkippedSourceCount is the number of sources skipped this poll, for any of three reasons: // nothing new for any pair; the source is permanently excluded from bridge service resolution - // (ErrSourceDisabled); or at least one pair's fetch group must be retried (url miss or not synced - // yet). Only the last one holds the source's block-window position back -- a source whose groups - // partially succeeded is that case, while a permanently excluded source is counted here yet lets - // the window advance past its verify row. + // (ErrSourceDisabled); or at least one pair must be retried (url miss, not synced yet, or a + // destination whose claim-state check failed this poll or is backed off). Only the last one holds + // the source's block-window position back -- a source whose groups partially succeeded is that + // case, while a permanently excluded source is counted here yet lets the window advance past its + // verify row. SkippedSourceCount int // CandidateCount is the total number of claim candidates fetched across every fetch group of // every processed source. @@ -232,8 +236,19 @@ type L2ToLxPollResult struct { EnqueuedCount int // AlreadyClaimedCount is the number of candidates skipped because the target already claimed them. AlreadyClaimedCount int - // CursorAdvanced reports whether the block-window cursor was advanced. + // CursorAdvanced reports whether the block-window cursor was advanced. It can be true together with + // a non-nil PollOnce error when only per-destination failures occurred (see FailedDestinations). CursorAdvanced bool + // DeferredCandidateCount is the number of candidates of a destination whose claim-state check had + // already failed earlier in the same fetch group: they are neither checked nor enqueued this poll and + // are re-fetched on a later poll from the destination's unchanged pair cursor. + DeferredCandidateCount int + // FailedDestinations lists, in ascending order, the destinations whose claim-state check (IsClaimed) + // failed this poll. Their pair cursors were not advanced past the failing group. + FailedDestinations []uint32 + // BackedOffDestinations lists, in ascending order, the destinations excluded from this poll because + // they are inside a back-off window after an earlier failure. + BackedOffDestinations []uint32 } // L2ToLx is the bridge detector that discovers L2-initiated (rollup-origin) bridge exits by watching @@ -263,6 +278,9 @@ type L2ToLx struct { // "/", so a persistent condition logs at most once per retryLogInterval // rather than on every poll. Only ever touched from the single poll goroutine. retryLogged map[string]time.Time + // backoff pauses detection for a destination whose claim-state check keeps failing. Only ever + // touched from the single poll goroutine. + backoff *destinationBackoff } // retryLogInterval bounds how often a persistent retry-later condition is logged per source. @@ -320,6 +338,7 @@ func NewL2ToLx( for _, option := range options { option(detector) } + detector.backoff = newDestinationBackoff(detector.pollPeriod, destinationBackoffMax, destinationBackoffJitter) return detector, nil } @@ -383,19 +402,47 @@ const ( // sourceProcessed means every fetch group's candidates were enqueued and every pair's LER cursor // advanced. sourceProcessed - // sourceRetryLater means at least one of the source's fetch groups was skipped for a reason that - // can still clear on its own (a finder miss for a service not yet announced or unhealthy, or a - // source bridge service not synced yet) and must be retried on a later poll. Groups that did - // succeed keep their persisted pair cursors. A permanently excluded source is NOT this outcome: - // it can never clear, so holding the shared cursor for it would deadlock every other source. + // sourceRetryLater means at least one of the source's pending pairs was not advanced for a reason + // that can still clear on its own (a finder miss for a service not yet announced or unhealthy, a + // source bridge service not synced yet, a destination whose claim-state check failed, or a + // destination inside its back-off window) and must be retried on a later poll. Pairs that did + // succeed keep their persisted cursors. A permanently excluded source is NOT this outcome: it can + // never clear, so holding the shared cursor for it would deadlock every other source. sourceRetryLater ) +// pollDestinationOutcomes accumulates per-destination results across all sources of one poll. +type pollDestinationOutcomes struct { + // failed holds the first IsClaimed error per destination this poll. + failed map[uint32]error + // succeeded holds the destinations with at least one LER pair cursor advanced this poll. + succeeded map[uint32]struct{} + // backedOff holds the destinations skipped by back-off this poll (computed once at poll start). + backedOff map[uint32]struct{} +} + +func newPollDestinationOutcomes() *pollDestinationOutcomes { + return &pollDestinationOutcomes{ + failed: make(map[uint32]error), + succeeded: make(map[uint32]struct{}), + backedOff: make(map[uint32]struct{}), + } +} + // PollOnce processes at most one L1 block window of verified-batch rows. Each source found in the // window resolves an independent from_ler per destination claimer and issues one claim-candidates // query per distinct from_ler (issue #1651), so the number of fetcher calls per poll stays bounded by // the number of enabled claimers times the pages each of their queries needs, and collapses back to // today's single batched query set as soon as every destination of a source shares one cursor. +// +// Failures of a destination's claim-state check (IsClaimed) are isolated per destination: the other +// destinations keep processing and advance their pair cursors, the failed destination's pair cursor is +// held, its source holds the block-window cursor just before its verify row, and the destination is +// then paused by a per-destination back-off (it is excluded from fetches until its window ends). Those +// per-destination errors are returned joined, after the block-window cursor has been handled, so a +// non-nil error together with L2ToLxPollResult.CursorAdvanced == true is possible; FailedDestinations +// lists the destinations concerned. Any other failure (registry, enqueue, fetch, cursor storage) is a +// hard error that leaves the block-window cursor untouched, as before. func (w *L2ToLx) PollOnce(ctx context.Context) (*L2ToLxPollResult, error) { if !w.enabled { return &L2ToLxPollResult{}, nil @@ -433,14 +480,27 @@ func (w *L2ToLx) PollOnce(ctx context.Context) (*L2ToLxPollResult, error) { return result, err } + outcomes := newPollDestinationOutcomes() + defer w.updateBackoff(ctx, outcomes) + for _, destination := range sortedNetworks(destinationNetworks) { + // dup guards against the same destination being listed twice: sortedNetworks does not de-duplicate. + if _, dup := outcomes.backedOff[destination]; dup || !w.backoff.shouldSkip(destination, w.now()) { + continue + } + w.logDebugf("autoclaim l2-to-lx bridge detector: skipping destination %d, backed off until %s", + destination, w.backoff.skipUntil(destination)) + outcomes.backedOff[destination] = struct{}{} + result.BackedOffDestinations = append(result.BackedOffDestinations, destination) + } + // retryBlock is the block of the earliest verify row whose source was skipped for a reason that // can still clear on its own (finder miss or source not synced yet); 0 = none. A source that can // never resolve (permanently excluded) deliberately does not contribute here. var retryBlock uint64 for _, source := range orderedSourceLERs(latestBySource) { - outcome, err := w.processSource(ctx, source, destinationNetworks, result) + outcome, err := w.processSource(ctx, source, destinationNetworks, outcomes, result) if err != nil { - return result, err + return result, w.pollError(err, outcomes, result) } switch outcome { case sourceProcessed: @@ -464,13 +524,14 @@ func (w *L2ToLx) PollOnce(ctx context.Context) (*L2ToLxPollResult, error) { // the retry fetch still uses from_ler = , which covers every candidate missed in // between; a pair whose group did succeed keeps its advanced cursor and is simply not re-fetched. // A hard error above returns before this point, leaving the cursor unchanged so the whole window - // is retried. + // is retried. A destination whose claim-state check failed (or that is backed off) is a retryable + // skip of its source, so it holds the cursor like any other; its error is returned after this point. cursorToBlock := toBlock if retryBlock > 0 { if retryBlock <= fromBlock { // The retried row sits at the very start of the window: there is no forward progress to // record, keep the stored cursor untouched and retry the same window next poll. - return result, nil + return result, w.pollError(nil, outcomes, result) } cursorToBlock = retryBlock - 1 } @@ -481,11 +542,52 @@ func (w *L2ToLx) PollOnce(ctx context.Context) (*L2ToLxPollResult, error) { BlockPos: 0, } if err := w.cursorStore.SaveBridgeCursor(ctx, w.cursorName, nextCursor, w.now()); err != nil { - return result, fmt.Errorf("save autoclaim l2-to-lx bridge detector cursor %s: %w", w.cursorName, err) + return result, w.pollError( + fmt.Errorf("save autoclaim l2-to-lx bridge detector cursor %s: %w", w.cursorName, err), outcomes, result) } result.CursorAdvanced = true - return result, nil + return result, w.pollError(nil, outcomes, result) +} + +// pollError fills result.FailedDestinations (ascending) from outcomes and returns hardErr (nil when +// none) joined with one "destination %d: ..." error per failed destination, in ascending order. The +// result is nil when there is neither a hard error nor a failed destination. +func (w *L2ToLx) pollError(hardErr error, outcomes *pollDestinationOutcomes, result *L2ToLxPollResult) error { + errs := []error{hardErr} + result.FailedDestinations = nil + for _, destination := range sortedDestinationKeys(outcomes.failed) { + result.FailedDestinations = append(result.FailedDestinations, destination) + errs = append(errs, fmt.Errorf("destination %d: %w", destination, outcomes.failed[destination])) + } + return errors.Join(errs...) +} + +// updateBackoff feeds the per-destination outcome of a poll into the back-off tracker: a destination +// that advanced at least one pair cursor and did not fail is reset, and a destination whose claim-state +// check failed opens its next back-off window (one Warn per window). A destination with no activity this +// poll is left unchanged. When ctx is done the failure may be a cancellation artefact, so failures are +// not recorded. +func (w *L2ToLx) updateBackoff(ctx context.Context, outcomes *pollDestinationOutcomes) { + defer func() { metrics.SetStalledDestinations(metrics.DetectorL2ToLx, w.backoff.stalledCount()) }() + for _, destination := range sortedDestinationKeys(outcomes.succeeded) { + if _, failed := outcomes.failed[destination]; !failed { + w.backoff.recordSuccess(destination) + } + } + if ctx.Err() != nil { + return + } + for _, destination := range sortedDestinationKeys(outcomes.failed) { + failure := outcomes.failed[destination] + w.backoff.recordFailureAndWarn( + metrics.DetectorL2ToLx, "l2-to-lx", destination, failure, w.now(), w.logWarnf) + } +} + +// sortedDestinationKeys returns the keys of a destination-keyed map in ascending order. +func sortedDestinationKeys[V any](m map[uint32]V) []uint32 { + return slices.Sorted(maps.Keys(m)) } // processSource evaluates one source network's newest LER against every destination the enabled @@ -495,14 +597,20 @@ func (w *L2ToLx) PollOnce(ctx context.Context) (*L2ToLxPollResult, error) { // baseline instead of inheriting an already-advanced cursor that would skip its history, and it does // not disturb the established destinations, which keep batching exactly as before. // +// Pending pairs whose destination is backed off this poll (outcomes.backedOff) are not fetched at all: +// they are left out of every fetch group, keep their pair cursors, and make the source retry later, so +// a backed-off pair holds the block-window cursor exactly like a failed one. +// // It returns sourceUpToDate when there is nothing to do -- no pair has anything new, the source has // no destination pair at all, or the source is permanently excluded from bridge service resolution -- -// sourceRetryLater when at least one group was skipped for a reason that can still clear (finder miss -// or not synced yet), and sourceProcessed when every group advanced its pairs' LER cursors. +// sourceRetryLater when at least one pair was not advanced for a reason that can still clear (finder +// miss, not synced yet, a failed claim-state check, or a backed-off destination), and sourceProcessed +// when every pending pair advanced its LER cursor. func (w *L2ToLx) processSource( ctx context.Context, source sourceLER, destinationNetworks []uint32, + outcomes *pollDestinationOutcomes, result *L2ToLxPollResult, ) (sourceOutcome, error) { destinationIDs := excludeNetwork(destinationNetworks, source.sourceID) @@ -556,6 +664,21 @@ func (w *L2ToLx) processSource( } result.NewLERSourceCount++ + active := make([]pendingDestination, 0, len(pending)) + deferred := 0 + for _, pair := range pending { + if _, backedOff := outcomes.backedOff[pair.destination]; backedOff { + deferred++ + continue + } + active = append(active, pair) + } + if len(active) == 0 { + // Every pending pair is backed off: nothing is fetched, and the source is retried later so its + // verify row is re-observed once the back-off window ends. + return sourceRetryLater, nil + } + url, err := w.fetcher.GetURL(source.sourceID) if err != nil { if errors.Is(err, ErrSourceDisabled) { @@ -579,7 +702,11 @@ func (w *L2ToLx) processSource( return sourceRetryLater, nil } - return w.processLERGroups(ctx, source, url, groupPendingByFromLER(pending), result) + outcome, err := w.processLERGroups(ctx, source, url, groupPendingByFromLER(active), outcomes, result) + if err == nil && outcome == sourceProcessed && deferred > 0 { + outcome = sourceRetryLater + } + return outcome, err } // processLERGroups fetches, enqueues and persists each from_ler group of one source in turn. A group @@ -588,11 +715,18 @@ func (w *L2ToLx) processSource( // gates it. A group that failed leaves its pairs' cursors untouched, so the next poll re-fetches only // that group; the source itself still reports retry-later (or the hard error), which holds the // block-window cursor just before this source's verify row so the row is re-observed. +// +// Within a group, a destination whose claim-state check failed is recorded in outcomes.failed and +// excluded from the cursor advance, while the group's other destinations still advance (and are +// recorded in outcomes.succeeded); the source then reports retry-later. On the next poll the failed +// pair's from_ler differs from its former group mates', so it forms its own group and only that group +// is re-fetched. func (w *L2ToLx) processLERGroups( ctx context.Context, source sourceLER, url string, groups []lerGroup, + outcomes *pollDestinationOutcomes, result *L2ToLxPollResult, ) (sourceOutcome, error) { var ( @@ -616,19 +750,42 @@ func (w *L2ToLx) processLERGroups( } result.CandidateCount += len(candidates) - if err := w.enqueueCandidates(ctx, source, candidates, result); err != nil { + failed, err := w.enqueueCandidates(ctx, source, candidates, result) + // Record soft destination failures even when the same group also hit a hard error, so the + // failing destination still gets its back-off window. + for destination, failure := range failed { + if _, seen := outcomes.failed[destination]; !seen { + outcomes.failed[destination] = failure + } + } + if len(failed) > 0 { + retryLater = true + } + if err != nil { if firstErr == nil { firstErr = err } continue } - if err := w.advanceLERCursors(ctx, source, group.destinations); err != nil { + advance := make([]uint32, 0, len(group.destinations)) + for _, destination := range group.destinations { + if _, isFailed := failed[destination]; !isFailed { + advance = append(advance, destination) + } + } + if len(advance) == 0 { + continue + } + if err := w.advanceLERCursors(ctx, source, advance); err != nil { if firstErr == nil { firstErr = err } continue } + for _, destination := range advance { + outcomes.succeeded[destination] = struct{}{} + } } if firstErr != nil { @@ -722,12 +879,7 @@ func lerGroupKey(fromLER *common.Hash) string { // sortedNetworks returns a copy of networks in ascending order. func sortedNetworks(networks []uint32) []uint32 { - sorted := make([]uint32, len(networks)) - copy(sorted, networks) - sort.Slice(sorted, func(i, j int) bool { - return sorted[i] < sorted[j] - }) - return sorted + return slices.Sorted(slices.Values(networks)) } // initialFromLER derives the exclusive lower-bound LER the first time a source network is seen. When @@ -827,20 +979,33 @@ func (w *L2ToLx) fetchCandidatesForDestinations( // enqueueCandidates routes every candidate to its destination claimer and enqueues the ones that are // not already claimed on the target. +// +// A failing claim-state check (IsClaimed, the only call that reaches the destination's RPC) is isolated +// per destination: it is recorded in the returned failed map (first error per destination), that +// destination's remaining candidates in this call are skipped (counted in +// result.DeferredCandidateCount, neither checked nor enqueued), and the other destinations' candidates +// continue. Every other failure (claimer registry, EnqueueRequest) is a hard error returned as err, +// which aborts the call. func (w *L2ToLx) enqueueCandidates( ctx context.Context, source sourceLER, candidates []ClaimCandidate, result *L2ToLxPollResult, -) error { +) (map[uint32]error, error) { + failed := make(map[uint32]error) for i := range candidates { candidate := candidates[i] exit := candidate.Bridge exit.SourceNetwork = source.sourceID + if _, isFailed := failed[exit.DestinationNetwork]; isFailed { + result.DeferredCandidateCount++ + continue + } + claimer, ok, err := w.registry.ClaimerForDestination(ctx, exit.DestinationNetwork) if err != nil { - return fmt.Errorf("resolve claimer for destination %d: %w", exit.DestinationNetwork, err) + return failed, fmt.Errorf("resolve claimer for destination %d: %w", exit.DestinationNetwork, err) } if !ok { // The destination was requested from the source bridge service but no claimer handles it; @@ -850,11 +1015,17 @@ func (w *L2ToLx) enqueueCandidates( claimed, err := claimer.IsClaimed(ctx, exit) if err != nil { - return fmt.Errorf("check target claim state for source %d deposit %d: %w", + failed[exit.DestinationNetwork] = fmt.Errorf("check target claim state for source %d deposit %d: %w", source.sourceID, exit.DepositCount, err) + continue } if claimed { result.AlreadyClaimedCount++ + metrics.IncSkippedAlreadyClaimed(metrics.DetectorL2ToLx, exit.DestinationNetwork) + w.logDebugf("autoclaim %s bridge detector: skipped bridge %s (destination %d, global index %s): "+ + "already claimed on target", metrics.DetectorL2ToLx, + autoclaimtypes.DeriveRequestKey(source.sourceID, exit.DestinationNetwork, exit.DepositCount), + exit.DestinationNetwork, globalIndexString(exit)) continue } @@ -864,18 +1035,19 @@ func (w *L2ToLx) enqueueCandidates( request.VerifyBlockNum = source.verifyNum if _, inserted, err := w.enqueuer.EnqueueRequest(ctx, request); err != nil { - return fmt.Errorf("enqueue autoclaim request %s: %w", request.Key, err) + return failed, fmt.Errorf("enqueue autoclaim request %s: %w", request.Key, err) } else if inserted { result.EnqueuedCount++ } } - return nil + return failed, nil } // advanceLERCursors records the source's newest LER as processed for each of the given destinations. -// It is called once per fetch group and only after every candidate of that group has been enqueued, -// so a pair's cursor never moves past candidates that were not persisted. Destinations whose group -// did not complete are not passed in, which leaves their own cursors at their previous value. +// It is called once per fetch group and only after every candidate of that group has been enqueued or +// classified as already claimed, so a pair's cursor never moves past candidates that were not +// persisted. Destinations whose group did not complete, and destinations whose claim-state check failed +// within the group, are not passed in, which leaves their own cursors at their previous value. func (w *L2ToLx) advanceLERCursors(ctx context.Context, source sourceLER, destinations []uint32) error { for _, destination := range destinations { cursor := autoclaimtypes.LERCursor{ @@ -945,6 +1117,12 @@ func (w *L2ToLx) logWarnf(format string, args ...interface{}) { } } +func (w *L2ToLx) logDebugf(format string, args ...interface{}) { + if w.log != nil { + w.log.Debugf(format, args...) + } +} + // shouldLogRetry reports whether the given retry-later condition for sourceID is due to be logged, // and records the time if so. func (w *L2ToLx) shouldLogRetry(condition string, sourceID uint32) bool { diff --git a/autoclaim/bridgedetector/l2_to_lx_test.go b/autoclaim/bridgedetector/l2_to_lx_test.go index 6d5109fe9..9699ceccf 100644 --- a/autoclaim/bridgedetector/l2_to_lx_test.go +++ b/autoclaim/bridgedetector/l2_to_lx_test.go @@ -5,17 +5,22 @@ import ( "errors" "fmt" "math/big" + "net/http" "strings" "testing" "time" + autoclaimmetrics "github.com/agglayer/aggkit/autoclaim/metrics" autoclaimtypes "github.com/agglayer/aggkit/autoclaim/types" "github.com/agglayer/aggkit/bridgeservice" "github.com/agglayer/aggkit/bridgeservicefinder" bridgesynctypes "github.com/agglayer/aggkit/bridgesync/types" commonmocks "github.com/agglayer/aggkit/common/mocks" "github.com/agglayer/aggkit/l1infotreesync" + "github.com/agglayer/aggkit/prometheus" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/rpc" + "github.com/prometheus/client_golang/prometheus/testutil" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" ) @@ -727,14 +732,27 @@ func TestL2ToLxAlreadyClaimedSkip(t *testing.T) { } lerStore := newFakePerPairLERStore() enqueuer := newFakeEnqueuer() + log := &capturingLogger{} detector := newTestL2ToLxDetector( t, source, fetcher, newFakeRegistry(claimer0), newMemoryCursorStore(), lerStore, enqueuer, - WithL2ToLxBlockWindow(50), + WithL2ToLxBlockWindow(50), WithL2ToLxLogger(log), ) + prometheus.Init() + autoclaimmetrics.Register() + vec, ok := prometheus.CounterVec("detector_skipped_already_claimed_total") + require.True(t, ok) + skipped := vec.WithLabelValues(string(autoclaimmetrics.DetectorL2ToLx), "0") + before := testutil.ToFloat64(skipped) + result, err := detector.PollOnce(ctx) require.NoError(t, err) require.Equal(t, 1, result.AlreadyClaimedCount) + require.Equal(t, before+1, testutil.ToFloat64(skipped)) + require.Contains(t, log.debugs, + "autoclaim l2_to_lx bridge detector: skipped bridge 1:0:5 (destination 0, global index unknown): "+ + "already claimed on target") + require.Empty(t, log.warns) require.Equal(t, 0, result.EnqueuedCount) require.Empty(t, enqueuer.order) require.Len(t, claimer0.claimChecks, 1) @@ -1583,3 +1601,410 @@ func (e *fakeEnqueuer) EnqueueRequest( e.order = append(e.order, request) return &request, true, nil } + +// --- S23: per-destination isolation and back-off --- + +const ( + isoSource uint32 = 1 + isoFailing uint32 = 10 + isoHealthy uint32 = 11 + isoClaimed uint32 = 12 + isoVerify = uint64(40) + isoFirstTo = isoVerify - 1 // block cursor held just before the source's verify row + isoLastBlck = uint64(50) +) + +// destinationFilteringFetcher wraps fakeFetcher and, like the real bridge service, only returns the +// candidates whose destination is in the query's DestinationNetworkIDs (with a matching count). +type destinationFilteringFetcher struct { + *fakeFetcher +} + +func (f *destinationFilteringFetcher) GetClaimCandidates( + ctx context.Context, query ClaimCandidatesQuery, +) ([]ClaimCandidate, int, error) { + candidates, _, err := f.fakeFetcher.GetClaimCandidates(ctx, query) + if err != nil { + return nil, 0, err + } + wanted := make(map[uint32]struct{}, len(query.DestinationNetworkIDs)) + for _, destination := range query.DestinationNetworkIDs { + wanted[destination] = struct{}{} + } + filtered := make([]ClaimCandidate, 0, len(candidates)) + for _, candidate := range candidates { + if _, ok := wanted[candidate.Bridge.DestinationNetwork]; ok { + filtered = append(filtered, candidate) + } + } + return filtered, len(filtered), nil +} + +type isolationFixture struct { + detector *L2ToLx + fetcher *destinationFilteringFetcher + registry *fakeRegistry + failing *fakeClaimer + healthy *fakeClaimer + claimed *fakeClaimer + lerStore *fakePerPairLERStore + cursorStore *memoryCursorStore + enqueuer *fakeEnqueuer + log *capturingLogger + clock *time.Time + lerX, lerY common.Hash + // published maps a source LER to every candidate the source published up to (and including) it. + published map[common.Hash][]ClaimCandidate +} + +// newIsolationFixture builds one source whose three destination pairs share from_ler = lerX and whose +// newest LER lerY (verify row at block 40) carries two candidates for the failing destination, one for +// the healthy destination and one for a destination where the bridge is already claimed. +func newIsolationFixture(t *testing.T, failure error) *isolationFixture { + t.Helper() + f := &isolationFixture{ + failing: &fakeClaimer{ + target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer10ID, DestinationNetwork: isoFailing}, + claimErr: failure, + }, + healthy: &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: fakeClaimer11ID, DestinationNetwork: isoHealthy}}, + claimed: &fakeClaimer{target: autoclaimtypes.ClaimerTarget{ID: "claimer-12", DestinationNetwork: isoClaimed}, claimed: true}, + lerStore: newFakePerPairLERStore(), + enqueuer: newFakeEnqueuer(), + log: &capturingLogger{}, + lerX: lerHash(20), + lerY: lerHash(30), + } + clock := testNow + f.clock = &clock + f.registry = newFakeRegistry(f.failing, f.healthy, f.claimed) + f.cursorStore = newMemoryCursorStore() + + candidates := []ClaimCandidate{ + makeCandidate(5, isoFailing), makeCandidate(6, isoFailing), + makeCandidate(7, isoHealthy), makeCandidate(8, isoClaimed), + } + f.published = map[common.Hash][]ClaimCandidate{f.lerX: nil, f.lerY: candidates} + f.fetcher = &destinationFilteringFetcher{fakeFetcher: newFakeFetcher()} + f.fetcher.urls[isoSource] = fakeSrcURL1 + f.fetcher.setPageForLER(fakeSrcURL1, &f.lerX, 1, candidates, len(candidates)) + + for _, destination := range []uint32{isoFailing, isoHealthy, isoClaimed} { + f.lerStore.set(isoSource, destination, autoclaimtypes.LERCursor{ + SourceNetwork: isoSource, DestinationNetwork: destination, LastLER: f.lerX, LastVerifyBlockNum: 5, + }) + } + + row := makeVerifyRow(isoSource, f.lerY, isoVerify) + source := &fakeVerifiedBatchSource{ + lastProcessedBlock: isoLastBlck, + rowsByRange: map[blockRange][]*l1infotreesync.VerifyBatches{ + {from: 0, to: 49}: {row}, + {from: isoFirstTo, to: isoLastBlck}: {row}, // re-observed window (overlap 1) while the cursor is held + }, + } + f.detector = newTestL2ToLxDetector(t, source, f.fetcher, f.registry, f.cursorStore, f.lerStore, f.enqueuer, + WithL2ToLxBlockWindow(50), WithL2ToLxLogger(f.log), WithL2ToLxNow(func() time.Time { return *f.clock })) + f.detector.backoff = newDestinationBackoff(time.Second, 2*time.Minute, 0) + return f +} + +func (f *isolationFixture) pairLER(t *testing.T, destination uint32) common.Hash { + t.Helper() + cursor, found, err := f.lerStore.GetLERCursor(context.Background(), isoSource, destination) + require.NoError(t, err) + require.True(t, found) + return cursor.LastLER +} + +func (f *isolationFixture) blockCursorTo(t *testing.T) uint64 { + t.Helper() + cursor, ok := f.cursorStore.cursors[defaultL2ToLxCursorName] + require.True(t, ok) + return cursor.ToBlock +} + +// requireLERCursorInvariant asserts the S23 safety invariant: for every persisted LER pair cursor, every +// candidate the source published up to that cursor's LER for that destination was enqueued or is already +// claimed on the destination. A pair cursor therefore never moves past an un-enqueued candidate. +func requireLERCursorInvariant(t *testing.T, f *isolationFixture) { + t.Helper() + require.NotEmpty(t, f.lerStore.cursors) + for key, cursor := range f.lerStore.cursors { + sourceID, destination := key[0], key[1] + upTo, known := f.published[cursor.LastLER] + require.True(t, known, "pair (%d, %d) cursor points at an unknown LER %s", sourceID, destination, cursor.LastLER) + for _, candidate := range upTo { + if candidate.Bridge.DestinationNetwork != destination { + continue + } + requestKey := autoclaimtypes.DeriveRequestKey(sourceID, destination, candidate.Bridge.DepositCount) + _, enqueued := f.enqueuer.requests[requestKey] + claimer := f.registry.claimers[destination] + alreadyClaimed := claimer != nil && claimer.claimed + require.True(t, enqueued || alreadyClaimed, + "pair (%d, %d) cursor at %s is past candidate %s, which was neither enqueued nor already claimed", + sourceID, destination, cursor.LastLER, requestKey) + } + } +} + +func TestL2ToLx_FailingDestinationIsIsolated(t *testing.T) { + ctx := context.Background() + claimErr := errors.New("destination rpc down") + f := newIsolationFixture(t, claimErr) + + result, err := f.detector.PollOnce(ctx) + require.ErrorIs(t, err, claimErr) + require.ErrorContains(t, err, fmt.Sprintf("destination %d:", isoFailing)) + require.True(t, result.CursorAdvanced, "a per-destination failure still saves the (held) block cursor") + require.Equal(t, []uint32{isoFailing}, result.FailedDestinations) + require.Empty(t, result.BackedOffDestinations) + require.Equal(t, 1, result.DeferredCandidateCount, "the failing destination's second candidate is deferred") + require.Len(t, f.failing.claimChecks, 1, "no IsClaimed call after the destination failed") + require.Equal(t, 1, result.SkippedSourceCount) + require.Equal(t, 0, result.ProcessedSourceCount) + + // One query for the shared from_ler group. + require.Len(t, f.fetcher.queries, 1) + require.Equal(t, []uint32{isoFailing, isoHealthy, isoClaimed}, f.fetcher.queries[0].DestinationNetworkIDs) + + // The healthy destination's candidate is enqueued and its pair advances; the already-claimed pair + // advances too; the failing pair stays. + _, enqueued := f.enqueuer.requests[autoclaimtypes.DeriveRequestKey(isoSource, isoHealthy, 7)] + require.True(t, enqueued) + require.Len(t, f.enqueuer.order, 1) + require.Equal(t, f.lerY, f.pairLER(t, isoHealthy)) + require.Equal(t, f.lerY, f.pairLER(t, isoClaimed)) + require.Equal(t, f.lerX, f.pairLER(t, isoFailing)) + + // The block cursor is held just before the source's verify row. + require.Equal(t, isoFirstTo, f.blockCursorTo(t)) + + require.Len(t, f.log.warns, 1) + require.Contains(t, f.log.warns[0], "autoclaim l2-to-lx bridge detector: detection for destination 10 "+ + "paused for 1s after error (attempt 1)") + require.Equal(t, 1, f.detector.backoff.stalledCount()) + requireLERCursorInvariant(t, f) +} + +// TestL2ToLx_DestinationFailingAfterSomeCandidatesEnqueued covers a destination whose claim-state check fails +// on its second candidate: the first one is enqueued, the pair cursor stays, and once the destination +// recovers the first is re-enqueued idempotently before the cursor advances. +func TestL2ToLx_DestinationFailingAfterSomeCandidatesEnqueued(t *testing.T) { + ctx := context.Background() + f := newIsolationFixture(t, nil) + failure := errors.New("destination rpc down") + f.failing.claimErr, f.failing.claimErrFrom = failure, 2 + + result, err := f.detector.PollOnce(ctx) + require.ErrorIs(t, err, failure) + require.Equal(t, []uint32{isoFailing}, result.FailedDestinations) + _, enqueued := f.enqueuer.requests[autoclaimtypes.DeriveRequestKey(isoSource, isoFailing, 5)] + require.True(t, enqueued, "the first candidate was enqueued before the failure") + _, enqueued = f.enqueuer.requests[autoclaimtypes.DeriveRequestKey(isoSource, isoFailing, 6)] + require.False(t, enqueued) + require.Equal(t, f.lerX, f.pairLER(t, isoFailing), "the pair cursor does not move") + require.Equal(t, isoFirstTo, f.blockCursorTo(t)) + requireLERCursorInvariant(t, f) + + f.failing.claimErr = nil + *f.clock = testNow.Add(time.Second) + _, err = f.detector.PollOnce(ctx) + require.NoError(t, err) + for _, deposit := range []uint32{5, 6} { + _, enqueued = f.enqueuer.requests[autoclaimtypes.DeriveRequestKey(isoSource, isoFailing, deposit)] + require.True(t, enqueued) + } + require.Len(t, f.enqueuer.order, 3, "failing deposits 5 and 6 plus the healthy one, each enqueued once") + require.Equal(t, f.lerY, f.pairLER(t, isoFailing)) + requireLERCursorInvariant(t, f) +} + +// TestL2ToLx_SoftFailureRecordedWhenGroupAlsoHitsHardError covers a fetch group where one destination's +// claim-state check fails (soft) and a later enqueue fails with a hard error: the soft failure must still +// open the destination's back-off window. +func TestL2ToLx_SoftFailureRecordedWhenGroupAlsoHitsHardError(t *testing.T) { + ctx := context.Background() + soft := errors.New("destination rpc down") + hard := errors.New("local db down") + f := newIsolationFixture(t, soft) + f.enqueuer.errForDestination = map[uint32]error{isoHealthy: hard} + + result, err := f.detector.PollOnce(ctx) + require.ErrorIs(t, err, hard) + require.Equal(t, []uint32{isoFailing}, result.FailedDestinations) + require.Equal(t, 1, f.detector.backoff.stalledCount(), "the soft failure opens a back-off window") + require.Len(t, f.log.warns, 1) + require.Contains(t, f.log.warns[0], "detection for destination 10 paused for 1s") + require.Equal(t, f.lerX, f.pairLER(t, isoFailing)) + require.Equal(t, f.lerX, f.pairLER(t, isoHealthy), "nothing advances in the group that hit the hard error") + requireLERCursorInvariant(t, f) +} + +func TestL2ToLx_BackedOffDestinationHoldsBlockCursor(t *testing.T) { + ctx := context.Background() + f := newIsolationFixture(t, errors.New("boom")) + + _, err := f.detector.PollOnce(ctx) + require.Error(t, err) + queries := len(f.fetcher.queries) + + // Same clock, then just before the window ends: the destination is backed off, nothing is fetched, + // and the source still holds the block cursor at the verify row. + for _, at := range []time.Time{testNow, testNow.Add(time.Second - time.Nanosecond)} { + *f.clock = at + result, err := f.detector.PollOnce(ctx) + require.NoError(t, err) + require.Equal(t, []uint32{isoFailing}, result.BackedOffDestinations) + require.Empty(t, result.FailedDestinations) + require.Equal(t, 1, result.NewLERSourceCount) + require.Equal(t, 1, result.SkippedSourceCount) + require.Equal(t, 0, result.ProcessedSourceCount) + require.Len(t, f.fetcher.queries, queries, "a backed-off pair is not fetched") + require.Len(t, f.failing.claimChecks, 1, "no IsClaimed call while backed off") + require.Equal(t, isoFirstTo, f.blockCursorTo(t), "the block cursor stays held at the verify row") + require.Equal(t, f.lerX, f.pairLER(t, isoFailing)) + requireLERCursorInvariant(t, f) + } + require.Len(t, f.log.warns, 1, "no Warn while backed off") + var backoffDebugs []string + for _, line := range f.log.debugs { + if strings.Contains(line, "backed off until") { + backoffDebugs = append(backoffDebugs, line) + } + } + require.Len(t, backoffDebugs, 2) + require.Contains(t, backoffDebugs[0], + "autoclaim l2-to-lx bridge detector: skipping destination 10, backed off until") +} + +func TestL2ToLx_FailedDestinationRetriedAloneAfterWindow(t *testing.T) { + ctx := context.Background() + f := newIsolationFixture(t, errors.New("boom")) + + _, err := f.detector.PollOnce(ctx) + require.Error(t, err) + queries := len(f.fetcher.queries) + + f.failing.claimErr = nil + *f.clock = testNow.Add(time.Second) + result, err := f.detector.PollOnce(ctx) + require.NoError(t, err) + require.Empty(t, result.BackedOffDestinations) + require.Empty(t, result.FailedDestinations) + require.Equal(t, 1, result.ProcessedSourceCount) + + // Only the failed destination's own group is re-fetched, from its unchanged pair cursor. + require.Len(t, f.fetcher.queries, queries+1) + retry := f.fetcher.queries[queries] + require.Equal(t, []uint32{isoFailing}, retry.DestinationNetworkIDs) + require.NotNil(t, retry.FromLER) + require.Equal(t, f.lerX, *retry.FromLER) + + for _, deposit := range []uint32{5, 6} { + _, enqueued := f.enqueuer.requests[autoclaimtypes.DeriveRequestKey(isoSource, isoFailing, deposit)] + require.True(t, enqueued) + } + require.Equal(t, f.lerY, f.pairLER(t, isoFailing)) + require.Equal(t, isoLastBlck, f.blockCursorTo(t), "the block cursor moves on once the destination recovered") + require.Equal(t, 0, f.detector.backoff.stalledCount()) + requireLERCursorInvariant(t, f) +} + +func TestL2ToLx_BackoffWindowsGrowAndSuccessResets(t *testing.T) { + ctx := context.Background() + f := newIsolationFixture(t, errors.New("boom")) + + for i, window := range []time.Duration{time.Second, 2 * time.Second, 4 * time.Second} { + result, err := f.detector.PollOnce(ctx) + require.Error(t, err) + require.Equal(t, []uint32{isoFailing}, result.FailedDestinations) + require.Len(t, f.log.warns, i+1) + require.Contains(t, f.log.warns[i], "paused for "+window.String()) + require.Contains(t, f.log.warns[i], fmt.Sprintf("after error (attempt %d)", i+1)) + require.Equal(t, f.clock.Add(window), f.detector.backoff.skipUntil(isoFailing)) + require.Equal(t, isoFirstTo, f.blockCursorTo(t)) + requireLERCursorInvariant(t, f) + *f.clock = f.clock.Add(window) + } + + f.failing.claimErr = nil + result, err := f.detector.PollOnce(ctx) + require.NoError(t, err) + require.Empty(t, result.FailedDestinations) + require.Equal(t, 0, f.detector.backoff.stalledCount(), "success resets the back-off") + require.Len(t, f.log.warns, 3) + requireLERCursorInvariant(t, f) + + // A new failure starts again from the initial window. + window, attempt := f.detector.backoff.recordFailure(isoFailing, errors.New("again"), *f.clock) + require.Equal(t, time.Second, window) + require.Equal(t, 1, attempt) +} + +func TestL2ToLx_BackoffRateLimitedUsesHTTP429Wording(t *testing.T) { + ctx := context.Background() + f := newIsolationFixture(t, rpc.HTTPError{StatusCode: http.StatusTooManyRequests, Status: "429 Too Many Requests"}) + + _, err := f.detector.PollOnce(ctx) + require.Error(t, err) + require.Len(t, f.log.warns, 1) + require.Contains(t, f.log.warns[0], "autoclaim l2-to-lx bridge detector: detection for destination 10 "+ + "paused for 1s after rate limiting (HTTP 429, attempt 1)") +} + +func TestL2ToLx_BackoffNotUpdatedWhenContextCancelled(t *testing.T) { + f := newIsolationFixture(t, errors.New("boom")) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + result, err := f.detector.PollOnce(ctx) + require.Error(t, err) + require.Equal(t, []uint32{isoFailing}, result.FailedDestinations) + require.Equal(t, 0, f.detector.backoff.stalledCount()) + require.Empty(t, f.log.warns) + require.Equal(t, f.lerX, f.pairLER(t, isoFailing)) + requireLERCursorInvariant(t, f) +} + +func TestL2ToLx_BackoffMetrics(t *testing.T) { + prometheus.Init() + autoclaimmetrics.Register() + ctx := context.Background() + f := newIsolationFixture(t, errors.New("boom")) + + gauge, ok := prometheus.Gauge("l2_to_lx_stalled_destinations") // the wrapper keys series by bare name + require.True(t, ok) + vec, ok := prometheus.CounterVec("detector_destination_errors_total") + require.True(t, ok) + errCounter := vec.WithLabelValues(string(autoclaimmetrics.DetectorL2ToLx), "10") + before := testutil.ToFloat64(errCounter) // counters are process-global: assert deltas + + steps := []struct { + name string + recover bool + advance time.Duration + wantErr bool + wantGauge float64 + wantErrInc float64 + }{ + {name: "failure opens a window", wantErr: true, wantGauge: 1, wantErrInc: 1}, + {name: "poll inside the window adds nothing", wantGauge: 1, wantErrInc: 1}, + {name: "recovery clears the gauge", recover: true, advance: time.Second, wantGauge: 0, wantErrInc: 1}, + } + for _, step := range steps { + if step.recover { + f.failing.claimErr = nil + } + *f.clock = f.clock.Add(step.advance) + _, err := f.detector.PollOnce(ctx) + if step.wantErr { + require.Error(t, err, step.name) + } else { + require.NoError(t, err, step.name) + } + require.Equal(t, step.wantGauge, testutil.ToFloat64(gauge), step.name) + require.Equal(t, before+step.wantErrInc, testutil.ToFloat64(errCounter), step.name) + } +} diff --git a/autoclaim/metrics/metrics.go b/autoclaim/metrics/metrics.go new file mode 100644 index 000000000..2687a786d --- /dev/null +++ b/autoclaim/metrics/metrics.go @@ -0,0 +1,82 @@ +// Package metrics exposes the Prometheus series of the autoclaim bridge detectors. +package metrics + +import ( + "strconv" + + "github.com/agglayer/aggkit/prometheus" + prometheusclient "github.com/prometheus/client_golang/prometheus" +) + +// Detector identifies the bridge detector a metric sample belongs to. +type Detector string + +const ( + // DetectorL1ToL2 is the L1-to-L2 bridge detector. + DetectorL1ToL2 Detector = "l1_to_l2" + // DetectorL2ToLx is the L2-to-Lx bridge detector. + DetectorL2ToLx Detector = "l2_to_lx" + + namespace = "autoclaim" + l1ToL2StalledDestinationsName = "l1_to_l2_stalled_destinations" + l2ToLxStalledDestinationsName = "l2_to_lx_stalled_destinations" + destinationErrorsName = "detector_destination_errors_total" + skippedAlreadyClaimedName = "detector_skipped_already_claimed_total" + detectorLabel = "detector" + destinationLabel = "destination" +) + +// Register registers the autoclaim detector metrics. It is a no-op until prometheus.Init() has run, +// and a duplicate registration only logs a warning. +func Register() { + prometheus.RegisterGauges( + prometheusclient.GaugeOpts{ + Namespace: namespace, + Name: l1ToL2StalledDestinationsName, + Help: "Number of destinations currently failing or backed off in the L1-to-L2 bridge detector", + }, + prometheusclient.GaugeOpts{ + Namespace: namespace, + Name: l2ToLxStalledDestinationsName, + Help: "Number of destinations currently failing or backed off in the L2-to-Lx bridge detector", + }, + ) + prometheus.RegisterCounterVecs( + prometheus.CounterVecOpts{ + CounterOpts: prometheusclient.CounterOpts{ + Namespace: namespace, + Name: destinationErrorsName, + Help: "Number of failing detector polls per detector and destination (one per back-off window)", + }, + Labels: []string{detectorLabel, destinationLabel}, + }, + prometheus.CounterVecOpts{ + CounterOpts: prometheusclient.CounterOpts{ + Namespace: namespace, + Name: skippedAlreadyClaimedName, + Help: "Number of bridges skipped by a detector because they were already claimed", + }, + Labels: []string{detectorLabel, destinationLabel}, + }, + ) +} + +// SetStalledDestinations sets the number of stalled destinations of the given detector. +func SetStalledDestinations(detector Detector, n int) { + switch detector { + case DetectorL1ToL2: + prometheus.GaugeSet(l1ToL2StalledDestinationsName, float64(n)) + case DetectorL2ToLx: + prometheus.GaugeSet(l2ToLxStalledDestinationsName, float64(n)) + } +} + +// IncDestinationError increments the error counter of the given detector and destination network. +func IncDestinationError(detector Detector, destination uint32) { + prometheus.CounterVecInc(destinationErrorsName, string(detector), strconv.FormatUint(uint64(destination), 10)) +} + +// IncSkippedAlreadyClaimed increments the already-claimed skip counter of the given detector and destination. +func IncSkippedAlreadyClaimed(detector Detector, destination uint32) { + prometheus.CounterVecInc(skippedAlreadyClaimedName, string(detector), strconv.FormatUint(uint64(destination), 10)) +} diff --git a/autoclaim/metrics/metrics_test.go b/autoclaim/metrics/metrics_test.go new file mode 100644 index 000000000..0ce81c56e --- /dev/null +++ b/autoclaim/metrics/metrics_test.go @@ -0,0 +1,70 @@ +package metrics + +import ( + "testing" + + "github.com/agglayer/aggkit/prometheus" + prometheusclient "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" +) + +func TestRegister_ExposesSeries(t *testing.T) { + prometheus.Init() + Register() + require.NotPanics(t, Register, "duplicate registration must not panic") + + for _, name := range []string{l1ToL2StalledDestinationsName, l2ToLxStalledDestinationsName} { + _, ok := prometheus.Gauge(name) + require.True(t, ok, name) + } + for _, name := range []string{destinationErrorsName, skippedAlreadyClaimedName} { + _, ok := prometheus.CounterVec(name) + require.True(t, ok, name) + } + + // Series appear in the default registry once they have a sample. + IncDestinationError(DetectorL2ToLx, 4242) + IncSkippedAlreadyClaimed(DetectorL2ToLx, 4242) + SetStalledDestinations(DetectorL2ToLx, 0) + families, err := prometheusclientGather() + require.NoError(t, err) + for _, want := range []string{ + "autoclaim_l1_to_l2_stalled_destinations", + "autoclaim_l2_to_lx_stalled_destinations", + "autoclaim_detector_destination_errors_total", + "autoclaim_detector_skipped_already_claimed_total", + } { + require.Contains(t, families, want) + } +} + +func TestHelpers_UpdateSeries(t *testing.T) { + prometheus.Init() + Register() + + SetStalledDestinations(DetectorL2ToLx, 3) + gauge, ok := prometheus.Gauge(l2ToLxStalledDestinationsName) + require.True(t, ok) + require.Equal(t, 3.0, testutil.ToFloat64(gauge)) + + vec, ok := prometheus.CounterVec(destinationErrorsName) + require.True(t, ok) + counter := vec.WithLabelValues(string(DetectorL2ToLx), "7777") + before := testutil.ToFloat64(counter) + IncDestinationError(DetectorL2ToLx, 7777) + require.Equal(t, before+1, testutil.ToFloat64(counter)) +} + +// prometheusclientGather returns the names of the metric families in the default registry. +func prometheusclientGather() ([]string, error) { + mfs, err := prometheusclient.DefaultGatherer.Gather() + if err != nil { + return nil, err + } + names := make([]string, 0, len(mfs)) + for _, mf := range mfs { + names = append(names, mf.GetName()) + } + return names, nil +} diff --git a/autoclaim/runtime/runtime.go b/autoclaim/runtime/runtime.go index 035eb4b18..6eba58198 100644 --- a/autoclaim/runtime/runtime.go +++ b/autoclaim/runtime/runtime.go @@ -15,6 +15,7 @@ import ( "github.com/agglayer/aggkit/autoclaim/bridgedetector" "github.com/agglayer/aggkit/autoclaim/claimer" autoclaimcfg "github.com/agglayer/aggkit/autoclaim/config" + autoclaimmetrics "github.com/agglayer/aggkit/autoclaim/metrics" "github.com/agglayer/aggkit/autoclaim/policy" "github.com/agglayer/aggkit/autoclaim/proof" "github.com/agglayer/aggkit/autoclaim/sender" @@ -344,6 +345,7 @@ func Start(ctx context.Context, deps Dependencies, factories Factories) (*Runtim } factories = withDefaultFactories(factories, deps.LogConfig) + autoclaimmetrics.Register() logger := deps.Logger if logger == nil { logger = log.WithFields("module", aggkitcommon.AUTOCLAIM) diff --git a/common/backoff.go b/common/backoff.go new file mode 100644 index 000000000..44c2079cd --- /dev/null +++ b/common/backoff.go @@ -0,0 +1,140 @@ +package common + +import ( + "context" + "fmt" + "math" + "math/rand/v2" + "time" +) + +// ExponentialBackoffConfig configures an ExponentialBackoff. +type ExponentialBackoffConfig struct { + // Initial is the delay of the first attempt; must be > 0 + Initial time.Duration + // Max is the hard upper bound of any returned delay; must be >= Initial + Max time.Duration + // Factor is the growth per attempt; must be > 1 + Factor float64 + // Jitter is the symmetric relative jitter in [0, 1); 0.2 means +-20% + Jitter float64 +} + +// Validate reports whether the configuration is usable, wrapping ErrInvalidConfig. +func (c ExponentialBackoffConfig) Validate() error { + if c.Initial <= 0 { + return fmt.Errorf("%w: backoff Initial must be > 0 (got %s)", ErrInvalidConfig, c.Initial) + } + if c.Max < c.Initial { + return fmt.Errorf("%w: backoff Max (%s) must be >= Initial (%s)", ErrInvalidConfig, c.Max, c.Initial) + } + if !(c.Factor > 1) || math.IsInf(c.Factor, 0) { + return fmt.Errorf("%w: backoff Factor must be > 1 (got %v)", ErrInvalidConfig, c.Factor) + } + if !(c.Jitter >= 0 && c.Jitter < 1) { + return fmt.Errorf("%w: backoff Jitter must be in [0, 1) (got %v)", ErrInvalidConfig, c.Jitter) + } + return nil +} + +// SleepFunc sleeps for d or until ctx is done, returning ctx.Err() in the latter case. +type SleepFunc func(ctx context.Context, d time.Duration) error + +// ExponentialBackoff yields capped, jittered exponential delays. Not safe for concurrent use. +type ExponentialBackoff struct { + cfg ExponentialBackoffConfig + attempt int + randFloat func() float64 + sleep SleepFunc +} + +// ExponentialBackoffOption customises an ExponentialBackoff (mainly for tests). +type ExponentialBackoffOption func(*ExponentialBackoff) + +// WithBackoffRandFloat replaces the jitter source; fn must return values in [0, 1). +func WithBackoffRandFloat(fn func() float64) ExponentialBackoffOption { + return func(b *ExponentialBackoff) { + if fn != nil { + b.randFloat = fn + } + } +} + +// WithBackoffSleeper replaces the function used by Sleep to wait. +func WithBackoffSleeper(fn SleepFunc) ExponentialBackoffOption { + return func(b *ExponentialBackoff) { + if fn != nil { + b.sleep = fn + } + } +} + +// NewExponentialBackoff creates an ExponentialBackoff after validating cfg. +func NewExponentialBackoff( + cfg ExponentialBackoffConfig, opts ...ExponentialBackoffOption, +) (*ExponentialBackoff, error) { + if err := cfg.Validate(); err != nil { + return nil, err + } + b := &ExponentialBackoff{ + cfg: cfg, + randFloat: rand.Float64, + sleep: SleepContext, + } + for _, opt := range opts { + opt(b) + } + return b, nil +} + +// Next returns the delay for the current attempt and advances the attempt counter. +// +// base = min(Initial * Factor^attempt, Max) +// delay = base * (1 + Jitter*(2*r-1)), r in [0,1) +// delay = min(delay, Max) +func (b *ExponentialBackoff) Next() time.Duration { + maxF := float64(b.cfg.Max) + base := float64(b.cfg.Initial) * math.Pow(b.cfg.Factor, float64(b.attempt)) + if base > maxF || math.IsInf(base, 0) || math.IsNaN(base) { + base = maxF + } + if b.attempt < math.MaxInt32 { + b.attempt++ + } + delay := base * (1 + b.cfg.Jitter*(2*b.randFloat()-1)) + if delay > maxF { + delay = maxF + } + return time.Duration(delay) +} + +// Sleep calls Next and sleeps that long; it returns the delay and ctx.Err() if interrupted. +func (b *ExponentialBackoff) Sleep(ctx context.Context) (time.Duration, error) { + d := b.Next() + return d, b.sleep(ctx, d) +} + +// Reset sets the attempt counter back to 0. +func (b *ExponentialBackoff) Reset() { + b.attempt = 0 +} + +// Attempts returns the number of Next calls since construction or the last Reset. +func (b *ExponentialBackoff) Attempts() int { + return b.attempt +} + +// SleepContext sleeps for d using a timer, returning early with ctx.Err() when ctx is done. +func SleepContext(ctx context.Context, d time.Duration) error { + if err := ctx.Err(); err != nil { + return err + } + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} diff --git a/common/backoff_test.go b/common/backoff_test.go new file mode 100644 index 000000000..d622cc24d --- /dev/null +++ b/common/backoff_test.go @@ -0,0 +1,138 @@ +package common + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func testBackoffConfig(jitter float64) ExponentialBackoffConfig { + return ExponentialBackoffConfig{Initial: 100 * time.Millisecond, Max: time.Second, Factor: 2, Jitter: jitter} +} + +func TestExponentialBackoffConfig_Validate(t *testing.T) { + tests := []struct { + name string + cfg ExponentialBackoffConfig + wantErr bool + }{ + {"valid", testBackoffConfig(0.2), false}, + {"zero jitter", testBackoffConfig(0), false}, + {"initial zero", ExponentialBackoffConfig{Initial: 0, Max: time.Second, Factor: 2}, true}, + {"max below initial", ExponentialBackoffConfig{Initial: time.Second, Max: time.Millisecond, Factor: 2}, true}, + {"factor one", ExponentialBackoffConfig{Initial: time.Millisecond, Max: time.Second, Factor: 1}, true}, + {"jitter one", ExponentialBackoffConfig{Initial: time.Millisecond, Max: time.Second, Factor: 2, Jitter: 1}, true}, + {"jitter negative", ExponentialBackoffConfig{Initial: time.Millisecond, Max: time.Second, Factor: 2, Jitter: -0.1}, true}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.cfg.Validate() + _, ctorErr := NewExponentialBackoff(tt.cfg) + if tt.wantErr { + require.ErrorIs(t, err, ErrInvalidConfig) + require.ErrorIs(t, ctorErr, ErrInvalidConfig) + return + } + require.NoError(t, err) + require.NoError(t, ctorErr) + }) + } +} + +func TestExponentialBackoff_GrowthCapReset(t *testing.T) { + b, err := NewExponentialBackoff(testBackoffConfig(0), WithBackoffRandFloat(func() float64 { return 0.5 })) + require.NoError(t, err) + + expected := []time.Duration{100, 200, 400, 800, 1000, 1000, 1000} + for i, e := range expected { + require.Equal(t, e*time.Millisecond, b.Next(), "attempt %d", i) + require.Equal(t, i+1, b.Attempts()) + } + + b.Reset() + require.Zero(t, b.Attempts()) + require.Equal(t, 100*time.Millisecond, b.Next()) +} + +func TestExponentialBackoff_HugeAttemptDoesNotOverflow(t *testing.T) { + b, err := NewExponentialBackoff(testBackoffConfig(0), WithBackoffRandFloat(func() float64 { return 0.5 })) + require.NoError(t, err) + b.attempt = 1 << 20 + require.Equal(t, time.Second, b.Next()) +} + +func TestExponentialBackoff_JitterBounds(t *testing.T) { + const jitter = 0.2 + tests := []struct { + name string + r float64 + }{ + {"lowest", 0}, + {"middle", 0.5}, + {"highest", 0.999999}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + b, err := NewExponentialBackoff(testBackoffConfig(jitter), WithBackoffRandFloat(func() float64 { return tt.r })) + require.NoError(t, err) + for i := 0; i < 8; i++ { + base := float64(100*time.Millisecond) * float64(int(1)< float64(time.Second) { + base = float64(time.Second) + } + lo := time.Duration(base * (1 - jitter)) + hi := time.Duration(base * (1 + jitter)) + if hi > time.Second { + hi = time.Second + } + got := b.Next() + require.GreaterOrEqual(t, got, lo-time.Nanosecond, "attempt %d", i) + require.LessOrEqual(t, got, hi, "attempt %d", i) + } + }) + } +} + +func TestExponentialBackoff_SleepUsesInjectedSleeper(t *testing.T) { + var slept []time.Duration + b, err := NewExponentialBackoff(testBackoffConfig(0), + WithBackoffRandFloat(func() float64 { return 0.5 }), + WithBackoffSleeper(func(_ context.Context, d time.Duration) error { + slept = append(slept, d) + return nil + })) + require.NoError(t, err) + + for i := 0; i < 3; i++ { + d, err := b.Sleep(context.Background()) + require.NoError(t, err) + require.Equal(t, slept[i], d) + } + require.Equal(t, []time.Duration{100 * time.Millisecond, 200 * time.Millisecond, 400 * time.Millisecond}, slept) +} + +func TestExponentialBackoff_SleepCtxCancelled(t *testing.T) { + b, err := NewExponentialBackoff(ExponentialBackoffConfig{Initial: time.Hour, Max: time.Hour, Factor: 2}) + require.NoError(t, err) + + ctx, cancel := context.WithCancel(context.Background()) + go func() { + time.Sleep(10 * time.Millisecond) + cancel() + }() + + start := time.Now() + _, err = b.Sleep(ctx) + require.ErrorIs(t, err, context.Canceled) + require.Less(t, time.Since(start), time.Second) +} + +func TestSleepContext(t *testing.T) { + require.NoError(t, SleepContext(context.Background(), time.Millisecond)) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + require.ErrorIs(t, SleepContext(ctx, time.Hour), context.Canceled) +} diff --git a/common/errors.go b/common/errors.go index 782b5f861..14463d6fd 100644 --- a/common/errors.go +++ b/common/errors.go @@ -1,8 +1,12 @@ package common import ( + "errors" + "net/http" "regexp" "strings" + + "github.com/ethereum/go-ethereum/rpc" ) const maxRangeMatchGroups = 2 @@ -16,6 +20,17 @@ var ( reEthGetLogsLimited = regexp.MustCompile(`eth_getLogs is limited to a\s+([\d,]+)\s+range`) // Matches "query exceeds max block range 100000" reQueryExceedsMaxBlockRange = regexp.MustCompile(`query exceeds max block range\s+(\d+)`) + + // Matches "too many batch requests, max is 40" + reTooManyBatchRequests = regexp.MustCompile(`too many batch requests, max is\s*(\d+)`) + // Matches "batch limit 100 exceeded" + reBatchLimitExceeded = regexp.MustCompile(`batch limit (\d+) exceeded`) + // Matches "batch of more than 100 requests" + reBatchOfMoreThan = regexp.MustCompile(`(?i)batch of more than (\d+) requests`) + // Matches batch-limit messages that do not carry the limit (the caller must reduce the size by itself) + reBatchLimitKeyword = regexp.MustCompile( + `(?i)(batch too large|too many batch requests|batch size (limit|too large|exceeded)|` + + `batch (request )?limit (exceeded|reached)|batch request (was )?too large)`) ) // ParseMaxRangeFromError extracts the max range value from error message @@ -48,3 +63,42 @@ func ParseMaxRangeFromError(errMsg string) (uint64, bool) { return maxRange, true } + +// ParseMaxBatchSizeFromError extracts a provider's maximum JSON-RPC batch size from an error message. +// Expected formats: +// - "too many batch requests, max is 40" +// - "batch limit 100 exceeded" +// - "batch of more than 100 requests" +// +// It returns false when no pattern matches or when the parsed value is 0. +func ParseMaxBatchSizeFromError(errMsg string) (uint64, bool) { + for _, re := range []*regexp.Regexp{reTooManyBatchRequests, reBatchLimitExceeded, reBatchOfMoreThan} { + matches := re.FindStringSubmatch(errMsg) + if len(matches) < maxRangeMatchGroups { + continue + } + maxSize, err := ParseUint64HexOrDecimal(matches[1]) + if err != nil || maxSize == 0 { + return 0, false + } + return maxSize, true + } + return 0, false +} + +// IsBatchLimitError reports whether err means that a JSON-RPC batch was too large for the provider. +// HTTP 429 is deliberately not considered a batch-limit error. +func IsBatchLimitError(err error) bool { + if err == nil { + return false + } + msg := err.Error() + if _, ok := ParseMaxBatchSizeFromError(msg); ok { + return true + } + if reBatchLimitKeyword.MatchString(msg) { + return true + } + var httpErr rpc.HTTPError + return errors.As(err, &httpErr) && httpErr.StatusCode == http.StatusRequestEntityTooLarge +} diff --git a/common/errors_test.go b/common/errors_test.go index f5af8d543..bc7b3af0f 100644 --- a/common/errors_test.go +++ b/common/errors_test.go @@ -5,6 +5,7 @@ import ( "fmt" "testing" + "github.com/ethereum/go-ethereum/rpc" "github.com/stretchr/testify/require" ) @@ -168,3 +169,58 @@ func TestParseMaxRangeFromError_WrappedErrors(t *testing.T) { require.Equal(t, uint64(2500), result) }) } + +func TestParseMaxBatchSizeFromError(t *testing.T) { + tests := []struct { + name string + msg string + expected uint64 + ok bool + }{ + {"issue string", "400 Bad Request: too many batch requests, max is 40", 40, true}, + {"no space after is", "too many batch requests, max is100", 100, true}, + {"batch limit exceeded", "batch limit 100 exceeded", 100, true}, + {"batch of more than", "Batch of more than 50 requests is not allowed", 50, true}, + {"zero limit", "too many batch requests, max is 0", 0, false}, + {"keyword without number", "batch too large", 0, false}, + {"unrelated", "connection refused", 0, false}, + {"empty", "", 0, false}, + {"overflow", "too many batch requests, max is 99999999999999999999999", 0, false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got, ok := ParseMaxBatchSizeFromError(tt.msg) + require.Equal(t, tt.ok, ok) + require.Equal(t, tt.expected, got) + }) + } +} + +func TestIsBatchLimitError(t *testing.T) { + tests := []struct { + name string + err error + expected bool + }{ + {"nil", nil, false}, + {"issue string", errors.New("400 Bad Request: too many batch requests, max is 40"), true}, + {"wrapped HTTPError with limit", + fmt.Errorf("wrap: %w", rpc.HTTPError{StatusCode: 400, Status: "400 Bad Request", + Body: []byte("too many batch requests, max is 40")}), true}, + {"geth batch too large", errors.New("batch too large"), true}, + {"keyword without number", errors.New("too many batch requests"), true}, + {"HTTP 413", fmt.Errorf("wrap: %w", rpc.HTTPError{StatusCode: 413, Status: "413 Request Entity Too Large"}), true}, + {"HTTP 429", rpc.HTTPError{StatusCode: 429, Status: "429 Too Many Requests"}, false}, + {"connection refused", errors.New("connection refused"), false}, + {"response too large", errors.New("response too large"), false}, + {"max is 0", errors.New("some error, max is 0"), false}, + {"batch limit exceeded without number", errors.New("batch limit exceeded"), true}, + {"batch rate limit is not a size limit", errors.New("batch request limit per second, retry later"), false}, + {"unrelated rate limit", errors.New("rate limit reached for batch endpoint"), false}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + require.Equal(t, tt.expected, IsBatchLimitError(tt.err)) + }) + } +} diff --git a/common/repeated_error_logger.go b/common/repeated_error_logger.go new file mode 100644 index 000000000..71ec78819 --- /dev/null +++ b/common/repeated_error_logger.go @@ -0,0 +1,98 @@ +package common + +import ( + "regexp" + "sync" + "time" + + "github.com/agglayer/aggkit/log" +) + +// reVariablePart matches the parts of an error message that usually change between attempts of the same +// failure (block numbers, ranges, hashes, request ids), so they do not make each retry a "distinct" error. +// Only numbers of four or more digits are replaced, so short numbers such as HTTP status codes (429 vs +// 500) keep two different failures distinguishable. +var reVariablePart = regexp.MustCompile(`0x[0-9a-fA-F]+|\d{4,}`) + +// RepeatedErrorLogger bounds the log volume of an error that may repeat indefinitely in a retry loop. +// Messages are compared after replacing long numbers and hex strings, so errors that only differ in them are +// considered the same. The first occurrence of a distinct message is logged at Warn, repeats at Debug, +// and a Warn summary carrying the number of occurrences is emitted once per window. +type RepeatedErrorLogger struct { + log Logger + prefix string + window time.Duration + now func() time.Time + mu sync.Mutex + lastMsg string // normalised key of the last error + lastErr string // text of the last error, for the Reset summary + hasLast bool + suppressed int + lastWarn time.Time +} + +// NewRepeatedErrorLogger creates a RepeatedErrorLogger. A nil logger is replaced by a no-op logger. +func NewRepeatedErrorLogger(logger Logger, prefix string, window time.Duration) *RepeatedErrorLogger { + if logger == nil { + logger = log.NewLoggerNil() + } + return &RepeatedErrorLogger{ + log: logger, + prefix: prefix, + window: window, + now: time.Now, + } +} + +// WithClock replaces the clock (tests); it returns the receiver. +func (l *RepeatedErrorLogger) WithClock(now func() time.Time) *RepeatedErrorLogger { + l.mu.Lock() + defer l.mu.Unlock() + if now != nil { + l.now = now + } + return l +} + +// Log records one occurrence of err. A nil err is ignored. +func (l *RepeatedErrorLogger) Log(err error) { + if err == nil { + return + } + l.mu.Lock() + defer l.mu.Unlock() + + key := reVariablePart.ReplaceAllString(err.Error(), "#") + now := l.now() + l.lastErr = err.Error() + switch { + case !l.hasLast || key != l.lastMsg: + l.log.Warnf("%s: %v", l.prefix, err) + l.lastMsg, l.hasLast = key, true + l.suppressed = 0 + l.lastWarn = now + case now.Sub(l.lastWarn) >= l.window: + l.log.Warnf("%s: %v (repeated %d times in the last %s)", + l.prefix, err, l.suppressed+1, now.Sub(l.lastWarn).Truncate(time.Second)) + l.suppressed = 0 + l.lastWarn = now + default: + l.log.Debugf("%s: %v", l.prefix, err) + l.suppressed++ + } +} + +// Reset forgets the current error (call after a success) so the next error is logged at Warn again. +// If occurrences were suppressed since the last Warn, it emits one Warn summary with their count; +// otherwise it logs nothing. +func (l *RepeatedErrorLogger) Reset() { + l.mu.Lock() + defer l.mu.Unlock() + if l.hasLast && l.suppressed > 0 { + l.log.Warnf("%s: %q repeated %d more times in the last %s before recovering", + l.prefix, l.lastErr, l.suppressed, l.now().Sub(l.lastWarn).Truncate(time.Second)) + } + l.hasLast = false + l.lastMsg = "" + l.suppressed = 0 +} diff --git a/common/repeated_error_logger_test.go b/common/repeated_error_logger_test.go new file mode 100644 index 000000000..3bdc2834c --- /dev/null +++ b/common/repeated_error_logger_test.go @@ -0,0 +1,174 @@ +package common + +import ( + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +type loggedLine struct { + level string + msg string +} + +// captureLogger implements Logger, recording only the Warn/Debug lines used by RepeatedErrorLogger. +type captureLogger struct { + mu sync.Mutex + lines []loggedLine +} + +func (c *captureLogger) add(level, format string, args ...interface{}) { + c.mu.Lock() + defer c.mu.Unlock() + c.lines = append(c.lines, loggedLine{level, fmt.Sprintf(format, args...)}) +} + +func (c *captureLogger) levels() []string { + c.mu.Lock() + defer c.mu.Unlock() + res := make([]string, 0, len(c.lines)) + for _, l := range c.lines { + res = append(res, l.level) + } + return res +} + +func (c *captureLogger) Panicf(string, ...interface{}) {} +func (c *captureLogger) Fatalf(string, ...interface{}) {} +func (c *captureLogger) Info(...interface{}) {} +func (c *captureLogger) Infof(string, ...interface{}) {} +func (c *captureLogger) Error(...interface{}) {} +func (c *captureLogger) Errorf(string, ...interface{}) {} +func (c *captureLogger) Warn(args ...interface{}) { c.add("warn", "%s", fmt.Sprint(args...)) } +func (c *captureLogger) Warnf(f string, a ...interface{}) { + c.add("warn", f, a...) +} +func (c *captureLogger) Debug(args ...interface{}) { c.add("debug", "%s", fmt.Sprint(args...)) } +func (c *captureLogger) Debugf(f string, a ...interface{}) { + c.add("debug", f, a...) +} + +func TestRepeatedErrorLogger_LevelSequence(t *testing.T) { + logger := &captureLogger{} + now := time.Unix(1000, 0) + rl := NewRepeatedErrorLogger(logger, "step", time.Minute).WithClock(func() time.Time { return now }) + errA := errors.New("boom") + + // First occurrence: Warn; repeats inside the window: Debug. + rl.Log(errA) + for i := 0; i < 3; i++ { + now = now.Add(10 * time.Second) + rl.Log(errA) + } + require.Equal(t, []string{"warn", "debug", "debug", "debug"}, logger.levels()) + + // Window elapsed: summary Warn with the count (3 suppressed + this one). + now = now.Add(40 * time.Second) + rl.Log(errA) + levels := logger.levels() + require.Equal(t, "warn", levels[len(levels)-1]) + require.True(t, strings.Contains(logger.lines[len(logger.lines)-1].msg, "repeated 4 times in the last 1m10s"), + logger.lines[len(logger.lines)-1].msg) + + // Next repeat is Debug again. + now = now.Add(time.Second) + rl.Log(errA) + require.Equal(t, "debug", logger.levels()[len(logger.lines)-1]) + + // A distinct message is a Warn immediately. + rl.Log(errors.New("other")) + require.Equal(t, "warn", logger.levels()[len(logger.lines)-1]) + + // After Reset the same message is a Warn again, and Reset itself is silent. + n := len(logger.lines) + rl.Reset() + require.Len(t, logger.lines, n) + rl.Log(errors.New("other")) + require.Equal(t, "warn", logger.levels()[n]) +} + +func TestRepeatedErrorLogger_WarnCountEqualsWindows(t *testing.T) { + logger := &captureLogger{} + now := time.Unix(0, 0) + rl := NewRepeatedErrorLogger(logger, "p", time.Minute).WithClock(func() time.Time { return now }) + err := errors.New("same") + for i := 0; i < 301; i++ { // 5 minutes at one call per second + rl.Log(err) + now = now.Add(time.Second) + } + warns := 0 + for _, l := range logger.levels() { + if l == "warn" { + warns++ + } + } + require.Equal(t, 1+5, warns) +} + +func TestRepeatedErrorLogger_NilCases(t *testing.T) { + logger := &captureLogger{} + NewRepeatedErrorLogger(logger, "p", time.Minute).Log(nil) + require.Empty(t, logger.lines) + + require.NotPanics(t, func() { + NewRepeatedErrorLogger(nil, "p", time.Minute).Log(errors.New("x")) + }) +} + +func TestRepeatedErrorLogger_VaryingNumbersAreTheSameError(t *testing.T) { + logger := &captureLogger{} + now := time.Unix(0, 0) + rl := NewRepeatedErrorLogger(logger, "p", time.Minute).WithClock(func() time.Time { return now }) + + rl.Log(errors.New("cannot update ToBlock: block 123457 (0xabc1) not found")) + now = now.Add(time.Second) + rl.Log(errors.New("cannot update ToBlock: block 123458 (0xdef2) not found")) + now = now.Add(time.Second) + rl.Log(errors.New("cannot update ToBlock: block 123459 (0x1234) not found")) + require.Equal(t, []string{"warn", "debug", "debug"}, logger.levels()) + + // A really different message is still a Warn. + rl.Log(errors.New("connection refused")) + require.Equal(t, "warn", logger.levels()[3]) +} + +func TestRepeatedErrorLogger_HTTPStatusCodesStayDistinct(t *testing.T) { + logger := &captureLogger{} + now := time.Unix(0, 0) + rl := NewRepeatedErrorLogger(logger, "p", time.Minute).WithClock(func() time.Time { return now }) + + rl.Log(errors.New("429 Too Many Requests")) + now = now.Add(time.Second) + rl.Log(errors.New("500 Internal Server Error")) + now = now.Add(time.Second) + rl.Log(errors.New("500 Internal Server Error")) + require.Equal(t, []string{"warn", "warn", "debug"}, logger.levels()) +} + +func TestRepeatedErrorLogger_ResetEmitsSummaryOfSuppressed(t *testing.T) { + logger := &captureLogger{} + now := time.Unix(0, 0) + rl := NewRepeatedErrorLogger(logger, "p", time.Minute).WithClock(func() time.Time { return now }) + err := errors.New("boom") + + rl.Log(err) // warn + rl.Log(err) // debug (suppressed 1) + rl.Log(err) // debug (suppressed 2) + n := len(logger.lines) + now = now.Add(5 * time.Second) + rl.Reset() + require.Len(t, logger.lines, n+1) + require.Equal(t, "warn", logger.levels()[n]) + require.Contains(t, logger.lines[n].msg, "repeated 2 more times") + + // Nothing suppressed: Reset stays silent. + rl.Log(err) + n = len(logger.lines) + rl.Reset() + require.Len(t, logger.lines, n) +} diff --git a/config/config_test.go b/config/config_test.go index 98727c0f7..f4f10b4d3 100644 --- a/config/config_test.go +++ b/config/config_test.go @@ -63,8 +63,10 @@ func TestLoadDefaultConfig(t *testing.T) { Mode: ethermanconfig.RPCModeBasic, HashFromJSON: true, BatchBlockHeaderRetrieval: true, + BatchRequestMaxSize: ethermanconfig.DefaultBatchRequestMaxSize, }, cfg.Common.L2RPC) require.True(t, cfg.L1NetworkConfig.RPC.BatchBlockHeaderRetrieval) + require.Equal(t, ethermanconfig.DefaultBatchRequestMaxSize, cfg.L1NetworkConfig.RPC.BatchRequestMaxSize) require.Equal(t, cfg.Profiling.ProfilingEnabled, false) require.Equal(t, cfg.Profiling.ProfilingHost, "localhost") require.Equal(t, cfg.Profiling.ProfilingPort, 6060) diff --git a/config/default.go b/config/default.go index ae32bca37..4715995de 100644 --- a/config/default.go +++ b/config/default.go @@ -53,6 +53,7 @@ defaultDBQueryTimeout = "5m" BackoffMultiplier = 2.0 HashFromJSON = true BatchBlockHeaderRetrieval = true + BatchRequestMaxSize = 1000 ` // DefaultValues is the default configuration @@ -85,6 +86,7 @@ BlocksChunkSize = {{L1Config.BlocksChunkSize}} BackoffMultiplier = 2.0 HashFromJSON = true BatchBlockHeaderRetrieval = true + BatchRequestMaxSize = 1000 [L2NetworkConfig] # InitialLER: optional override for the initial Local Exit Root (0x000...000 is a valid value). diff --git a/docs/autoclaim.md b/docs/autoclaim.md index dc0a5ad40..f41c10506 100644 --- a/docs/autoclaim.md +++ b/docs/autoclaim.md @@ -851,12 +851,65 @@ is created for GER tracking. the relevant leaf. - Auto Claim logs startup, API startup, bridge detector polling errors, claimer recovery errors, and per-request errors through the standard Aggkit logger. Request-level error details are also stored in `last_error` and exposed - by the API. The component does not export Prometheus metrics. + by the API. The bridge detectors export a small set of Prometheus metrics (see + [RPC failure handling](#rpc-failure-handling)). - Failed or evicted transaction-manager results are retried while retry budget remains. Exhausted requests become `failed` and require operator investigation. - Use `api-approve` when an operator must explicitly inspect each request before claim submission. Expose the API only on trusted networks or behind access controls; it can approve or reject pending manual requests. +### RPC failure handling + +A failing RPC endpoint of one destination network must not stall detection for the others. Both bridge detectors +(L1-to-L2 and L2-to-Lx) isolate failures per destination: + +- Only failures of the destination's own calls are soft: `IsClaimed` and claimer `Enqueue` for the L1-to-L2 + detector, and `IsClaimed` for the L2-to-Lx detector. Other errors (local database, claimer registry, source + fetch, cursor persistence) still abort the poll as before. +- A destination that fails is paused with a per-destination back-off: the first window is the detector poll period, + doubling on each consecutive failure up to 2 minutes, with +-20 % jitter. A successful poll resets it. HTTP `429` + uses the same curve. A server-provided `Retry-After` is not honoured, because go-ethereum drops the response + headers before the error reaches the detector. +- While a destination is paused its cursor does not advance, so nothing is skipped; healthy destinations keep + advancing normally. When the pause ends, detection resumes from the held cursor. +- Known L2-to-Lx liveness limitation: while a destination stays failed, the shared block window restarts at that + source's verify row on every poll, so verify rows more than `BlockWindow` (default `1000`) L1 blocks later are not + reached until the destination recovers. This is the same behaviour as the existing finder-miss path. + +Log lines to grep for (`` is the destination network ID; the L2-to-Lx detector uses the prefix +`autoclaim l2-to-lx bridge detector:`): + +```text +autoclaim l1-to-l2 bridge detector: detection for destination paused for after error (attempt ): +autoclaim l1-to-l2 bridge detector: detection for destination paused for after rate limiting (HTTP 429, attempt ): +autoclaim l1-to-l2 bridge detector: skipping destination , backed off until