chainsource: escalate unrecoverable block epoch loss to the daemon

chainsource/block_epoch_actor.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"log/slog"
+	"math/rand/v2"
 	"sync"
 	"time"
 
@@ -24,8 +25,29 @@ const (
 	// defaultBlockEpochMaxReconnectBackoff caps reconnect backoff so a
 	// backend outage does not spin, but recovery still happens promptly.
 	defaultBlockEpochMaxReconnectBackoff = 30 * time.Second
+
+	// defaultBlockEpochFatalTimeout bounds how long a subscription may stay
+	// continuously down before the actor stops retrying and escalates. A
+	// notifier restart heals in seconds; a streak this long means the
+	// backend connection is stuck (the production symptom was dozens of
+	// subscribers re-registering against a dead notifier every 30s forever,
+	// which no amount of in-process retrying can recover). Escalating lets
+	// the daemon exit and restart with a fresh backend connection.
+	defaultBlockEpochFatalTimeout = 5 * time.Minute
+
+	// defaultBlockEpochFatalJitter is the maximum random padding added to
+	// the fatal timeout on the default-timeout path. A backend outage tends
+	// to drop every daemon's streams at the same instant; without jitter a
+	// whole fleet would hit the 5m budget together and restart in lockstep
+	// (thundering herd). Spreading the effective budget over a ~1m window
+	// decorrelates those restarts.
+	defaultBlockEpochFatalJitter = time.Minute
 )
 
