From 03322c9f43a72b011faf95b68778dd113117a0ce Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Thu, 18 Jun 2026 14:22:37 -0700 Subject: [PATCH 1/2] chainsource: escalate unrecoverable block epoch loss to the daemon MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The block epoch reconnect work treats a closed backend stream as transient and re-registers with bounded backoff, which heals a notifier that briefly churns. But it retries forever. When the backend connection is genuinely stuck — the production symptom was dozens of block subscribers all re-registering against a dead notifier every 30s, indefinitely — every subscriber spins in lockstep, no subscription ever heals, and the daemon keeps serving RPCs as a zombie while its readiness probe stays green. On that node a Lightning -> Ark in-swap never completed: the ark client could not hand the swap server a server pubkey, so intercepted HTLCs were canceled back. In this commit, we bound the reconnect. The BlockEpochActor now tracks how long a subscription has been continuously down — stamped when the stream is first lost and cleared only once a replacement actually delivers a block, so a backend that hands back streams which immediately close still counts as down. Once the down streak outlives FatalReconnectTimeout, the actor stops retrying and escalates through a new OnFatal hook instead of spinning. The fatal timeout carries a small one-shot jitter so a fleet that loses a shared backend at the same instant does not all escalate and restart in lockstep. Because every block subscriber escalates through this one hook, the escalation lives in a single place and covers them all. The hook is threaded from the daemon through ChainSourceConfig: the server wires it to a context.CancelCause on the run context, so run() returns the cause, the process exits non-zero, and the orchestrator restarts darepod with a fresh backend connection. Normal signal-driven shutdown still returns nil. A clock is injected so the timeout is exercised in tests without real waits. --- chainsource/block_epoch_actor.go | 187 ++++++++++++++++++++++++++++++ chainsource/chainsource.go | 10 +- chainsource/chainsource_test.go | 188 +++++++++++++++++++++++++++++++ darepod/server.go | 39 ++++++- 4 files changed, 422 insertions(+), 2 deletions(-) diff --git a/chainsource/block_epoch_actor.go b/chainsource/block_epoch_actor.go index a23cb7c00..30ddc1185 100644 --- a/chainsource/block_epoch_actor.go +++ b/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,6 +431,31 @@ 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 } @@ -290,6 +463,7 @@ func (a *BlockEpochActor) monitorBlocks() { 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)), ) diff --git a/chainsource/chainsource.go b/chainsource/chainsource.go index 9115d51ad..c32fb0ebf 100644 --- a/chainsource/chainsource.go +++ b/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) diff --git a/chainsource/chainsource_test.go b/chainsource/chainsource_test.go index e155edda2..cfe97d79d 100644 --- a/chainsource/chainsource_test.go +++ b/chainsource/chainsource_test.go @@ -3,6 +3,7 @@ package chainsource import ( "context" "errors" + "sync/atomic" "testing" "time" @@ -1717,3 +1718,190 @@ func TestConfActorIncludeBlock(t *testing.T) { }) } } + +// deadStreamBackend hands out block registrations whose epoch stream is already +// closed, modelling a backend that can never sustain a subscription (e.g. a +// chain notifier connection stuck after an LND restart). Every reconnect +// attempt therefore fails immediately. +type deadStreamBackend struct { + *mockBackend +} + +// RegisterBlocks returns a registration whose epoch channel is already closed. +func (b *deadStreamBackend) RegisterBlocks(ctx context.Context) ( + *BlockRegistration, error) { + + if ctx.Err() != nil { + return nil, ctx.Err() + } + + epochs := make(chan *BlockEpoch) + close(epochs) + + return &BlockRegistration{ + Epochs: epochs, + Cancel: func() {}, + }, nil +} + +// steppedClock advances by a fixed step on every read. Only the actor's monitor +// goroutine reads it, so no synchronization is needed; it lets the fatal +// reconnect timeout be exercised without real waits. +type steppedClock struct { + cur time.Time + step time.Duration +} + +// now advances the clock and returns the new time. +func (c *steppedClock) now() time.Time { + c.cur = c.cur.Add(c.step) + + return c.cur +} + +// TestBlockEpochActorEscalatesWhenBackendStaysDown verifies that a subscription +// which cannot be re-established within the fatal timeout stops retrying +// forever and escalates through OnFatal, so the daemon can restart with a +// fresh backend connection instead of spinning silently. +func TestBlockEpochActorEscalatesWhenBackendStaysDown(t *testing.T) { + t.Parallel() + + backend := &deadStreamBackend{mockBackend: newMockBackend()} + clock := &steppedClock{cur: time.Unix(0, 0), step: time.Minute} + + fatalCh := make(chan error, 1) + epochActor := NewBlockEpochActor(BlockEpochConfig{ + Backend: backend, + ReconnectBackoff: time.Millisecond, + MaxReconnectBackoff: time.Millisecond, + FatalReconnectTimeout: 90 * time.Second, + Now: clock.now, + OnFatal: fn.Some(func(err error) { + select { + case fatalCh <- err: + default: + } + }), + }) + defer epochActor.Stop() + + notifier := actor.NewChannelTellOnlyRef[BlockEpoch]( + "test-escalate-notify", 10, + ) + result := epochActor.Receive(t.Context(), &SubscribeBlocksRequest{ + CallerID: "test-epoch-escalate", + NotifyActor: fn.Some( + actor.TellOnlyRef[BlockEpoch](notifier), + ), + }) + require.True(t, result.IsOk()) + + select { + case err := <-fatalCh: + require.ErrorContains( + t, err, "block epoch subscription unrecoverable", + ) + + case <-time.After(5 * time.Second): + t.Fatal("expected fatal escalation when backend stays down") + } +} + +// countingDeadStreamBackend is a deadStreamBackend that counts reconnect +// attempts so a test can assert the monitor keeps retrying. +type countingDeadStreamBackend struct { + *deadStreamBackend + + registerCount atomic.Int64 +} + +// RegisterBlocks counts the attempt and delegates to the dead-stream backend. +func (b *countingDeadStreamBackend) RegisterBlocks(ctx context.Context) ( + *BlockRegistration, error) { + + b.registerCount.Add(1) + + return b.deadStreamBackend.RegisterBlocks(ctx) +} + +// TestBlockEpochActorRetriesWithoutOnFatal verifies that, with no OnFatal hook +// installed, a subscription that stays down past the fatal timeout keeps +// retrying instead of silently killing itself. Bounded escalation is opt-in via +// OnFatal; hookless callers keep the original unbounded-retry contract. +func TestBlockEpochActorRetriesWithoutOnFatal(t *testing.T) { + t.Parallel() + + backend := &countingDeadStreamBackend{ + deadStreamBackend: &deadStreamBackend{ + mockBackend: newMockBackend(), + }, + } + clock := &steppedClock{cur: time.Unix(0, 0), step: time.Minute} + + epochActor := NewBlockEpochActor(BlockEpochConfig{ + Backend: backend, + ReconnectBackoff: time.Millisecond, + MaxReconnectBackoff: time.Millisecond, + FatalReconnectTimeout: 90 * time.Second, + Now: clock.now, + // OnFatal deliberately unset: the actor must keep retrying. + }) + defer epochActor.Stop() + + notifier := actor.NewChannelTellOnlyRef[BlockEpoch]( + "test-noescalate-notify", 10, + ) + result := epochActor.Receive(t.Context(), &SubscribeBlocksRequest{ + CallerID: "test-epoch-noescalate", + NotifyActor: fn.Some( + actor.TellOnlyRef[BlockEpoch](notifier), + ), + }) + require.True(t, result.IsOk()) + + // The clock leaps a minute per read, so the fatal budget is exceeded + // almost immediately. Without OnFatal the monitor must keep + // reconnecting rather than exiting, so the attempt count climbs well + // past the point where an escalating actor would have stopped (~1 + // attempt). + require.Eventually(t, func() bool { + return backend.registerCount.Load() > 50 + }, 5*time.Second, 5*time.Millisecond, + "monitor should keep retrying when no OnFatal hook is set") +} + +// TestBlockEpochFatalTimeoutJitter verifies the effective fatal timeout: an +// explicit timeout with no jitter stays exact (so escalation tests are +// deterministic), while a configured jitter adds a bounded, rand-driven pad on +// top of the base. +func TestBlockEpochFatalTimeoutJitter(t *testing.T) { + t.Parallel() + + // An explicit timeout with no jitter configured is exact. + exact := NewBlockEpochActor(BlockEpochConfig{ + FatalReconnectTimeout: 90 * time.Second, + }) + require.Equal( + t, 90*time.Second, exact.fatalReconnectTimeout(), + ) + + // A configured jitter adds rand()*jitter to the base; the injected + // rand source makes it deterministic. + jittered := NewBlockEpochActor(BlockEpochConfig{ + FatalReconnectTimeout: time.Minute, + FatalReconnectJitter: 30 * time.Second, + Rand: func() float64 { return 0.5 }, + }) + require.Equal( + t, time.Minute+15*time.Second, jittered.fatalReconnectTimeout(), + ) + + // The default-timeout path opts into jitter automatically; with a zero + // rand fraction the effective budget is exactly the default. + def := NewBlockEpochActor(BlockEpochConfig{ + Rand: func() float64 { return 0 }, + }) + require.Equal( + t, defaultBlockEpochFatalTimeout, def.fatalReconnectTimeout(), + ) +} diff --git a/darepod/server.go b/darepod/server.go index c877aace0..3ac7b864c 100644 --- a/darepod/server.go +++ b/darepod/server.go @@ -209,6 +209,12 @@ type Server struct { // select on runCtx.Done(). runCtx context.Context //nolint:containedctx + // fatalCancel cancels runCtx with a cause when a subsystem reports an + // unrecoverable failure. run() returns that cause so the process exits + // non-zero and the orchestrator restarts it with fresh state. It is set + // once at the start of run() before any subsystem starts. + fatalCancel context.CancelCauseFunc + // walletState tracks the lifecycle state of the wallet // subsystem. In lnd mode this is always WalletStateReady // after successful lnd connection. In lwwallet mode it @@ -729,7 +735,15 @@ func (s *Server) RunWithContext(ctx context.Context) error { // subsystem so critical log events can trigger daemon shutdown. // //nolint:funlen -func (s *Server) run(ctx context.Context, shutdownFn func()) error { +func (s *Server) run(parentCtx context.Context, shutdownFn func()) error { + // Wrap the caller's context so a fatal subsystem failure can cancel the + // daemon with a cause. Normal signal-driven shutdown cancels parentCtx + // and run() returns nil; only a fatal cause is returned as an error, + // yielding a non-zero process exit for the orchestrator to restart. + ctx, cancelCause := context.WithCancelCause(parentCtx) + defer cancelCause(nil) + s.fatalCancel = cancelCause + // Store the run context so background goroutines (like the // btcwallet sync poller) can outlive individual RPC // handlers but still shut down with the daemon. @@ -1201,9 +1215,31 @@ func (s *Server) run(ctx context.Context, shutdownFn func()) error { s.log.InfoS(ctx, "Shutting down darepod") + // A non-cancel cause means a subsystem escalated an unrecoverable + // failure (e.g. a block epoch subscription that could not be + // re-established). Surface it so the process exits non-zero and is + // restarted with fresh state rather than running on as a zombie. + if cause := context.Cause(ctx); cause != nil && + !errors.Is(cause, context.Canceled) { + + s.log.ErrorS(ctx, "darepod exiting on fatal subsystem error", + cause) + + return cause + } + return nil } +// signalFatal cancels the run context with an unrecoverable-failure cause so +// run() returns it and the process exits non-zero. It is safe to call from any +// subsystem goroutine and is a no-op before run() has wired fatalCancel. +func (s *Server) signalFatal(err error) { + if s.fatalCancel != nil { + s.fatalCancel(err) + } +} + // startWalletReadyServices starts the services that need wallet-derived keys // and chain access. func (s *Server) startWalletReadyServices(ctx context.Context, @@ -1813,6 +1849,7 @@ func (s *Server) registerChainSourceActor( chainsource.ChainSourceConfig{ Backend: s.chainBackend, System: s.actorSystem, + OnFatal: fn.Some(s.signalFatal), }, ) From 54f707582a6d2f42590604fe783d75c79af0c500 Mon Sep 17 00:00:00 2001 From: Olaoluwa Osuntokun Date: Mon, 22 Jun 2026 15:56:07 -0700 Subject: [PATCH 2/2] darepod: expose /v1/health and /v1/ready on the HTTP gateway The block epoch fail-fast makes the daemon exit on the wedge we know about, and k8s restarts an exited container. But the chart only probes the RPC port with tcpSocket, which stays green for a process that is listening yet making no progress. This adds the progress-aware backstop, mirroring tapd and nautilus. In this commit, we mount unauthenticated /v1/health and /v1/ready routes directly on the existing grpc-gateway mux via HandlePath. Both answer from in-process state only and never touch the chain backend, so they keep responding even when it is stuck, and a probe cannot be used to amplify load onto it. Liveness (/v1/health) fails when a subsystem has escalated a fatal failure, or when the daemon stalls in startup past a generous deadline; a failure restarts the pod. The wedged-backend class that motivated this is already surfaced by the block epoch fatal escalation, which latches that flag, so liveness needs no separate backend probe. Readiness (/v1/ready) fails until the wallet subsystem finishes starting, draining the pod from the Service endpoints without restarting it. The chart change that points the probes at these routes lives in lightning-infra. --- darepod/gateway_server.go | 91 +++++++++++++++++++++++++++++++++ darepod/gateway_server_test.go | 93 ++++++++++++++++++++++++++++++++++ darepod/server.go | 75 ++++++++++++++++++++++++++- 3 files changed, 258 insertions(+), 1 deletion(-) create mode 100644 darepod/gateway_server_test.go diff --git a/darepod/gateway_server.go b/darepod/gateway_server.go index e846c761d..48cb29361 100644 --- a/darepod/gateway_server.go +++ b/darepod/gateway_server.go @@ -28,6 +28,11 @@ const ( defaultGatewayIdleTimeout = 60 * time.Second ) +// errHealthUnavailable flags that the daemon health checks were not available +// when the gateway started, so the liveness/readiness probe routes could not be +// mounted. +var errHealthUnavailable = errors.New("daemon health check unavailable") + // gatewayServer serves HTTP/JSON requests through generated grpc-gateway // handlers. type gatewayServer struct { @@ -38,6 +43,16 @@ type gatewayServer struct { registrars []RPCGatewayRegistrar log btclog.Logger + // liveness and readiness answer the /v1/health and /v1/ready probe + // routes. When nil they default to the daemon's LivenessCheck and + // ReadinessCheck. Liveness reports process viability (a failure should + // restart the pod); readiness reports serve-readiness (a failure only + // drains the pod). Both perform in-process checks only and never touch + // the chain backend, so an unauthenticated probe cannot amplify load + // onto it. + liveness func(context.Context) error + readiness func(context.Context) error + listener net.Listener httpSrv *http.Server wg sync.WaitGroup @@ -105,6 +120,50 @@ func (g *gatewayServer) Start(ctx context.Context) error { } } + // Mount unauthenticated liveness/readiness routes directly on the mux + // so a k8s probe can detect a wedged-but-listening daemon. Both perform + // only in-process checks (no backend round-trip), so they keep + // answering when the chain backend is stuck and cannot be abused to + // amplify load onto it. Liveness failure restarts the pod; readiness + // failure only drains it. + liveness, readiness := g.liveness, g.readiness + if g.rpcServer != nil && g.rpcServer.server != nil { + if liveness == nil { + liveness = g.rpcServer.server.LivenessCheck + } + if readiness == nil { + readiness = g.rpcServer.server.ReadinessCheck + } + } + + if liveness == nil || readiness == nil { + // Make the absence visible: an operator wiring k8s probes to a + // daemon whose health checks are unavailable would otherwise + // see only silent 404s and assume the routes work. + g.log.WarnS(ctx, "Health probe routes not registered", + errHealthUnavailable, + ) + } else { + if err := mux.HandlePath( + http.MethodGet, "/v1/health", healthHandler(liveness), + ); err != nil { + + cancelRegister() + _ = listener.Close() + + return fmt.Errorf("register health route: %w", err) + } + if err := mux.HandlePath( + http.MethodGet, "/v1/ready", healthHandler(readiness), + ); err != nil { + + cancelRegister() + _ = listener.Close() + + return fmt.Errorf("register ready route: %w", err) + } + } + g.listener = listener g.cancel = cancelRegister g.httpSrv = &http.Server{ @@ -155,6 +214,38 @@ func (g *gatewayServer) Stop(_ context.Context) error { return err } +// healthHandler builds an HTTP handler that answers a liveness or readiness +// probe from the given check: 200 with `{"status":"ok"}` when healthy, 503 with +// the reason otherwise, so a wedged-but-listening daemon fails its probe. The +// check reports only in-process state, so the reason string carries no chain +// backend transport detail (no information disclosure to the unauthenticated +// caller). +func healthHandler(health func(context.Context) error) runtime.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request, + _ map[string]string) { + + w.Header().Set("Content-Type", "application/json") + + if err := health(r.Context()); err != nil { + w.WriteHeader(http.StatusServiceUnavailable) + + // The reason is a controlled in-process status string + // (no backend detail), and the response is JSON with + // %q-escaping, so reflecting it is safe. + //nolint:gosec // G705: reason controlled, non-HTML. + _, _ = fmt.Fprintf( + w, `{"status":"unavailable","reason":%q}`+"\n", + err.Error(), + ) + + return + } + + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte(`{"status":"ok"}` + "\n")) + } +} + // Addr returns the address the gateway is listening on. func (g *gatewayServer) Addr() net.Addr { if g == nil || g.listener == nil { diff --git a/darepod/gateway_server_test.go b/darepod/gateway_server_test.go new file mode 100644 index 000000000..4ddfa8a6d --- /dev/null +++ b/darepod/gateway_server_test.go @@ -0,0 +1,93 @@ +package darepod + +import ( + "context" + "errors" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +// TestGatewayHealthHandler verifies the liveness route returns 200 when the +// health check passes and 503 with the reason when it fails, so a k8s probe can +// restart a wedged-but-listening daemon. +func TestGatewayHealthHandler(t *testing.T) { + t.Parallel() + + var healthErr error + h := healthHandler(func(context.Context) error { return healthErr }) + + rec := httptest.NewRecorder() + h(rec, httptest.NewRequest(http.MethodGet, "/v1/health", nil), nil) + require.Equal(t, http.StatusOK, rec.Code) + require.Contains(t, rec.Body.String(), `"ok"`) + + healthErr = errors.New("chain backend unreachable: dial timeout") + rec = httptest.NewRecorder() + h(rec, httptest.NewRequest(http.MethodGet, "/v1/health", nil), nil) + require.Equal(t, http.StatusServiceUnavailable, rec.Code) + require.Contains(t, rec.Body.String(), "unreachable") +} + +// TestServerLivenessReadiness verifies the split health checks: while the +// wallet is still starting (within the startup deadline) liveness is healthy +// but readiness reports not-ready; once the wallet is ready both pass; and a +// fatal escalation latches both unhealthy. Neither check touches the chain +// backend. +func TestServerLivenessReadiness(t *testing.T) { + t.Parallel() + + s := &Server{startedAt: time.Now()} + + // Still starting: liveness healthy (within the deadline), readiness + // not. + require.NoError(t, s.LivenessCheck(context.Background())) + require.ErrorContains( + t, + s.ReadinessCheck( + context.Background(), + ), + "not ready", + ) + + // Once the wallet subsystem is ready, readiness passes too. + s.walletState.Store(int32(WalletStateReady)) + require.NoError(t, s.ReadinessCheck(context.Background())) + require.NoError(t, s.LivenessCheck(context.Background())) + + // A fatal escalation latches both unhealthy. + s.signalFatal(errors.New("boom")) + require.ErrorContains( + t, + s.LivenessCheck( + context.Background(), + ), + "fatal", + ) + require.ErrorContains( + t, + s.ReadinessCheck( + context.Background(), + ), + "fatal", + ) +} + +// TestServerLivenessStartupDeadline verifies liveness fails once the daemon has +// sat in startup past the deadline without the wallet becoming ready, so a +// daemon wedged before init is restarted rather than reported live forever. +func TestServerLivenessStartupDeadline(t *testing.T) { + t.Parallel() + + s := &Server{startedAt: time.Now().Add(-2 * healthStartupDeadline)} + require.ErrorContains( + t, + s.LivenessCheck( + context.Background(), + ), + "not ready", + ) +} diff --git a/darepod/server.go b/darepod/server.go index 3ac7b864c..a1e174dd7 100644 --- a/darepod/server.go +++ b/darepod/server.go @@ -215,6 +215,16 @@ type Server struct { // once at the start of run() before any subsystem starts. fatalCancel context.CancelCauseFunc + // fatalFlag latches once a subsystem escalates an unrecoverable + // failure, so the health check reports unhealthy even in the brief + // window before the process exits. + fatalFlag atomic.Bool + + // startedAt records when run() began so the liveness check can restart + // a daemon that never finishes starting. It is set once at the top of + // run(), before the gateway goroutine that reads it is spawned. + startedAt time.Time + // walletState tracks the lifecycle state of the wallet // subsystem. In lnd mode this is always WalletStateReady // after successful lnd connection. In lwwallet mode it @@ -749,6 +759,11 @@ func (s *Server) run(parentCtx context.Context, shutdownFn func()) error { // handlers but still shut down with the daemon. s.runCtx = ctx + // Stamp the start so the liveness check can bound how long the daemon + // may sit in startup. Set here, before any subsystem (and the gateway + // goroutine that reads it) starts. + s.startedAt = time.Now() + // ------------------------------------------------------- // 0. Initialize the logging backend and subsystem loggers. // ------------------------------------------------------- @@ -1223,7 +1238,8 @@ func (s *Server) run(parentCtx context.Context, shutdownFn func()) error { !errors.Is(cause, context.Canceled) { s.log.ErrorS(ctx, "darepod exiting on fatal subsystem error", - cause) + cause, + ) return cause } @@ -1235,11 +1251,68 @@ func (s *Server) run(parentCtx context.Context, shutdownFn func()) error { // run() returns it and the process exits non-zero. It is safe to call from any // subsystem goroutine and is a no-op before run() has wired fatalCancel. func (s *Server) signalFatal(err error) { + s.fatalFlag.Store(true) if s.fatalCancel != nil { s.fatalCancel(err) } } +// errFatalSubsystem is the liveness/readiness error returned once a subsystem +// has escalated an unrecoverable failure via signalFatal. +var errFatalSubsystem = fmt.Errorf("daemon escalated a fatal subsystem failure") + +// healthStartupDeadline bounds how long the daemon may sit in startup before +// the liveness probe fails. A daemon that never finishes starting (wedged +// before the wallet and chain backend come up) is as dead as one that latched a +// fatal failure; this lets the orchestrator restart it. It is generous so a +// slow first chain sync is not mistaken for a wedge. +const healthStartupDeadline = 15 * time.Minute + +// LivenessCheck reports whether the daemon process is viable and should keep +// running. It fails only when a subsystem has latched a fatal failure or the +// daemon has stalled in startup past healthStartupDeadline. It performs no +// backend I/O, so an unauthenticated /v1/health probe cannot drive load onto +// the chain backend; a wedged backend is surfaced through the block-epoch fatal +// escalation, which latches the fatal flag. A failed liveness probe should +// restart the pod. +func (s *Server) LivenessCheck(_ context.Context) error { + if s.fatalFlag.Load() { + return errFatalSubsystem + } + + // A daemon that never finishes starting is wedged just as surely as one + // that latched a fatal failure. Bound the startup grace so liveness can + // restart a daemon stuck before the wallet and chain backend ever came + // up — a state the readiness gate alone would never recover from. + if s.WalletLifecycleState() != WalletStateReady { + if since := time.Since( + s.startedAt, + ); since > healthStartupDeadline { + return fmt.Errorf("daemon not ready %s after start", + since.Round(time.Second)) + } + } + + return nil +} + +// ReadinessCheck reports whether the daemon is ready to serve requests. It +// fails when a fatal failure has latched or the wallet subsystem has not +// finished starting, so traffic is held off a still-starting or wedged daemon. +// Like LivenessCheck it performs no backend I/O. A failed readiness probe only +// removes the pod from the Service endpoints; it does not restart it. +func (s *Server) ReadinessCheck(_ context.Context) error { + if s.fatalFlag.Load() { + return errFatalSubsystem + } + + if state := s.WalletLifecycleState(); state != WalletStateReady { + return fmt.Errorf("wallet subsystem not ready: %s", state) + } + + return nil +} + // startWalletReadyServices starts the services that need wallet-derived keys // and chain access. func (s *Server) startWalletReadyServices(ctx context.Context,