From fd17edb15258bad96709bbd0edb8eb58043fdca2 Mon Sep 17 00:00:00 2001 From: Lukasz Klimek <842586+lklimek@users.noreply.github.com> Date: Thu, 24 Sep 2026 14:34:23 +0000 Subject: [PATCH 1/3] fix(consensus): wait for receive routine during shutdown MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Join the receive routine before State.Wait returns and cancel its context on explicit Stop. This prevents node cleanup racing in-flight consensus file writes. Add a regression test for the shutdown contract. Co-Authored-By: OpenAI Codex GPT-6 🤖 Co-authored by [Claudius the Magnificent](https://github.com/lklimek/claudius) AI Agent --- internal/consensus/state.go | 21 ++++++++++++++++-- internal/consensus/state_test.go | 37 ++++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/internal/consensus/state.go b/internal/consensus/state.go index b1f28890d5..ffdadb8d9b 100644 --- a/internal/consensus/state.go +++ b/internal/consensus/state.go @@ -192,7 +192,9 @@ type State struct { proposedAppVersion uint64 // wait the channel event happening for shutting down the state gracefully - onStopCh chan *cstypes.RoundState + onStopCh chan *cstypes.RoundState + receiveWG sync.WaitGroup + receiveCancel context.CancelFunc msgInfoQueue *msgInfoQueue msgDispatcher *msgInfoDispatcher @@ -545,7 +547,13 @@ func (cs *State) OnStart(ctx context.Context) error { } // now start the receiveRoutine - go cs.receiveRoutine(ctx, cs.stopFn) + receiveCtx, receiveCancel := context.WithCancel(ctx) + cs.receiveCancel = receiveCancel + cs.receiveWG.Add(1) + go func() { + defer cs.receiveWG.Done() + cs.receiveRoutine(receiveCtx, cs.stopFn) + }() // schedule the first round! // use GetRoundState so we don't race the receiveRoutine for access @@ -586,12 +594,21 @@ func (cs *State) OnStop() { } } + if cs.receiveCancel != nil { + cs.receiveCancel() + } if cs.timeoutTicker.IsRunning() { cs.timeoutTicker.Stop() } // WAL is stopped in receiveRoutine. } +// Wait waits for the service and its consensus receive routine to finish. +func (cs *State) Wait() { + cs.BaseService.Wait() + cs.receiveWG.Wait() +} + // OpenWAL opens a file to log all consensus messages and timeouts for // deterministic accountability. func (cs *State) OpenWAL(ctx context.Context, walFile string) (WAL, error) { diff --git a/internal/consensus/state_test.go b/internal/consensus/state_test.go index b4ec0b7435..e2c9c97b50 100644 --- a/internal/consensus/state_test.go +++ b/internal/consensus/state_test.go @@ -3496,3 +3496,40 @@ func mockProposerApplicationCalls(t *testing.T, m *abcimocks.Application, round Once() } } + +func TestStateWaitDrainsReceiveRoutine(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cs, _ := makeState(ctx, t, makeStateArgs{config: configSetup(t)}) + entered, release := make(chan struct{}), make(chan struct{}) + notified := false + cs.stopFn = func(*State) bool { + if !notified { + close(entered) + notified = true + } + <-release + return false + } + require.NoError(t, cs.Start(ctx)) + <-entered + cs.Stop() + cs.BaseService.Wait() + + waited := make(chan struct{}) + go func() { + cs.Wait() + close(waited) + }() + select { + case <-waited: + t.Error("Wait returned before the consensus receive routine finished") + case <-time.After(50 * time.Millisecond): + } + close(release) + select { + case <-waited: + case <-time.After(time.Second): + t.Fatal("Wait did not return after the receive routine finished") + } +} From 8e431fc19816ee06e372f202c28a2c7c166908ba Mon Sep 17 00:00:00 2001 From: Lukasz Klimek <842586+lklimek@users.noreply.github.com> Date: Fri, 25 Sep 2026 09:24:19 +0000 Subject: [PATCH 2/3] fix(service)!: share worker lifecycle for consensus shutdown Port the focused lifecycle core from #1515, replacing the local State wait workaround while retaining other service ownership mechanisms. Co-Authored-By: OpenAI GPT-6 --- docs/architecture/service-lifecycle.md | 46 ++++ internal/blocksync/synchronizer.go | 10 +- internal/consensus/common_test.go | 1 + internal/consensus/reactor.go | 40 ++-- internal/consensus/reactor_test.go | 14 ++ internal/consensus/replay_test.go | 4 - internal/consensus/state.go | 88 +++----- internal/consensus/state_test.go | 16 +- internal/consensus/ticker.go | 18 +- internal/consensus/ticker_test.go | 28 +++ internal/consensus/wal.go | 62 +----- internal/consensus/wal_test.go | 39 +++- internal/libs/autofile/autofile.go | 34 ++- internal/libs/autofile/autofile_test.go | 16 ++ internal/libs/autofile/group.go | 62 +----- internal/p2p/conn/connection.go | 10 + internal/p2p/conn/lifecycle_test.go | 40 ++++ libs/service/lifecycle_test.go | 268 +++++++++++++++++++++++ libs/service/service.go | 274 ++++++++++++------------ node/node.go | 8 +- node/node_test.go | 2 - 21 files changed, 724 insertions(+), 356 deletions(-) create mode 100644 docs/architecture/service-lifecycle.md create mode 100644 internal/consensus/ticker_test.go create mode 100644 internal/p2p/conn/lifecycle_test.go create mode 100644 libs/service/lifecycle_test.go diff --git a/docs/architecture/service-lifecycle.md b/docs/architecture/service-lifecycle.md new file mode 100644 index 0000000000..5c4132117f --- /dev/null +++ b/docs/architecture/service-lifecycle.md @@ -0,0 +1,46 @@ +# Service lifecycle core + +`BaseService.Start` gives `OnStart` an attempt-scoped child context. Parent +cancellation and manual `Stop` both cancel it. `Go(ctx, fn)` registers workers +using that context or a derived context; registration is synchronized with +shutdown and rejected for a different attempt or after shutdown begins. + +Shutdown closes admission, marks `IsRunning` false, cancels the context and +calls `OnStop`. It then joins registered workers, calls optional `OnDrain`, and +releases `Wait`. `Stopping()` signals cancellation separately from completion. +`Stop` invokes `OnStop` synchronously but does not join registered workers. +Concurrent or recursive calls to `Stop` return immediately. + +Hooks run outside the lifecycle mutex. `OnStop` must unblock workers without +waiting for its own registered work. Workers may call `Stop`; neither hooks nor +workers may call their own service's `Wait`. Release resources shared with +workers in `OnDrain`, or after `Stop` followed by `Wait`. + +`IsRunning` is false during startup and shutdown. `Wait` joins the current +attempt, including startup and cleanup; it returns immediately before startup. +Failed `OnStart` cancels and joins its workers before retry becomes possible. +The implementation must roll back acquired resources: failed attempts do not +invoke `OnStop` or `OnDrain`. Successfully stopped services cannot restart. +Composite retries also require recreating children already successfully stopped. + +## First migration stage + +Consensus State registers its receive loop and joins its queue fan-in and WAL +before returning. The timeout ticker, WAL and autofile group use shared worker +tracking. AutoFile explicitly joins its timer/signal worker. WAL ownership lasts +until the receive loop's final write. Consensus handoff admission is also joined, +so a late blocksync handoff cannot start State after reactor shutdown. + +Node waits for reactors before closing event sinks and stores. The blocked +receive-loop regression and original node teardown tests verify this path +without using leaktest as an extra production-shutdown barrier. + +This is a partial migration. Plain goroutines are not tracked automatically. +RPC/WebSocket handlers, mempool rechecks, reactor gossip and other service +workers retain their existing ownership mechanisms. This change does not claim +that Node.Wait joins every helper or introduce the broader durable-finalization +and dependency-lifetime changes from #1515. + +Two compatibility bridges preserve existing parent-context lifetimes: blocksync +application during handoff and connection error callbacks after I/O shutdown. +Their local cancellation and join mechanisms remain in place in this stage. diff --git a/internal/blocksync/synchronizer.go b/internal/blocksync/synchronizer.go index 6d5317f10a..68429fc50e 100644 --- a/internal/blocksync/synchronizer.go +++ b/internal/blocksync/synchronizer.go @@ -236,6 +236,13 @@ func (s *Synchronizer) setStartHeight(height int64) { s.jobGen.mtx.Unlock() } +type applicationContextKey struct{} + +// Start preserves the caller's application lifetime across a block-sync handover. +func (s *Synchronizer) Start(ctx context.Context) error { + return s.BaseService.Start(context.WithValue(ctx, applicationContextKey{}, ctx)) +} + // OnStart implements service.Service by spawning requesters routine and recording // synchronizer's start time. func (s *Synchronizer) OnStart(ctx context.Context) error { @@ -245,6 +252,7 @@ func (s *Synchronizer) OnStart(ctx context.Context) error { s.lastAdvance = s.clock.Now() s.lastMonitorUpdate = s.lastAdvance s.ctx, s.cancel = context.WithCancel(ctx) + applicationCtx := ctx.Value(applicationContextKey{}).(context.Context) s.consumerDone = make(chan struct{}) s.workerPool.Run(s.ctx) go s.runHandler(s.ctx, s.produceJob) @@ -252,7 +260,7 @@ func (s *Synchronizer) OnStart(ctx context.Context) error { defer close(s.consumerDone) s.runHandler(s.ctx, func(handlerCtx context.Context) error { // Handover cancels consumer I/O; only node shutdown may cancel application. - return s.consumeJobResult(handlerCtx, ctx) + return s.consumeJobResult(handlerCtx, applicationCtx) }) }() return nil diff --git a/internal/consensus/common_test.go b/internal/consensus/common_test.go index d9df4e175e..af2b1a1c74 100644 --- a/internal/consensus/common_test.go +++ b/internal/consensus/common_test.go @@ -1192,6 +1192,7 @@ type mockTicker struct { } func (m *mockTicker) Start(context.Context) error { return nil } +func (m *mockTicker) Wait() {} func (m *mockTicker) Stop() {} func (m *mockTicker) IsRunning() bool { return false } diff --git a/internal/consensus/reactor.go b/internal/consensus/reactor.go index 8b6bde1a9f..e84301fb31 100644 --- a/internal/consensus/reactor.go +++ b/internal/consensus/reactor.go @@ -413,9 +413,7 @@ func (r *Reactor) OnStart(ctx context.Context) error { return nil } -// OnStop stops the reactor by signaling to all spawned goroutines to exit and -// blocking until they all exit, as well as unsubscribing from events and stopping -// state. +// OnStop requests reactor and consensus shutdown. func (r *Reactor) OnStop() { // Cancel the reactor context to signal all goroutines to stop if r.cancel != nil { @@ -423,10 +421,12 @@ func (r *Reactor) OnStop() { } r.state.Stop() +} - if !r.WaitSync() { - r.state.Wait() - } +// OnDrain joins consensus after any admitted handoff has finished. +func (r *Reactor) OnDrain() { + r.state.Stop() + r.state.Wait() } // WaitSync returns whether the consensus reactor is waiting for state/block sync. @@ -447,6 +447,20 @@ func (r *Reactor) WaitSync() bool { // targetHeight is the highest committed block height reported during block sync. // skipWAL says the node needs no WAL catchup. func (r *Reactor) SwitchToConsensus(ctx context.Context, state sm.State, skipWAL bool, targetHeight int64) { + if r.ctx == nil || ctx.Err() != nil { + return + } + done := make(chan struct{}) + if !r.Go(r.ctx, func(ctx context.Context) { + defer close(done) + r.switchToConsensus(ctx, state, skipWAL, targetHeight) + }) { + return + } + <-done +} + +func (r *Reactor) switchToConsensus(ctx context.Context, state sm.State, skipWAL bool, targetHeight int64) { r.logger.Info("switching to consensus", "target_height", targetHeight) if targetHeight > state.LastBlockHeight { @@ -473,6 +487,9 @@ func (r *Reactor) SwitchToConsensus(ctx context.Context, state sm.State, skipWAL r.state.eventPublisher.PublishNewRoundStepEvent(stateData.RoundState) if err := r.state.Start(ctx); err != nil { + if ctx.Err() != nil { + return + } panic(fmt.Sprintf(`failed to start consensus state: %v conS: @@ -520,8 +537,6 @@ func (r *Reactor) GetPeerState(peerID types.NodeID) (*PeerState, bool) { // internal pubsub defined in the consensus state to broadcast them to peers // upon receiving. func (r *Reactor) subscribeToBroadcastEvents(ctx context.Context, stateCh p2p.Channel) { - onStopCh := r.state.getOnStopCh() - r.state.emitter.AddListener( types.EventNewRoundStepValue, func(data eventemitter.EventData) error { @@ -531,14 +546,7 @@ func (r *Reactor) subscribeToBroadcastEvents(ctx context.Context, stateCh p2p.Ch return err } r.logResult(err, r.logger, "broadcasting round step message", "height", rs.Height, "round", rs.Round) - select { - case onStopCh <- data.(*cstypes.RoundState): - return nil - case <-ctx.Done(): - return ctx.Err() - default: - return nil - } + return nil }, ) diff --git a/internal/consensus/reactor_test.go b/internal/consensus/reactor_test.go index 9787e40d1e..c2e4e5bd55 100644 --- a/internal/consensus/reactor_test.go +++ b/internal/consensus/reactor_test.go @@ -834,3 +834,17 @@ func TestReactorPeerDownDeletesPeerStateSynchronously(t *testing.T) { "peerDown must delete the peer state before returning, so a fast reconnect cannot reuse it") require.False(t, ps.IsRunning(), "peerDown must stop the peer state before returning") } + +func TestReactorRejectsConsensusHandoffAfterShutdown(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cs, _ := makeState(ctx, t, makeStateArgs{}) + rts := setup(ctx, t, 1, []*State{cs}, 32) + for _, r := range rts.reactors { + r.Stop() + r.Wait() + r.SwitchToConsensus(ctx, cs.GetStateData().state, false, 0) + require.False(t, cs.IsRunning(), "handoff started consensus after reactor shutdown") + require.True(t, r.WaitSync(), "rejected handoff changed sync state") + } +} diff --git a/internal/consensus/replay_test.go b/internal/consensus/replay_test.go index d2ce9c1332..bb08f1b80e 100644 --- a/internal/consensus/replay_test.go +++ b/internal/consensus/replay_test.go @@ -1521,10 +1521,6 @@ func testWALRoundsSkipper(t *testing.T, slowProposer bool) { <-replayStopped cs.Stop() cs.Wait() - // The stop predicate bypasses receiveRoutine's WAL and queue shutdown. - cs.wal.Stop() - cs.wal.Wait() - cs.msgInfoQueue.stop() }() newBlockSub, err := cs.eventBus.SubscribeWithArgs(ctx, pubsub.SubscribeArgs{ diff --git a/internal/consensus/state.go b/internal/consensus/state.go index ffdadb8d9b..9075b67336 100644 --- a/internal/consensus/state.go +++ b/internal/consensus/state.go @@ -191,11 +191,6 @@ type State struct { // proposer's latest available app protocol version that goes to block header proposedAppVersion uint64 - // wait the channel event happening for shutting down the state gracefully - onStopCh chan *cstypes.RoundState - receiveWG sync.WaitGroup - receiveCancel context.CancelFunc - msgInfoQueue *msgInfoQueue msgDispatcher *msgInfoDispatcher blockExecutor *blockExecutor @@ -287,7 +282,6 @@ func NewState( evpool: evpool, emitter: eventemitter.New(eventemitter.WithLogger(logger)), metrics: NopMetrics(), - onStopCh: make(chan *cstypes.RoundState), verificationBudget: newVerificationBudget( cfg.VerificationRateLimit, ), @@ -473,7 +467,20 @@ func (cs *State) SetPrivValidator(ctx context.Context, priv types.PrivValidator) // OnStart loads the latest state via the WAL, and starts the timeout and // receive routines. -func (cs *State) OnStart(ctx context.Context) error { +func (cs *State) OnStart(ctx context.Context) (err error) { + defer func() { + if err != nil { + cs.wal.Stop() + cs.wal.Wait() + cs.wal = nilWAL{} + cs.timeoutTicker.Stop() + cs.timeoutTicker.Wait() + if _, ok := cs.timeoutTicker.(*timeoutTicker); ok { + cs.timeoutTicker = NewTimeoutTicker(cs.logger) + cs.roundScheduler.timeoutTicker = cs.timeoutTicker + } + } + }() if err := cs.updateStateFromStore(); err != nil { return err } @@ -520,6 +527,7 @@ func (cs *State) OnStart(ctx context.Context) error { // 1) prep work cs.wal.Stop() + cs.wal.Wait() repairAttempted = true @@ -547,13 +555,9 @@ func (cs *State) OnStart(ctx context.Context) error { } // now start the receiveRoutine - receiveCtx, receiveCancel := context.WithCancel(ctx) - cs.receiveCancel = receiveCancel - cs.receiveWG.Add(1) - go func() { - defer cs.receiveWG.Done() - cs.receiveRoutine(receiveCtx, cs.stopFn) - }() + if !cs.Go(ctx, func(ctx context.Context) { cs.receiveRoutine(ctx, cs.stopFn) }) { + return context.Canceled + } // schedule the first round! // use GetRoundState so we don't race the receiveRoutine for access @@ -564,7 +568,8 @@ func (cs *State) OnStart(ctx context.Context) error { // loadWalFile loads WAL data from file. It overwrites cs.wal. func (cs *State) loadWalFile(ctx context.Context) error { - wal, err := cs.OpenWAL(ctx, cs.config.WalFile()) + // receiveRoutine owns the WAL until its final write and explicit Stop/Wait. + wal, err := cs.OpenWAL(context.WithoutCancel(ctx), cs.config.WalFile()) if err != nil { cs.logger.Error("failed to load state WAL", "err", err) return err @@ -574,40 +579,11 @@ func (cs *State) loadWalFile(ctx context.Context) error { return nil } -func (cs *State) getOnStopCh() chan *cstypes.RoundState { - cs.mtx.RLock() - defer cs.mtx.RUnlock() +// OnStop requests timeout worker shutdown. Wait also joins receiveRoutine. +func (cs *State) OnStop() { cs.timeoutTicker.Stop() } - return cs.onStopCh -} - -// OnStop implements service.Service. -func (cs *State) OnStop() { - stateData := cs.stateDataStore.Get() - // If the node is committing a new block, wait until it is finished! - if cs.GetRoundState().Step == cstypes.RoundStepApplyCommit { - select { - case <-cs.getOnStopCh(): - case <-time.After(stateData.state.ConsensusParams.Timeout.Vote): - // we wait vote timeout, just in case - cs.logger.Error("OnStop: timeout waiting for commit to finish", "time", stateData.state.ConsensusParams.Timeout.Vote) - } - } - - if cs.receiveCancel != nil { - cs.receiveCancel() - } - if cs.timeoutTicker.IsRunning() { - cs.timeoutTicker.Stop() - } - // WAL is stopped in receiveRoutine. -} - -// Wait waits for the service and its consensus receive routine to finish. -func (cs *State) Wait() { - cs.BaseService.Wait() - cs.receiveWG.Wait() -} +// OnDrain joins the timeout worker before callers release consensus stores. +func (cs *State) OnDrain() { cs.timeoutTicker.Wait() } // OpenWAL opens a file to log all consensus messages and timeouts for // deterministic accountability. @@ -620,6 +596,7 @@ func (cs *State) OpenWAL(ctx context.Context, walFile string) (WAL, error) { if err := wal.Start(ctx); err != nil { cs.logger.Error("failed to start WAL", "err", err) + wal.Group().Close() return nil, err } @@ -717,12 +694,6 @@ func (cs *State) receiveRoutine(ctx context.Context, stopFn func(*State) bool) { if r := recover(); r != nil { cs.logger.Error("CONSENSUS FAILURE!!!", "err", r, "stack", string(debug.Stack())) - // Make a best-effort attempt to close the WAL, but otherwise do not - // attempt to gracefully terminate. Once consensus has irrecoverably - // failed, any additional progress we permit the node to make may - // complicate diagnosing and recovering from the failure. - onExit(cs) - // There are a couple of cases where the we // panic with an error from deeper within the // state machine and in these cases, typically @@ -756,7 +727,12 @@ func (cs *State) receiveRoutine(ctx context.Context, stopFn func(*State) bool) { } }() - go cs.msgInfoQueue.fanIn(ctx) + queueDone := make(chan struct{}) + go func() { + defer close(queueDone) + cs.msgInfoQueue.fanIn(ctx) + }() + defer func() { onExit(cs); <-queueDone }() for { if stopFn != nil && stopFn(cs) { @@ -786,7 +762,6 @@ func (cs *State) receiveRoutine(ctx context.Context, stopFn func(*State) bool) { if err != nil { // Leaving without this would strand the queue's reader // goroutines: nothing else ever tells them to stop. - onExit(cs) return } err = stateData.Save() @@ -807,7 +782,6 @@ func (cs *State) receiveRoutine(ctx context.Context, stopFn func(*State) bool) { cs.logger.Error("failed update state-data", "err", err) } case <-ctx.Done(): - onExit(cs) return } diff --git a/internal/consensus/state_test.go b/internal/consensus/state_test.go index e2c9c97b50..6b76104a85 100644 --- a/internal/consensus/state_test.go +++ b/internal/consensus/state_test.go @@ -3496,7 +3496,6 @@ func mockProposerApplicationCalls(t *testing.T, m *abcimocks.Application, round Once() } } - func TestStateWaitDrainsReceiveRoutine(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() @@ -3514,7 +3513,6 @@ func TestStateWaitDrainsReceiveRoutine(t *testing.T) { require.NoError(t, cs.Start(ctx)) <-entered cs.Stop() - cs.BaseService.Wait() waited := make(chan struct{}) go func() { @@ -3533,3 +3531,17 @@ func TestStateWaitDrainsReceiveRoutine(t *testing.T) { t.Fatal("Wait did not return after the receive routine finished") } } + +func TestDirectReceiveRoutineDrainsQueue(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cs, _ := makeState(ctx, t, makeStateArgs{}) + cs.receiveRoutine(ctx, func(*State) bool { return true }) + select { + case _, ok := <-cs.msgInfoQueue.read(): + require.False(t, ok) + default: + t.Fatal("receiveRoutine returned before queue was closed") + } + require.NoError(t, ctx.Err()) +} diff --git a/internal/consensus/ticker.go b/internal/consensus/ticker.go index 30b94d3adb..95c42f83fe 100644 --- a/internal/consensus/ticker.go +++ b/internal/consensus/ticker.go @@ -18,6 +18,7 @@ var ( type TimeoutTicker interface { Start(context.Context) error Stop() + Wait() IsRunning() bool Chan() <-chan timeoutInfo // on which to receive a timeout ScheduleTimeout(ti timeoutInfo) // reset the timer @@ -52,13 +53,16 @@ func NewTimeoutTicker(logger log.Logger) TimeoutTicker { // OnStart implements service.Service. It starts the timeout routine. func (t *timeoutTicker) OnStart(ctx context.Context) error { - go t.timeoutRoutine(ctx) + t.Go(ctx, t.timeoutRoutine) return nil } // OnStop implements service.Service. It stops the timeout routine. -func (t *timeoutTicker) OnStop() { t.stopTimer() } +func (t *timeoutTicker) OnStop() {} + +// OnDrain stops the timer after its sole scheduling worker exits. +func (t *timeoutTicker) OnDrain() { t.stopTimer() } // Chan returns a channel on which timeouts are sent. func (t *timeoutTicker) Chan() <-chan timeoutInfo { @@ -69,7 +73,10 @@ func (t *timeoutTicker) Chan() <-chan timeoutInfo { // The timeoutRoutine is always available to read from tickChan, so this won't block. // The scheduling may fail if the timeoutRoutine has already scheduled a timeout for a later height/round/step. func (t *timeoutTicker) ScheduleTimeout(ti timeoutInfo) { - t.tickChan <- ti + select { + case t.tickChan <- ti: + case <-t.Stopping(): + } } //------------------------------------------------------------- @@ -122,12 +129,13 @@ func (t *timeoutTicker) timeoutRoutine(ctx context.Context) { // Determinism comes from playback in the receiveRoutine. // We can eliminate it by merging the timeoutRoutine into receiveRoutine // and managing the timeouts ourselves with a millisecond ticker - go func(toi timeoutInfo) { + toi := ti + t.Go(ctx, func(ctx context.Context) { select { case t.tockChan <- toi: case <-ctx.Done(): } - }(ti) + }) case <-ctx.Done(): return } diff --git a/internal/consensus/ticker_test.go b/internal/consensus/ticker_test.go new file mode 100644 index 0000000000..4c44c9da0b --- /dev/null +++ b/internal/consensus/ticker_test.go @@ -0,0 +1,28 @@ +package consensus + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/dashpay/tenderdash/libs/log" +) + +func TestScheduleTimeoutDoesNotBlockAfterShutdown(t *testing.T) { + ticker := NewTimeoutTicker(log.NewNopLogger()).(*timeoutTicker) + require.NoError(t, ticker.Start(context.Background())) + ticker.Stop() + ticker.Wait() + for range cap(ticker.tickChan) { + ticker.tickChan <- timeoutInfo{} + } + done := make(chan struct{}) + go func() { ticker.ScheduleTimeout(timeoutInfo{}); close(done) }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("timeout scheduling blocked after ticker shutdown") + } +} diff --git a/internal/consensus/wal.go b/internal/consensus/wal.go index dbf3661dc4..31c19dfccc 100644 --- a/internal/consensus/wal.go +++ b/internal/consensus/wal.go @@ -8,7 +8,6 @@ import ( "hash/crc32" "io" "path/filepath" - "sync" "time" "github.com/cosmos/gogoproto/proto" @@ -103,17 +102,6 @@ type BaseWAL struct { flushTicker *time.Ticker flushInterval time.Duration - - // procCancel cancels the context supplied to processFlushTicks. It is - // derived from the caller ctx in OnStart (context.WithCancel) so the - // goroutine exits on either procCancel() or the parent ctx. OnStop calls - // procCancel() directly, which makes goroutineWg.Wait() in Wait() safe to - // call immediately after Stop() regardless of whether the caller ctx fired. - procCancel context.CancelFunc - - // goroutineWg tracks the processFlushTicks goroutine so that Wait() can - // block until it has truly exited. - goroutineWg sync.WaitGroup } var _ WAL = &BaseWAL{} @@ -163,17 +151,7 @@ func (wal *BaseWAL) OnStart(ctx context.Context) error { return err } wal.flushTicker = time.NewTicker(wal.flushInterval) - // Derive a child context so that OnStop can cancel the goroutine via - // procCancel() independently of the caller ctx. The child inherits - // cancellation from ctx so the existing "exit when caller ctx cancels" - // behavior is preserved. - procCtx, procCancel := context.WithCancel(ctx) - wal.procCancel = procCancel - wal.goroutineWg.Add(1) - go func() { - defer wal.goroutineWg.Done() - wal.processFlushTicks(procCtx) - }() + wal.Go(ctx, wal.processFlushTicks) return nil } @@ -200,45 +178,11 @@ func (wal *BaseWAL) FlushAndSync() error { // Use Wait() to ensure it's finished shutting down // before cleaning up files. func (wal *BaseWAL) OnStop() { - // Cancel the goroutine's context first so processFlushTicks exits on its - // next select iteration without waiting for the caller ctx to be canceled. - wal.procCancel() wal.flushTicker.Stop() - if err := wal.FlushAndSync(); err != nil { - wal.logger.Error("error on flush data to disk", "error", err) - } - wal.group.Stop() - wal.group.Close() } -// Wait for the underlying autofile group to finish shutting down -// so it's safe to cleanup files. -// -// This blocks until: -// 1. The WAL BaseService has stopped (OnStop has returned). -// 2. The Group BaseService has stopped. -// 3. The processFlushTicks goroutine has fully exited. -// 4. The Group's processTicks goroutine has fully exited. -// -// Points 3 and 4 are important: those goroutines run with a child context -// derived from the caller ctx. OnStop() cancels that child context via -// procCancel() so goroutineWg.Wait() / WaitForGoroutines() always unblock -// promptly after Stop() — regardless of whether the original caller ctx has -// fired. Without waiting for them, a goroutine still executing a flush tick -// could re-open the WAL file (via AutoFile) after the caller believes the -// WAL is fully stopped, causing a race with t.TempDir() cleanup -// ("directory not empty"). -func (wal *BaseWAL) Wait() { - if wal.IsRunning() { - wal.BaseService.Wait() - } - if wal.group.IsRunning() { - wal.group.Wait() - } - // Wait for context-driven goroutines that are not tracked by BaseService. - wal.group.WaitForGoroutines() - wal.goroutineWg.Wait() -} +// OnDrain flushes and closes files after the flush worker has exited. +func (wal *BaseWAL) OnDrain() { wal.group.Close() } // Write is called in newStep and for each receive on the // peerMsgQueue and the timeoutTicker. diff --git a/internal/consensus/wal_test.go b/internal/consensus/wal_test.go index cdc703da21..d409c12836 100644 --- a/internal/consensus/wal_test.go +++ b/internal/consensus/wal_test.go @@ -43,7 +43,7 @@ func TestWALTruncate(t *testing.T) { require.NoError(t, err) err = wal.Start(ctx) require.NoError(t, err) - t.Cleanup(func() { wal.Stop(); wal.Group().Stop(); wal.Group().Wait(); wal.Wait() }) + t.Cleanup(func() { wal.Stop(); wal.Wait() }) // 60 block's size nearly 70K, greater than group's headBuf size(4096 * 10), // when headBuf is full, truncate content will Flush to the file. at this @@ -111,7 +111,7 @@ func TestWALWrite(t *testing.T) { require.NoError(t, err) err = wal.Start(ctx) require.NoError(t, err) - t.Cleanup(func() { wal.Stop(); wal.Group().Stop(); wal.Group().Wait(); wal.Wait() }) + t.Cleanup(func() { wal.Stop(); wal.Wait() }) // 1) Write returns an error if msg is too big msg := &BlockPartMessage{ @@ -253,7 +253,7 @@ func TestWALPeriodicSync(t *testing.T) { assert.NotZero(t, wal.Group().Buffered()) require.NoError(t, wal.Start(ctx)) - t.Cleanup(func() { wal.Stop(); wal.Group().Stop(); wal.Group().Wait(); wal.Wait() }) + t.Cleanup(func() { wal.Stop(); wal.Wait() }) time.Sleep(walTestFlushInterval + (20 * time.Millisecond)) @@ -271,3 +271,36 @@ func TestWALPeriodicSync(t *testing.T) { t.Cleanup(leaktest.Check(t)) } + +func TestWALShutdownFlushesBufferedWrites(t *testing.T) { + for _, parentStop := range []bool{false, true} { + t.Run(map[bool]string{false: "manual", true: "parent"}[parentStop], func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + path := filepath.Join(t.TempDir(), "wal") + wal, err := NewWAL(ctx, log.NewNopLogger(), path) + require.NoError(t, err) + wal.SetFlushInterval(time.Hour) + require.NoError(t, wal.Start(ctx)) + require.NoError(t, wal.Write(EndHeightMessage{123})) + if parentStop { + cancel() + } else { + wal.Stop() + } + wal.Wait() + file, err := os.Open(path) + require.NoError(t, err) + defer file.Close() + decoder := NewWALDecoder(file) + first, err := decoder.Decode() + require.NoError(t, err) + require.Equal(t, EndHeightMessage{0}, first.Msg) + last, err := decoder.Decode() + require.NoError(t, err) + require.Equal(t, EndHeightMessage{123}, last.Msg) + require.ErrorIs(t, wal.Group().Head.Sync(), autofile.ErrAutoFileClosed) + require.NoError(t, os.Remove(path)) + }) + } +} diff --git a/internal/libs/autofile/autofile.go b/internal/libs/autofile/autofile.go index 7f2dcc0ee8..6f670fe40c 100644 --- a/internal/libs/autofile/autofile.go +++ b/internal/libs/autofile/autofile.go @@ -56,6 +56,7 @@ type AutoFile struct { closeTicker *time.Ticker // signals periodic close cancel func() // cancels the lifecycle context + done chan struct{} mtx sync.Mutex // guards the fields below closed bool // true when the the autofile is no longer usable @@ -78,9 +79,11 @@ func OpenAutoFile(ctx context.Context, path string) (*AutoFile, error) { Path: path, closeTicker: time.NewTicker(autoFileClosePeriod), cancel: cancel, + done: make(chan struct{}), } if err := af.openFile(); err != nil { - af.Close() + cancel() + af.closeTicker.Stop() return nil, err } @@ -89,42 +92,35 @@ func OpenAutoFile(ctx context.Context, path string) (*AutoFile, error) { hupc := make(chan os.Signal, 1) signal.Notify(hupc, syscall.SIGHUP) go func() { - defer close(hupc) + defer close(af.done) + defer signal.Stop(hupc) + defer af.closeTicker.Stop() for { select { case <-hupc: _ = af.closeFile() + case <-af.closeTicker.C: + _ = af.closeFile() case <-ctx.Done(): + _ = af.withLock(func() error { af.closed = true; return af.unsyncCloseFile() }) return } } }() - go af.closeFileRoutine(ctx) - return af, nil } // Close shuts down the service goroutine and marks af as invalid. Operations // on af after Close will report an error. func (af *AutoFile) Close() error { - return af.withLock(func() error { - af.cancel() // signal the close service to stop - af.closed = true // mark the file as invalid + err := af.withLock(func() error { + af.cancel() + af.closed = true return af.unsyncCloseFile() }) -} - -func (af *AutoFile) closeFileRoutine(ctx context.Context) { - for { - select { - case <-ctx.Done(): - _ = af.Close() - return - case <-af.closeTicker.C: - _ = af.closeFile() - } - } + <-af.done + return err } func (af *AutoFile) closeFile() (err error) { diff --git a/internal/libs/autofile/autofile_test.go b/internal/libs/autofile/autofile_test.go index 9dbba276ac..c79eb9719c 100644 --- a/internal/libs/autofile/autofile_test.go +++ b/internal/libs/autofile/autofile_test.go @@ -144,3 +144,19 @@ func mustReadFile(t *testing.T, filePath string) []byte { return fileBytes } + +func TestCloseJoinsAutoFileWorker(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + af, err := OpenAutoFile(ctx, filepath.Join(t.TempDir(), "wal")) + require.NoError(t, err) + require.NoError(t, af.Close()) + select { + case <-af.done: + default: + t.Fatal("Close returned before worker exited") + } + require.NoError(t, af.Close()) + _, err = af.Write([]byte("late write")) + require.ErrorIs(t, err, ErrAutoFileClosed) +} diff --git a/internal/libs/autofile/group.go b/internal/libs/autofile/group.go index 71bb8d04bf..fc0c484040 100644 --- a/internal/libs/autofile/group.go +++ b/internal/libs/autofile/group.go @@ -11,7 +11,6 @@ import ( "regexp" "strconv" "strings" - stdsync "sync" "time" sync "github.com/sasha-s/go-deadlock" @@ -71,19 +70,6 @@ type Group struct { minIndex int // Includes head maxIndex int // Includes head, where Head will move to - // procCancel cancels the context supplied to background goroutines - // (processTicks). It is derived from the caller ctx in OnStart so that - // goroutines also exit when the caller ctx is canceled. OnStop calls - // procCancel() directly, which makes goroutineWg.Wait() safe to call - // immediately after Stop() regardless of whether the caller ctx has fired. - procCancel context.CancelFunc - - // goroutineWg tracks the processTicks goroutine so that WaitForGoroutines - // can block until it has fully exited. - goroutineWg stdsync.WaitGroup - - // TODO: When we start deleting files, we need to start tracking GroupReaders - // and their dependencies. } // OpenGroup creates a new Group with head at headPath. It returns an error if @@ -93,7 +79,8 @@ func OpenGroup(ctx context.Context, logger log.Logger, headPath string, groupOpt if err != nil { return nil, err } - head, err := OpenAutoFile(ctx, headPath) + // The owner closes Head after the group's final flush, not on cancellation. + head, err := OpenAutoFile(context.WithoutCancel(ctx), headPath) if err != nil { return nil, err } @@ -148,56 +135,27 @@ func GroupTotalSizeLimit(limit int64) func(*Group) { // and group limits. func (g *Group) OnStart(ctx context.Context) error { g.ticker = time.NewTicker(g.groupCheckDuration) - // Derive a child context so that OnStop can cancel the goroutine by - // calling g.procCancel() independently of whether the caller ctx fires. - // The child inherits cancellation from ctx, preserving the existing - // "also exit when caller ctx cancels" behavior. - procCtx, procCancel := context.WithCancel(ctx) - g.procCancel = procCancel - g.goroutineWg.Add(1) - go func() { - defer g.goroutineWg.Done() - g.processTicks(procCtx) - }() + g.Go(ctx, g.processTicks) return nil } -// WaitForGoroutines blocks until the background goroutines started by this -// Group (e.g. processTicks) have fully exited. It is safe to call after -// Stop() because OnStop() cancels the goroutine's context via procCancel. -func (g *Group) WaitForGoroutines() { - g.goroutineWg.Wait() -} - -// Wait blocks until the Group service has stopped and its background goroutines -// have fully exited. It overrides BaseService.Wait — which returns as soon as -// OnStop() closes the quit channel, before the procCancel-driven processTicks -// goroutine is joined — so that every direct Group consumer gets the same -// shutdown guarantee BaseWAL.Wait relies on, not just callers that remember to -// chain WaitForGoroutines explicitly. -// -// It calls g.BaseService.Wait() (the embedded method), not g.Wait(), so there -// is no recursion. WaitForGoroutines is idempotent, so chaining it here on top -// of an explicit caller call is harmless. -func (g *Group) Wait() { - g.BaseService.Wait() - g.WaitForGoroutines() -} - // OnStop implements service.Service by stopping the goroutine described above. // NOTE: g.Head must be closed separately using Close. func (g *Group) OnStop() { - // Cancel the goroutine's context first so that processTicks exits on its - // next select iteration without waiting for the caller ctx. - g.procCancel() g.ticker.Stop() +} + +// OnDrain flushes buffered writes after the periodic worker has exited. +func (g *Group) OnDrain() { if err := g.FlushAndSync(); err != nil { g.logger.Error("error flushing to disk", "err", err) } } -// Close closes the head file. The group must be stopped by this moment. +// Close stops and drains the group before closing its head file. func (g *Group) Close() { + g.Stop() + g.Wait() if err := g.FlushAndSync(); err != nil { g.logger.Error("error flushing to disk", "err", err) } diff --git a/internal/p2p/conn/connection.go b/internal/p2p/conn/connection.go index 63e7317287..fd3aa767bd 100644 --- a/internal/p2p/conn/connection.go +++ b/internal/p2p/conn/connection.go @@ -199,6 +199,13 @@ func NewMConnection( return mconn } +type errorContextKey struct{} + +// Start preserves the caller's context for error notification after I/O stops. +func (c *MConnection) Start(ctx context.Context) error { + return c.BaseService.Start(context.WithValue(ctx, errorContextKey{}, ctx)) +} + // OnStart implements BaseService func (c *MConnection) OnStart(ctx context.Context) error { c.flushTimer = timer.NewThrottleTimer("flush", c.config.FlushThrottle) @@ -293,6 +300,9 @@ func (c *MConnection) stopForError(ctx context.Context, r interface{}) { if atomic.CompareAndSwapUint32(&c.errored, 0, 1) { if c.onError != nil { + if parent, ok := ctx.Value(errorContextKey{}).(context.Context); ok { + ctx = parent + } c.onError(ctx, r) } } diff --git a/internal/p2p/conn/lifecycle_test.go b/internal/p2p/conn/lifecycle_test.go new file mode 100644 index 0000000000..4411924758 --- /dev/null +++ b/internal/p2p/conn/lifecycle_test.go @@ -0,0 +1,40 @@ +package conn + +import ( + "context" + "net" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/dashpay/tenderdash/libs/log" +) + +func TestMConnectionErrorCallbackRetainsParentContext(t *testing.T) { + local, remote := net.Pipe() + t.Cleanup(func() { _ = local.Close(); _ = remote.Close() }) + type observation struct { + err error + running bool + } + observed := make(chan observation, 1) + release := make(chan struct{}) + var connection *MConnection + connection = createMConnectionWithCallbacks(log.NewNopLogger(), local, nil, + func(ctx context.Context, _ interface{}) { + observed <- observation{err: ctx.Err(), running: connection.IsRunning()} + <-release + }) + require.NoError(t, connection.Start(context.Background())) + t.Cleanup(func() { connection.Stop(); connection.Wait() }) + defer close(release) + require.NoError(t, remote.Close()) + select { + case result := <-observed: + require.NoError(t, result.err) + require.False(t, result.running) + case <-time.After(time.Second): + t.Fatal("error callback did not run") + } +} diff --git a/libs/service/lifecycle_test.go b/libs/service/lifecycle_test.go new file mode 100644 index 0000000000..4a0b2479dc --- /dev/null +++ b/libs/service/lifecycle_test.go @@ -0,0 +1,268 @@ +package service + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/dashpay/tenderdash/libs/log" +) + +type lifecycleService struct { + *BaseService + start func(context.Context) error + stop func() +} + +func (s *lifecycleService) OnStart(ctx context.Context) error { return s.start(ctx) } +func (s *lifecycleService) OnStop() { + if s.stop != nil { + s.stop() + } +} +func newLifecycleService(t *testing.T) *lifecycleService { + s := &lifecycleService{} + s.BaseService = NewBaseService(log.NewNopLogger(), t.Name(), s) + return s +} + +func TestManualStopCancelsStartContext(t *testing.T) { + s := newLifecycleService(t) + var workCtx context.Context + s.start = func(ctx context.Context) error { workCtx = ctx; return nil } + require.NoError(t, s.Start(context.Background())) + s.Stop() + s.Wait() + require.ErrorIs(t, workCtx.Err(), context.Canceled) +} + +func assertPending(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + t.Error("Wait returned before worker cleanup completed") + case <-time.After(20 * time.Millisecond): + } +} + +func TestWaitDrainsManagedWork(t *testing.T) { + for _, parentStop := range []bool{false, true} { + t.Run(map[bool]string{false: "manual", true: "parent"}[parentStop], func(t *testing.T) { + s := newLifecycleService(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + entered, release := make(chan struct{}), make(chan struct{}) + s.start = func(ctx context.Context) error { + require.True(t, s.Go(ctx, func(ctx context.Context) { <-ctx.Done(); close(entered); <-release })) + return nil + } + require.NoError(t, s.Start(ctx)) + if parentStop { + cancel() + } else { + s.Stop() + } + <-entered + done := make(chan struct{}) + go func() { s.Wait(); close(done) }() + assertPending(t, done) + close(release) + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("worker was not drained") + } + require.False(t, s.IsRunning()) + require.ErrorIs(t, s.Start(ctx), errAlreadyStopped) + }) + } +} + +func TestFailedStartDrainsBeforeRetry(t *testing.T) { + s := newLifecycleService(t) + entered, release := make(chan struct{}), make(chan struct{}) + var old context.Context + s.start = func(ctx context.Context) error { + old = ctx + require.True(t, s.Go(ctx, func(ctx context.Context) { <-ctx.Done(); close(entered); <-release })) + return context.DeadlineExceeded + } + s.Wait() + done := make(chan struct{}) + go func() { defer close(done); require.ErrorIs(t, s.Start(context.Background()), context.DeadlineExceeded) }() + <-entered + assertPending(t, done) + close(release) + <-done + s.Wait() + s.start = func(_ context.Context) error { + require.False(t, s.Go(old, func(context.Context) { t.Error("old attempt admitted") })) + return nil + } + require.NoError(t, s.Start(context.Background())) + require.True(t, s.IsRunning()) + s.Stop() + s.Wait() +} + +func TestWorkerCanStopAndHookCanInspectLifecycle(t *testing.T) { + s := newLifecycleService(t) + s.stop = func() { require.False(t, s.IsRunning()); s.Stop() } + s.start = func(ctx context.Context) error { + require.False(t, s.IsRunning()) + require.True(t, s.Go(ctx, func(context.Context) { s.Stop() })) + return nil + } + require.NoError(t, s.Start(context.Background())) + done := make(chan struct{}) + go func() { s.Wait(); close(done) }() + select { + case <-done: + case <-time.After(time.Second): + t.Fatal("lifecycle deadlock") + } +} + +func TestRegistrationRacesStop(t *testing.T) { + for range 100 { + s := newLifecycleService(t) + var ctx context.Context + s.start = func(c context.Context) error { ctx = c; return nil } + require.NoError(t, s.Start(context.Background())) + admittedDone := make(chan struct{}) + require.True(t, s.Go(ctx, func(ctx context.Context) { <-ctx.Done(); close(admittedDone) })) + registered := make(chan struct{}) + go func() { + defer close(registered) + for range 100 { + s.Go(ctx, func(ctx context.Context) { <-ctx.Done() }) + } + }() + s.Stop() + s.Wait() + <-registered + select { + case <-admittedDone: + default: + t.Error("admitted work escaped drain") + } + require.False(t, s.Go(ctx, func(context.Context) { t.Error("work admitted after stop") })) + } +} + +type finalizingService struct { + *lifecycleService + drain func() +} + +func (s *finalizingService) OnDrain() { s.drain() } + +func TestFinalizerFollowsWorkerCleanup(t *testing.T) { + workerDone := make(chan struct{}) + release := make(chan struct{}) + finalizerEntered := make(chan struct{}) + s := &finalizingService{lifecycleService: newLifecycleService(t)} + s.BaseService = NewBaseService(log.NewNopLogger(), t.Name(), s) + s.start = func(ctx context.Context) error { + require.True(t, s.Go(ctx, func(ctx context.Context) { <-ctx.Done(); close(workerDone) })) + return nil + } + s.drain = func() { + select { + case <-workerDone: + default: + t.Error("finalizer ran before worker cleanup") + } + close(finalizerEntered) + <-release + } + require.NoError(t, s.Start(context.Background())) + s.Stop() + <-finalizerEntered + done := make(chan struct{}) + go func() { s.Wait(); close(done) }() + assertPending(t, done) + close(release) + <-done +} + +func TestStopDuringStartup(t *testing.T) { + s := newLifecycleService(t) + entered, release := make(chan struct{}), make(chan struct{}) + stopped := make(chan struct{}) + s.start = func(ctx context.Context) error { + close(entered) + <-ctx.Done() + <-release + return nil + } + s.stop = func() { close(stopped) } + started := make(chan error, 1) + go func() { started <- s.Start(context.Background()) }() + <-entered + require.ErrorIs(t, s.Start(context.Background()), errStarting) + s.Stop() + select { + case <-stopped: + t.Fatal("OnStop ran concurrently with OnStart") + default: + } + close(release) + require.NoError(t, <-started) + s.Wait() + <-stopped + require.False(t, s.IsRunning()) +} + +func TestStartContextPreservesValuesAndDeadline(t *testing.T) { + type key struct{} + deadline := time.Now().Add(time.Hour) + parent, cancel := context.WithDeadline(context.WithValue(context.Background(), key{}, "value"), deadline) + defer cancel() + s := newLifecycleService(t) + s.start = func(ctx context.Context) error { + require.Equal(t, "value", ctx.Value(key{})) + got, ok := ctx.Deadline() + require.True(t, ok) + require.Equal(t, deadline, got) + return nil + } + require.False(t, s.Go(parent, func(context.Context) { t.Error("work before Start") })) + require.NoError(t, s.Start(parent)) + require.NoError(t, s.Start(parent)) + s.Stop() + s.Wait() + require.NoError(t, parent.Err()) +} + +func TestFailedStartupRacingStopSkipsShutdownHooks(t *testing.T) { + entered, cleaning, release := make(chan struct{}), make(chan struct{}), make(chan struct{}) + s := &finalizingService{lifecycleService: newLifecycleService(t)} + s.BaseService = NewBaseService(log.NewNopLogger(), t.Name(), s) + s.stop = func() { t.Error("OnStop called for failed startup") } + s.drain = func() { t.Error("OnDrain called for failed startup") } + s.start = func(ctx context.Context) error { + s.Go(ctx, func(ctx context.Context) { <-ctx.Done(); close(cleaning); <-release }) + close(entered) + <-ctx.Done() + return context.DeadlineExceeded + } + started := make(chan error, 1) + go func() { started <- s.Start(context.Background()) }() + <-entered + s.Stop() + <-cleaning + waited := make(chan struct{}) + go func() { s.Wait(); close(waited) }() + assertPending(t, waited) + if len(started) != 0 { + t.Error("failed Start returned before cleanup") + } + close(release) + require.ErrorIs(t, <-started, context.DeadlineExceeded) + <-waited + s.Wait() + require.False(t, s.IsRunning()) +} diff --git a/libs/service/service.go b/libs/service/service.go index 50220d1f53..32ca8b99e7 100644 --- a/libs/service/service.go +++ b/libs/service/service.go @@ -3,6 +3,7 @@ package service import ( "context" "errors" + stdsync "sync" "sync/atomic" sync "github.com/sasha-s/go-deadlock" @@ -11,187 +12,196 @@ import ( ) var ( - // errAlreadyStopped is returned when somebody tries to stop an already - // stopped service (without resetting it). - errAlreadyStopped = errors.New("already stopped") - - _ Service = (*BaseService)(nil) + errAlreadyStopped = errors.New("already stopped") + errStarting = errors.New("service is starting") + _ Service = (*BaseService)(nil) ) -// Service defines a service that can be started, stopped, and reset. +// Service defines a service that can be started and stopped. type Service interface { - // Start is called to start the service, which should run until - // the context terminates. If the service is already running, Start - // must report an error. Start(context.Context) error - - // Return true if the service is running IsRunning() bool - - // Wait blocks until the service is stopped. Wait() } -// Implementation describes the implementation that the -// BaseService implementation wraps. +// Implementation supplies startup and shutdown hooks. Hooks run without the +// lifecycle lock. They must not call Wait on their own service. type Implementation interface { - // Called by the Services Start Method OnStart(context.Context) error - - // Called when the service's context is canceled. + // OnStop unblocks workers. It must not wait for managed workers. OnStop() } -/* -Classical-inheritance-style service declarations. Services can be started, then -stopped, but cannot be restarted. - -Users must implement OnStart/OnStop methods. In the absence of errors, these -methods are guaranteed to be called at most once. If OnStart returns an error, -service won't be marked as started, so the user can call Start again. - -The BaseService implementation ensures that the OnStop method is -called after the context passed to Start is canceled. - -Typical usage: - - type FooService struct { - BaseService - // private fields - } +// Finalizer releases resources after OnStop and all managed workers finish. +// It is called only after a successful OnStart, before Wait returns. +type Finalizer interface { + OnDrain() +} - func NewFooService() *FooService { - fs := &FooService{ - // init - } - fs.BaseService = *NewBaseService(log, "FooService", fs) - return fs - } +type attemptKey struct{} - func (fs *FooService) OnStart(ctx context.Context) error { - // initialize private fields - // start subroutines, etc. - } +type serviceAttempt struct { + cancel context.CancelFunc + stoppingCh <-chan struct{} + done chan struct{} + workers stdsync.WaitGroup + starting bool + stopping bool +} - func (fs *FooService) OnStop() { - // close/destroy private fields and releases resources - } -*/ +// BaseService owns a service context and work registered through Go. Stop cancels +// that context and calls OnStop; Wait joins work and optional OnDrain cleanup. +// Plain go statements are not tracked. A successful start cannot be restarted. +// A failed OnStart must release its own resources; BaseService cancels and joins +// its managed work before allowing another attempt. type BaseService struct { - logger log.Logger - name string - mtx sync.Mutex - quit <-chan (struct{}) - cancel context.CancelFunc - - // running is an atomic flag set to 1 when the service is running, - // and 0 otherwise. It allows IsRunning to avoid locking and thus - // prevents recursive lock deadlocks from callers that also hold mtx. + logger log.Logger + name string + mtx sync.Mutex + attempt *serviceAttempt running uint32 - - // The "subclass" of BaseService - impl Implementation + impl Implementation } // NewBaseService creates a new BaseService. func NewBaseService(logger log.Logger, name string, impl Implementation) *BaseService { - return &BaseService{ - logger: logger, - name: name, - impl: impl, - } + return &BaseService{logger: logger, name: name, impl: impl} } -// Start starts the Service and calls its OnStart method. An error -// will be returned if the service is stopped, but not if it is -// already running. +// Start calls OnStart with a context canceled by either the parent or Stop. +// Repeated calls while running succeed; concurrent startup returns an error. func (bs *BaseService) Start(ctx context.Context) error { bs.mtx.Lock() - defer bs.mtx.Unlock() - - if bs.quit != nil { - return nil - } - - select { - case <-bs.quit: - return errAlreadyStopped - default: - bs.logger.Info("starting service", "service", bs.name, "impl", bs.name) - if err := bs.impl.OnStart(ctx); err != nil { - return err + if a := bs.attempt; a != nil { + var err error + switch { + case a.stopping: + err = errAlreadyStopped + case a.starting: + err = errStarting } - - // we need a separate context to ensure that we start - // a thread that will get cleaned up and that the - // Stop/Wait functions work as expected. - srvCtx, cancel := context.WithCancel(context.Background()) - bs.cancel = cancel - bs.quit = srvCtx.Done() + bs.mtx.Unlock() + return err + } + workCtx, cancel := context.WithCancel(ctx) + a := &serviceAttempt{cancel: cancel, stoppingCh: workCtx.Done(), done: make(chan struct{}), starting: true} + bs.attempt = a + workCtx = context.WithValue(workCtx, attemptKey{}, a) + bs.mtx.Unlock() + + bs.logger.Info("starting service", "service", bs.name) + err := bs.impl.OnStart(workCtx) + bs.mtx.Lock() + a.starting = false + if err != nil { + a.stopping = true + bs.mtx.Unlock() + cancel() + a.workers.Wait() + bs.mtx.Lock() + close(a.done) + bs.attempt = nil + bs.mtx.Unlock() + return err + } + stopping := a.stopping + if !stopping { atomic.StoreUint32(&bs.running, 1) - - go func(ctx context.Context) { + } + bs.mtx.Unlock() + if stopping { + bs.finishStop(a) + } else { + go func() { select { - case <-srvCtx.Done(): - // this means stop was called manually - return - case <-ctx.Done(): - bs.Stop() + case <-workCtx.Done(): + bs.stopAttempt(a) + case <-a.done: } - - bs.logger.Info("stopped service", - "service", bs.name) - }(ctx) - - return nil + }() } + return nil } -// Stop manually terminates the service by calling OnStop method from -// the implementation and releases all resources related to the -// service. -func (bs *BaseService) Stop() { +// Go registers work belonging to the context received by OnStart (or a child). +// It returns false before startup, after stopping begins, or for an old attempt. +// A worker may call Stop, but must not call Wait on its own service. +func (bs *BaseService) Go(ctx context.Context, fn func(context.Context)) bool { bs.mtx.Lock() defer bs.mtx.Unlock() - - if bs.quit == nil { - return + a := bs.attempt + if a == nil || a.stopping || ctx.Value(attemptKey{}) != a { + return false } + a.workers.Add(1) + go func() { defer a.workers.Done(); fn(ctx) }() + return true +} - select { - case <-bs.quit: - return - default: - bs.logger.Info("stopping service", "service", bs.name) - bs.impl.OnStop() - // Clear running before cancel: cancel() closes quit, which unblocks Wait(). - // Storing first guarantees IsRunning() is false by the time Wait() returns. - atomic.StoreUint32(&bs.running, 0) - bs.cancel() +// Stop cancels work and invokes OnStop once, without waiting for workers. +// During startup it requests shutdown after OnStart returns. Concurrent or +// recursive calls return immediately; use Wait to observe complete shutdown. +func (bs *BaseService) Stop() { + bs.mtx.Lock() + a := bs.attempt + bs.mtx.Unlock() + bs.stopAttempt(a) +} +func (bs *BaseService) stopAttempt(a *serviceAttempt) { + bs.mtx.Lock() + if a == nil || bs.attempt != a || a.stopping { + bs.mtx.Unlock() return } + a.stopping = true + atomic.StoreUint32(&bs.running, 0) + starting := a.starting + bs.mtx.Unlock() + a.cancel() + if !starting { + bs.finishStop(a) + } +} + +func (bs *BaseService) finishStop(a *serviceAttempt) { + bs.logger.Info("stopping service", "service", bs.name) + bs.impl.OnStop() + go func() { + a.workers.Wait() + if finalizer, ok := bs.impl.(Finalizer); ok { + finalizer.OnDrain() + } + bs.logger.Info("stopped service", "service", bs.name) + close(a.done) + }() } -// IsRunning implements Service by returning true or false depending on the -// service's state. +// IsRunning reports successful startup until shutdown is requested. False does +// not imply cleanup has completed; use Wait before releasing owned resources. func (bs *BaseService) IsRunning() bool { return atomic.LoadUint32(&bs.running) == 1 } -func (bs *BaseService) getWait() <-chan struct{} { +// Stopping signals context cancellation, not completed shutdown. It is nil +// before Start and after a failed attempt has drained; use Wait for completion. +func (bs *BaseService) Stopping() <-chan struct{} { bs.mtx.Lock() defer bs.mtx.Unlock() - - if bs.quit == nil { - out := make(chan struct{}) - close(out) - return out + if bs.attempt == nil { + return nil } - - return bs.quit + return bs.attempt.stoppingCh } -// Wait blocks until the service is stopped. -func (bs *BaseService) Wait() { <-bs.getWait() } +// Wait joins the current startup attempt, shutdown hooks and registered work. +// It returns immediately before Start or after a failed Start has drained. +func (bs *BaseService) Wait() { + bs.mtx.Lock() + a := bs.attempt + bs.mtx.Unlock() + if a != nil { + <-a.done + } +} // String provides a human-friendly representation of the service. func (bs *BaseService) String() string { return bs.name } diff --git a/node/node.go b/node/node.go index d967353f21..f3a58380d9 100644 --- a/node/node.go +++ b/node/node.go @@ -630,16 +630,16 @@ func (n *nodeImpl) OnStop() { } } + for _, reactor := range n.services { + reactor.Wait() + } + for _, es := range n.eventSinks { if err := es.Stop(); err != nil { n.logger.Error("failed to stop event sink", "err", err) } } - for _, reactor := range n.services { - reactor.Wait() - } - n.router.Wait() n.rpcEnv.IsListening = false diff --git a/node/node_test.go b/node/node_test.go index 106b2e23f5..1289a77b9b 100644 --- a/node/node_test.go +++ b/node/node_test.go @@ -1214,8 +1214,6 @@ func (s startupPausedService) Start(ctx context.Context) error { } func TestNodeStartDuringFirstCommit(t *testing.T) { - // Wait for consensus goroutines before TempDir cleanup removes their files. - defer leaktest.CheckTimeout(t, 5*time.Second)() cfg, err := config.ResetTestRoot(t.TempDir(), t.Name()) require.NoError(t, err) cfg.Consensus = config.DefaultConsensusConfig() From ec8d306d087c31aca21dedc2bc1bbab0121d47d8 Mon Sep 17 00:00:00 2001 From: Lukasz Klimek <842586+lklimek@users.noreply.github.com> Date: Fri, 25 Sep 2026 11:39:08 +0000 Subject: [PATCH 3/3] fix(consensus): preserve in-flight work during manual shutdown Keep processing on the State parent's lifetime while canceling intake, and let handoff callers cancel their wait without abandoning reactor-owned work. Document the global cancellation contract, cleanup ownership and migration boundaries; harden context lookup and close logjack's group on return. Co-Authored-By: OpenAI GPT-6 --- UPGRADING.md | 35 ++++++ docs/architecture/service-lifecycle.md | 44 ++++++- internal/blocksync/synchronizer.go | 7 +- internal/consensus/reactor.go | 8 +- internal/consensus/shutdown_test.go | 154 +++++++++++++++++++++++++ internal/consensus/state.go | 20 +++- internal/consensus/ticker.go | 2 +- internal/consensus/wal.go | 4 +- internal/libs/autofile/cmd/logjack.go | 2 + internal/libs/autofile/group.go | 3 +- internal/p2p/router.go | 2 +- 11 files changed, 263 insertions(+), 18 deletions(-) create mode 100644 internal/consensus/shutdown_test.go diff --git a/UPGRADING.md b/UPGRADING.md index 9a1e0f541c..3845800e18 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -2,6 +2,41 @@ This guide provides instructions for upgrading to specific versions of Tenderdash. +## Unreleased + +### Service lifecycle + +All implementations embedding `libs/service.BaseService` receive a context in +`OnStart` that is canceled by either the caller or manual `Stop`. Cancellation +now precedes `OnStop`, including for implementations that have not adopted +managed workers. Audit shutdown callbacks that require a live context; preserve +the appropriate owner context explicitly instead of using the canceled work context. + +Register owned workers with `BaseService.Go`. `OnStop` must unblock them, not +wait for them; move shared-resource cleanup to optional `OnDrain`, which runs +after those workers exit. Call `Stop` then `Wait` before releasing resources. +Plain goroutines and their existing joins remain the implementation's responsibility. +`TimeoutTicker` implementations must now provide `Wait`. + +A stop requested during startup cancels work immediately but defers `OnStop` +until successful startup returns. Failed startup must roll back its resources; +BaseService joins registered work without calling shutdown hooks. Retry requires +recreating any child that already started successfully and then stopped. + +Direct consensus `State.Stop` drains a commit already in progress using the +context originally passed to `State.Start`. Cancel that parent to abort it; +reactor/node shutdown still cancels that parent and can require WAL recovery. +This is not an unconditional durable-finalization guarantee for node shutdown. +Once admitted, a consensus handoff belongs to its reactor. Canceling the caller +releases its wait; reactor shutdown cancels and joins the admitted handoff. +Router disconnect notifications during shutdown are best effort; subscribers +must terminate on their own context, not wait for a final `PeerStatusDown`. + +Autofile groups require explicit `Close`, even if never started. Canceling the +opening context or calling `Stop`/`Wait` alone does not close the head file. +See [service lifecycle](docs/architecture/service-lifecycle.md) for ownership +and migration boundaries. + ## v1.7.0 ### Consensus DoS hardening (peer verification limits) diff --git a/docs/architecture/service-lifecycle.md b/docs/architecture/service-lifecycle.md index 5c4132117f..a7da59cba5 100644 --- a/docs/architecture/service-lifecycle.md +++ b/docs/architecture/service-lifecycle.md @@ -9,7 +9,9 @@ Shutdown closes admission, marks `IsRunning` false, cancels the context and calls `OnStop`. It then joins registered workers, calls optional `OnDrain`, and releases `Wait`. `Stopping()` signals cancellation separately from completion. `Stop` invokes `OnStop` synchronously but does not join registered workers. -Concurrent or recursive calls to `Stop` return immediately. +Concurrent or recursive calls to `Stop` return immediately. During `OnStart`, +`Stop` cancels immediately and defers the shutdown hook until successful startup +returns; `Wait` still joins the attempt. Hooks run outside the lifecycle mutex. `OnStop` must unblock workers without waiting for its own registered work. Workers may call `Stop`; neither hooks nor @@ -29,7 +31,9 @@ Consensus State registers its receive loop and joins its queue fan-in and WAL before returning. The timeout ticker, WAL and autofile group use shared worker tracking. AutoFile explicitly joins its timer/signal worker. WAL ownership lasts until the receive loop's final write. Consensus handoff admission is also joined, -so a late blocksync handoff cannot start State after reactor shutdown. +so a late blocksync handoff cannot start State after reactor shutdown. The caller +may cancel its wait after admission, but the reactor still owns and joins the +handoff. Startup and the resulting State use the reactor's lifetime. Node waits for reactors before closing event sinks and stores. The blocked receive-loop regression and original node teardown tests verify this path @@ -41,6 +45,36 @@ workers retain their existing ownership mechanisms. This change does not claim that Node.Wait joins every helper or introduce the broader durable-finalization and dependency-lifetime changes from #1515. -Two compatibility bridges preserve existing parent-context lifetimes: blocksync -application during handoff and connection error callbacks after I/O shutdown. -Their local cancellation and join mechanisms remain in place in this stage. +Three compatibility bridges preserve parent-context lifetimes: blocksync +application during handoff, connection error callbacks after I/O shutdown, and +an admitted consensus commit during direct State.Stop. Parent cancellation still +aborts these operations. In particular, reactor/node shutdown can interrupt +finalization and require WAL recovery. Their existing ownership mechanisms remain +in place; this does not detach application work from node shutdown. + +## Cancellation compatibility audit + +Cancellation precedes OnStop for every BaseService implementation, not only the +managed-worker migrations. The in-tree hook/callback audit covers these groups: + +| Implementations | Cancellation and cleanup policy | +| --- | --- | +| Consensus State, reactor, ticker, WAL; autofile Group | Managed work is joined; commit uses the State parent context; file ownership ends with explicit Close. | +| Blocksync reactor and Synchronizer | Handoff preserves application work across Synchronizer.Stop; stopping its parent reactor may cancel it. | +| P2P Router and MConnection | Connection callbacks retain the parent. Router removes peer bookkeeping even with a canceled context; final PeerStatusDown broadcasts are best effort. | +| PEX, mempool, statesync and evidence reactors | Processing uses the work context; stop hooks close resources/dispatchers or require no extra signal. Existing plain-worker ownership is unchanged. | +| EventBus, pubsub Server and indexer Service | Event processing is cancellation-driven; indexer closes sinks in its existing hook. No guarantee of a live cleanup callback context. | +| ABCI local, socket, gRPC and routed clients; proxy; socket/gRPC servers | Stop hooks close transports, drain requests or stop children without requiring a live work context. | +| SignerServer, signer listener/dialer endpoints | Hooks close endpoints/listeners; dial/retry work observes cancellation. | +| ValidatorConnExecutor and light RPC Client | Unsubscribe uses a separate bounded cleanup context; client hooks invoke owned closers. | +| Node and seed node | Parent cancellation stops children; hooks wait for existing service completion before closing owned stores. | + +This is a cancellation compatibility audit, not a claim that unmigrated plain +workers are all joined. Subscribers must stop with their owner, without relying +on a final router broadcast. Avoid detaching broadcasts from cancellation: +unresponsive subscribers could otherwise block shutdown. + +Group rotation reuses its AutoFile and closes the current descriptor before +renaming; it does not create a worker or signal registration per rotation. The +owner must Close the group, including when it never started, to join the head's +single timer/signal worker. Context cancellation alone does not release the head. diff --git a/internal/blocksync/synchronizer.go b/internal/blocksync/synchronizer.go index 68429fc50e..cdc3f3140d 100644 --- a/internal/blocksync/synchronizer.go +++ b/internal/blocksync/synchronizer.go @@ -252,14 +252,17 @@ func (s *Synchronizer) OnStart(ctx context.Context) error { s.lastAdvance = s.clock.Now() s.lastMonitorUpdate = s.lastAdvance s.ctx, s.cancel = context.WithCancel(ctx) - applicationCtx := ctx.Value(applicationContextKey{}).(context.Context) + applicationCtx, ok := ctx.Value(applicationContextKey{}).(context.Context) + if !ok { + applicationCtx = ctx + } s.consumerDone = make(chan struct{}) s.workerPool.Run(s.ctx) go s.runHandler(s.ctx, s.produceJob) go func() { defer close(s.consumerDone) s.runHandler(s.ctx, func(handlerCtx context.Context) error { - // Handover cancels consumer I/O; only node shutdown may cancel application. + // Handover cancels consumer I/O; parent cancellation also stops application. return s.consumeJobResult(handlerCtx, applicationCtx) }) }() diff --git a/internal/consensus/reactor.go b/internal/consensus/reactor.go index e84301fb31..57094bee2e 100644 --- a/internal/consensus/reactor.go +++ b/internal/consensus/reactor.go @@ -445,7 +445,8 @@ func (r *Reactor) WaitSync() bool { // the state, turns off block-sync, and starts the consensus state-machine. // // targetHeight is the highest committed block height reported during block sync. -// skipWAL says the node needs no WAL catchup. +// skipWAL says the node needs no WAL catchup. After admission, the reactor owns +// the handoff; caller cancellation releases the wait without canceling startup. func (r *Reactor) SwitchToConsensus(ctx context.Context, state sm.State, skipWAL bool, targetHeight int64) { if r.ctx == nil || ctx.Err() != nil { return @@ -457,7 +458,10 @@ func (r *Reactor) SwitchToConsensus(ctx context.Context, state sm.State, skipWAL }) { return } - <-done + select { + case <-done: + case <-ctx.Done(): + } } func (r *Reactor) switchToConsensus(ctx context.Context, state sm.State, skipWAL bool, targetHeight int64) { diff --git a/internal/consensus/shutdown_test.go b/internal/consensus/shutdown_test.go new file mode 100644 index 0000000000..5b401ab122 --- /dev/null +++ b/internal/consensus/shutdown_test.go @@ -0,0 +1,154 @@ +package consensus + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/dashpay/tenderdash/abci/example/kvstore" + abci "github.com/dashpay/tenderdash/abci/types" + "github.com/dashpay/tenderdash/dash" +) + +type blockedFinalizeApp struct { + abci.Application + entered chan context.Context + release chan struct{} +} + +func (a *blockedFinalizeApp) FinalizeBlock(ctx context.Context, req *abci.RequestFinalizeBlock) (*abci.ResponseFinalizeBlock, error) { + a.entered <- ctx + <-a.release + if err := ctx.Err(); err != nil { + return nil, err + } + return a.Application.FinalizeBlock(ctx, req) +} + +func TestStateStopDuringFinalize(t *testing.T) { + for _, cancelParent := range []bool{false, true} { + name := "manual stop drains commit" + if cancelParent { + name = "parent cancellation aborts commit" + } + t.Run(name, func(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + app, err := kvstore.NewMemoryApp() + require.NoError(t, err) + defer func() { require.NoError(t, app.Close()) }() + blocked := &blockedFinalizeApp{Application: app, entered: make(chan context.Context, 1), release: make(chan struct{})} + cs, _ := makeState(ctx, t, makeStateArgs{validators: 1, application: blocked}) + height := cs.GetCurrentHeight() + defer func() { + select { + case <-blocked.release: + default: + close(blocked.release) + } + cs.Stop() + cs.Wait() + }() + require.NoError(t, cs.Start(dash.ContextWithProTxHash(ctx, cs.privValidator.ProTxHash))) + var finalizeCtx context.Context + select { + case finalizeCtx = <-blocked.entered: + case <-time.After(5 * time.Second): + t.Fatal("consensus did not enter FinalizeBlock") + } + require.NotNil(t, finalizeCtx.Value(msgInfoCtx), "processing must retain message metadata") + cs.Stop() + waited := make(chan struct{}) + go func() { cs.Wait(); close(waited) }() + select { + case <-waited: + t.Fatal("Wait returned with FinalizeBlock still active") + case <-time.After(20 * time.Millisecond): + } + if cancelParent { + cancel() + require.ErrorIs(t, finalizeCtx.Err(), context.Canceled) + } else { + require.NoError(t, finalizeCtx.Err(), "manual Stop must preserve an admitted commit") + } + close(blocked.release) + select { + case <-waited: + case <-time.After(5 * time.Second): + t.Fatal("consensus did not drain") + } + state, err := cs.stateStore.Load() + require.NoError(t, err) + if !cancelParent { + require.Equal(t, height, state.LastBlockHeight) + } + }) + } +} + +type blockedStartTicker struct { + TimeoutTicker + entered chan context.Context + release chan struct{} +} + +func (t *blockedStartTicker) Start(ctx context.Context) error { + t.entered <- ctx + <-t.release + return t.TimeoutTicker.Start(ctx) +} + +func TestConsensusHandoffCallerCanStopWaiting(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + cs, _ := makeState(ctx, t, makeStateArgs{}) + ticker := &blockedStartTicker{TimeoutTicker: cs.timeoutTicker, entered: make(chan context.Context, 1), release: make(chan struct{})} + cs.timeoutTicker = ticker + cs.roundScheduler.timeoutTicker = ticker + rts := setup(ctx, t, 1, []*State{cs}, 32) + for _, r := range rts.reactors { + callerCtx, cancelCaller := context.WithCancel(ctx) + defer cancelCaller() + defer func() { + select { + case <-ticker.release: + default: + close(ticker.release) + } + r.Stop() + r.Wait() + }() + returned := make(chan struct{}) + go func() { r.SwitchToConsensus(callerCtx, cs.GetStateData().state, false, 0); close(returned) }() + var startupCtx context.Context + select { + case startupCtx = <-ticker.entered: + case <-time.After(5 * time.Second): + t.Fatal("handoff did not enter startup") + } + cancelCaller() + select { + case <-returned: + case <-time.After(time.Second): + t.Fatal("caller cancellation did not release handoff wait") + } + require.NoError(t, startupCtx.Err(), "caller cancellation must not abandon admitted startup") + r.Stop() + waited := make(chan struct{}) + go func() { r.Wait(); close(waited) }() + select { + case <-waited: + t.Fatal("reactor stopped before its admitted handoff finished") + case <-time.After(20 * time.Millisecond): + } + close(ticker.release) + select { + case <-waited: + case <-time.After(5 * time.Second): + t.Fatal("reactor failed to drain handoff") + } + require.False(t, cs.IsRunning()) + } +} diff --git a/internal/consensus/state.go b/internal/consensus/state.go index 9075b67336..1237ba3d8e 100644 --- a/internal/consensus/state.go +++ b/internal/consensus/state.go @@ -465,6 +465,13 @@ func (cs *State) SetPrivValidator(ctx context.Context, priv types.PrivValidator) } } +type processingContextKey struct{} + +// Start preserves the parent's lifetime for in-flight processing during manual Stop. +func (cs *State) Start(ctx context.Context) error { + return cs.BaseService.Start(context.WithValue(ctx, processingContextKey{}, ctx)) +} + // OnStart loads the latest state via the WAL, and starts the timeout and // receive routines. func (cs *State) OnStart(ctx context.Context) (err error) { @@ -679,6 +686,10 @@ func (cs *State) loadLastCommit(lastBlockHeight int64) (*types.Commit, error) { // Updates (state transitions) happen on timeouts, complete proposals, and 2/3 majorities. // State must be locked before any internal state is updated. func (cs *State) receiveRoutine(ctx context.Context, stopFn func(*State) bool) { + processCtx, ok := ctx.Value(processingContextKey{}).(context.Context) + if !ok { + processCtx = ctx + } onExit := func(cs *State) { // NOTE: the internalMsgQueue may have signed messages from our // priv_val that haven't hit the WAL, but its ok because @@ -735,6 +746,9 @@ func (cs *State) receiveRoutine(ctx context.Context, stopFn func(*State) bool) { defer func() { onExit(cs); <-queueDone }() for { + if ctx.Err() != nil { + return + } if stopFn != nil && stopFn(cs) { return } @@ -742,14 +756,14 @@ func (cs *State) receiveRoutine(ctx context.Context, stopFn func(*State) bool) { select { case <-cs.txNotifier.TxsAvailable(): stateData := cs.stateDataStore.Get() - cs.handleTxsAvailable(ctx, &stateData) + cs.handleTxsAvailable(processCtx, &stateData) err := stateData.Save() if err != nil { cs.logger.Error("failed update state-data", "err", err) } case mi := <-cs.msgInfoQueue.read(): stateData := cs.stateDataStore.Get() - err := cs.msgDispatcher.dispatch(ctx, &stateData, mi) + err := cs.msgDispatcher.dispatch(processCtx, &stateData, mi) if mi.PeerID != "" { // The peer scheduler makes room in the verification budget for a // message before handing it over, which is only sound if what it @@ -776,7 +790,7 @@ func (cs *State) receiveRoutine(ctx context.Context, stopFn func(*State) bool) { // if the timeout is relevant to the rs // go to the next step - cs.handleTimeout(ctx, ti, &stateData) + cs.handleTimeout(processCtx, ti, &stateData) err := cs.stateDataStore.Update(stateData) if err != nil { cs.logger.Error("failed update state-data", "err", err) diff --git a/internal/consensus/ticker.go b/internal/consensus/ticker.go index 95c42f83fe..aa655c9897 100644 --- a/internal/consensus/ticker.go +++ b/internal/consensus/ticker.go @@ -58,7 +58,7 @@ func (t *timeoutTicker) OnStart(ctx context.Context) error { return nil } -// OnStop implements service.Service. It stops the timeout routine. +// OnStop needs no extra signal: the service context stops timeout work. func (t *timeoutTicker) OnStop() {} // OnDrain stops the timer after its sole scheduling worker exits. diff --git a/internal/consensus/wal.go b/internal/consensus/wal.go index 31c19dfccc..9152b035cd 100644 --- a/internal/consensus/wal.go +++ b/internal/consensus/wal.go @@ -174,9 +174,7 @@ func (wal *BaseWAL) FlushAndSync() error { return wal.group.FlushAndSync() } -// Stop the underlying autofile group. -// Use Wait() to ensure it's finished shutting down -// before cleaning up files. +// OnStop stops periodic flush scheduling; Wait joins the final flush and close. func (wal *BaseWAL) OnStop() { wal.flushTicker.Stop() } diff --git a/internal/libs/autofile/cmd/logjack.go b/internal/libs/autofile/cmd/logjack.go index d1bacbbb71..082aecb12f 100644 --- a/internal/libs/autofile/cmd/logjack.go +++ b/internal/libs/autofile/cmd/logjack.go @@ -64,6 +64,8 @@ func main() { stdlog.Fatalf("logjack couldn't create output file %q", headPath) } + defer group.Close() + if err = group.Start(ctx); err != nil { stdlog.Fatalf("logjack couldn't start with file %q", headPath) } diff --git a/internal/libs/autofile/group.go b/internal/libs/autofile/group.go index fc0c484040..220085e6d5 100644 --- a/internal/libs/autofile/group.go +++ b/internal/libs/autofile/group.go @@ -73,7 +73,8 @@ type Group struct { } // OpenGroup creates a new Group with head at headPath. It returns an error if -// it fails to open head file. +// it fails to open head file. The caller must Close the group, even if it was +// never started; context cancellation alone does not close its head. func OpenGroup(ctx context.Context, logger log.Logger, headPath string, groupOptions ...func(*Group)) (*Group, error) { dir, err := filepath.Abs(filepath.Dir(headPath)) if err != nil { diff --git a/internal/p2p/router.go b/internal/p2p/router.go index 375eecab50..8dc244ba7a 100644 --- a/internal/p2p/router.go +++ b/internal/p2p/router.go @@ -722,7 +722,7 @@ func (r *Router) routePeer(ctx context.Context, peerID types.NodeID, conn Connec connID := r.peerManager.Ready(ctx, peerID, channels) // we use context to manage the lifecycle of the peer - // note that original ctx will be used in cleanup + // Peer-manager cleanup uses ctx; shutdown notifications are best effort. ioCtx, cancel := context.WithCancel(ctx) defer cancel()