+// errBlockEpochUnrecoverable is the placeholder cause used when a subscription
+// stays down past the fatal timeout without a more specific reconnect error.
+var errBlockEpochUnrecoverable = fmt.Errorf("block epoch backend unreachable")
+
 // BlockEpochConfig holds configuration for BlockEpochActor.
 type BlockEpochConfig struct {
 	// Backend is the blockchain backend used to monitor blocks.
@@ -44,6 +66,34 @@ type BlockEpochConfig struct {
 	// MaxReconnectBackoff caps the exponential reconnect delay. Zero uses
 	// defaultBlockEpochMaxReconnectBackoff.
 	MaxReconnectBackoff time.Duration
+
+	// FatalReconnectTimeout bounds how long the subscription may stay
+	// continuously down before the actor stops retrying and escalates via
+	// OnFatal. Zero uses defaultBlockEpochFatalTimeout.
+	FatalReconnectTimeout time.Duration
+
+	// OnFatal, when set, is invoked once if the subscription cannot be
+	// re-established within FatalReconnectTimeout. The daemon wires this to
+	// its shutdown path so a stuck backend connection restarts the process
+	// instead of leaving every block subscriber silently spinning forever.
+	OnFatal fn.Option[func(error)]
+
+	// Now returns the current time. Tests override it to drive the fatal
+	// timeout without real waits. Nil uses time.Now.
+	Now func() time.Time
+
+	// FatalReconnectJitter caps the random padding added to the fatal
+	// timeout so a fleet sharing a backend outage does not escalate in
+	// lockstep. On the default-timeout path (FatalReconnectTimeout zero) a
+	// zero jitter uses defaultBlockEpochFatalJitter; an explicitly
+	// configured timeout with zero jitter disables jitter so tests stay
+	// deterministic.
+	FatalReconnectJitter time.Duration
+
+	// Rand returns a pseudo-random fraction in [0, 1) used to seed the
+	// fatal timeout jitter. Nil uses a real random source. Tests override
+	// it for determinism.
+	Rand func() float64
 }
 
 // WithLogger returns a new config with the given logger set.
@@ -222,6 +272,90 @@ func (a *BlockEpochActor) blockEpochReconnectBackoff() (time.Duration,
 	return initial, maxBackoff
 }
 
+// now returns the current time via the configured clock, defaulting to the real
+// clock. It lets tests drive the fatal timeout deterministically.
+func (a *BlockEpochActor) now() time.Time {
+	if a.cfg.Now != nil {
+		return a.cfg.Now()
+	}
+
+	return time.Now()
+}
+
+// randFloat returns a pseudo-random fraction in [0, 1) via the configured
+// source, defaulting to the global generator. It seeds the fatal timeout
+// jitter; tests override it for determinism.
+func (a *BlockEpochActor) randFloat() float64 {
+	if a.cfg.Rand != nil {
+		return a.cfg.Rand()
+	}
+
+	// The jitter only decorrelates restart timing across a fleet; it is not
+	// security-sensitive, so a weak PRNG is fine here.
+	//nolint:gosec // G404: jitter timing, not security-sensitive.
+	return rand.Float64()
+}
+
+// fatalReconnectTimeout returns the normalized down-time budget before the
+// actor escalates a stuck subscription, including a one-shot random jitter so a
+// fleet that loses a shared backend does not escalate in lockstep. It is
+// sampled once at the start of monitoring; reconnectExhausted takes the result
+// as a parameter so the budget stays fixed across the reconnect loop.
+func (a *BlockEpochActor) fatalReconnectTimeout() time.Duration {
+	base := a.cfg.FatalReconnectTimeout
+	jitter := a.cfg.FatalReconnectJitter
+	if base <= 0 {
+		base = defaultBlockEpochFatalTimeout
+
+		// Only the default path opts into jitter automatically; an
+		// explicit timeout stays exact unless jitter is set so tests
+		// remain deterministic.
+		if jitter <= 0 {
+			jitter = defaultBlockEpochFatalJitter
+		}
+	}
+
+	if jitter <= 0 {
+		return base
+	}
+
+	return base + time.Duration(a.randFloat()*float64(jitter))
+}
+
+// reconnectExhausted reports whether the subscription has stayed continuously
+// down past the given fatal timeout. A zero downSince means the subscription is
+// currently healthy. The caller samples the timeout once so jitter does not
+// wobble the budget between checks.
+func (a *BlockEpochActor) reconnectExhausted(downSince time.Time,
+	timeout time.Duration) bool {
+
+	if downSince.IsZero() {
+		return false
+	}
+
+	return a.now().Sub(downSince) >= timeout
+}
+
+// escalateFatal reports an unrecoverable subscription to the daemon via the
+// configured OnFatal hook. cause is the last reconnect error, if any.
+func (a *BlockEpochActor) escalateFatal(log btclog.Logger, downSince time.Time,
+	cause error) {
+
+	down := a.now().Sub(downSince).Round(time.Second)
+	if cause == nil {
+		cause = errBlockEpochUnrecoverable
+	}
+	err := fmt.Errorf("block epoch subscription unrecoverable after %s "+
+		"down: %w", down, cause)
+
+	log.ErrorS(a.ctx, "Block epoch subscription unrecoverable; escalating",
+		err,
+	)
+	a.cfg.OnFatal.WhenSome(func(onFatal func(error)) {
+		onFatal(err)
+	})
+}
+
 // waitForReconnect sleeps for the current backoff unless the actor is
 // stopping. It returns false when shutdown won the race.
 func (a *BlockEpochActor) waitForReconnect(backoff time.Duration) bool {
@@ -264,6 +398,20 @@ func (a *BlockEpochActor) monitorBlocks() {
 	currentBackoff := reconnectBackoff
 	registration := a.registration
 
+	// downSince stamps when the subscription first went down; it is cleared
+	// only once a replacement stream actually delivers a block. lastErr
+	// holds the most recent reconnect failure so the eventual escalation
+	// can surface a meaningful cause. Note the budget covers a stream that
+	// is closed/reconnecting: an open stream that goes silent without
+	// closing keeps the goroutine parked in the receive below and is out of
+	// scope (a healthy backend delivers blocks well within the budget).
+	var downSince time.Time
+	var lastErr error
+
+	// Sample the fatal budget once so its jitter stays fixed across every
+	// reconnect attempt for this subscription.
+	fatalTimeout := a.fatalReconnectTimeout()
+
 	// In iterator mode, the sender (this goroutine) is responsible for
 	// closing the channel to signal the receiver that no more values will
 	// be sent. This follows Go's channel ownership semantics.
@@ -283,13 +431,39 @@ func (a *BlockEpochActor) monitorBlocks() {
 
 	for {
 		if registration == nil {
+			// A normal shutdown cancels the actor context. Bail out
+			// before evaluating the fatal budget so a stop that
+			// coincides with a long outage exits cleanly instead of
+			// escalating a fatal failure the daemon does not really
+			// have (escalation latches the health flag).
+			if a.ctx.Err() != nil {
+				return
+			}
+
+			// If the subscription has stayed down past the fatal
+			// budget, stop retrying and escalate so the daemon can
+			// restart with a fresh backend connection. Only callers
+			// that opted into bounded retries by installing OnFatal
+			// take this path; without a hook there is nowhere to
+			// escalate, so we preserve the unbounded-retry contract
+			// and keep trying to heal in-process rather than
+			// silently killing the subscription.
+			if a.cfg.OnFatal.IsSome() &&
+				a.reconnectExhausted(downSince, fatalTimeout) {
+
+				a.escalateFatal(log, downSince, lastErr)
+
+				return
+			}
+
 			if !a.waitForReconnect(currentBackoff) {
 				return
 			}
 
 			var err error
 			registration, err = a.cfg.Backend.RegisterBlocks(a.ctx)
 			if err != nil {
+				lastErr = err
 				log.WarnS(a.ctx, "Block epoch reconnect failed",
 					err,
 					slog.Duration("backoff",
@@ -309,6 +483,14 @@ func (a *BlockEpochActor) monitorBlocks() {
 		select {
 		case epoch, ok := <-registration.Epochs:
 			if !ok {
+				// Stamp the start of the down streak on the
+				// first loss; a successful re-registration that
+				// immediately closes again must not reset it,
+				// or a storming backend would never escalate.
+				if downSince.IsZero() {
+					downSince = a.now()
+				}
+
 				log.InfoS(
 					a.ctx,
 					"Block epoch channel closed, "+
@@ -322,6 +504,11 @@ func (a *BlockEpochActor) monitorBlocks() {
 				continue
 			}
 
+			// A delivered block proves the subscription is healthy
+			// again, so clear the down streak and last error.
+			downSince = time.Time{}
+			lastErr = nil
+
 			log.InfoS(a.ctx, "Received block from backend",
 				slog.Int("height", int(epoch.Height)),
 			)

chainsource/chainsource.go

@@ -32,6 +32,12 @@ type ChainSourceConfig struct {
 	// falls back to extracting a logger from context via LoggerFromContext,
 	// or uses btclog.Disabled if no logger is found.
 	Log fn.Option[btclog.Logger]
+
+	// OnFatal, when set, is propagated to spawned block epoch subscriptions
+	// so that an unrecoverable backend (a stream that cannot be
+	// re-established within the fatal timeout) escalates to the daemon's
+	// shutdown path instead of spinning forever.
+	OnFatal fn.Option[func(error)]
 }
 
 // WithLogger returns a new config with the given logger set.
@@ -375,10 +381,12 @@ func (a *ChainSourceActor) handleSubscribeBlocks(ctx context.Context,
 	actorID := fmt.Sprintf("epoch.%s", req.CallerID)
 	serviceKey := epochActorServiceKey(req.CallerID)
 
-	// Pass the backend and logger to the sub-actor via config.
+	// Pass the backend and logger to the sub-actor via config. The fatal
+	// hook is forwarded so a wedged backend escalates to daemon shutdown.
 	epochCfg := BlockEpochConfig{
 		Backend: a.cfg.Backend,
 		Log:     fn.Some(a.logger(ctx)),
+		OnFatal: a.cfg.OnFatal,
 	}
 	epochActor := NewBlockEpochActor(epochCfg)
 	actorRef := serviceKey.Spawn(a.cfg.System, actorID, epochActor)