From 2817e55eb21f5486d0299258b964a7f4275b4c56 Mon Sep 17 00:00:00 2001 From: questfever Date: Thu, 13 Aug 2026 15:06:04 +0800 Subject: [PATCH] fix(da): serialize subscriber lifecycle transitions --- block/internal/da/subscriber.go | 45 ++++++-- block/internal/da/subscriber_test.go | 152 +++++++++++++++++++++++++++ 2 files changed, 188 insertions(+), 9 deletions(-) diff --git a/block/internal/da/subscriber.go b/block/internal/da/subscriber.go index cf3dd49084..9c3c28bc57 100644 --- a/block/internal/da/subscriber.go +++ b/block/internal/da/subscriber.go @@ -41,6 +41,14 @@ type SubscriberConfig struct { StartHeight uint64 // initial localDAHeight } +type subscriberLifecycleState uint8 + +const ( + subscriberStopped subscriberLifecycleState = iota + subscriberRunning + subscriberStopping +) + // Subscriber is a shared DA subscription primitive that encapsulates the // follow/catchup lifecycle. It subscribes to one or more DA namespaces, // tracks the highest seen DA height, and drives sequential catchup via @@ -81,7 +89,9 @@ type Subscriber struct { // lifecycle lifecycleMu sync.Mutex + state subscriberLifecycleState cancel context.CancelFunc + stopDone chan struct{} wg sync.WaitGroup } @@ -115,16 +125,16 @@ func (s *Subscriber) Start(ctx context.Context) error { } s.lifecycleMu.Lock() - if s.cancel != nil { - s.lifecycleMu.Unlock() + defer s.lifecycleMu.Unlock() + if s.state != subscriberStopped { return nil } ctx, cancel := context.WithCancel(ctx) - s.cancel = cancel - s.lifecycleMu.Unlock() - s.wg.Add(2) + s.state = subscriberRunning + s.cancel = cancel + s.stopDone = make(chan struct{}) if s.client.SupportsSubscribe() { go s.followLoop(ctx) } else { @@ -138,14 +148,31 @@ func (s *Subscriber) Start(ctx context.Context) error { // Stop gracefully stops the background goroutines. func (s *Subscriber) Stop() { s.lifecycleMu.Lock() + switch s.state { + case subscriberStopped: + s.lifecycleMu.Unlock() + return + case subscriberStopping: + stopDone := s.stopDone + s.lifecycleMu.Unlock() + <-stopDone + return + } + + s.state = subscriberStopping cancel := s.cancel - s.cancel = nil + stopDone := s.stopDone s.lifecycleMu.Unlock() - if cancel != nil { - cancel() - } + cancel() s.wg.Wait() + + s.lifecycleMu.Lock() + s.state = subscriberStopped + s.cancel = nil + s.stopDone = nil + close(stopDone) + s.lifecycleMu.Unlock() } // LocalDAHeight returns the current local DA height. diff --git a/block/internal/da/subscriber_test.go b/block/internal/da/subscriber_test.go index 2ed80886de..c396091edd 100644 --- a/block/internal/da/subscriber_test.go +++ b/block/internal/da/subscriber_test.go @@ -3,6 +3,8 @@ package da import ( "context" "errors" + "sync" + "sync/atomic" "testing" "time" @@ -19,6 +21,68 @@ type MockSubscriberHandler struct { mock.Mock } +type lifecycleTestClient struct { + Client + + subscribeCalls atomic.Int32 + entered [2]chan struct{} + canceled [2]chan struct{} + release [2]chan struct{} + releaseOnce [2]sync.Once +} + +func newLifecycleTestClient() *lifecycleTestClient { + client := &lifecycleTestClient{} + for i := range 2 { + client.entered[i] = make(chan struct{}) + client.canceled[i] = make(chan struct{}) + client.release[i] = make(chan struct{}) + } + return client +} + +func (c *lifecycleTestClient) SupportsSubscribe() bool { + return true +} + +func (c *lifecycleTestClient) Subscribe( + ctx context.Context, + _ []byte, + _ bool, +) (<-chan datypes.SubscriptionEvent, error) { + generation := int(c.subscribeCalls.Add(1) - 1) + close(c.entered[generation]) + <-ctx.Done() + close(c.canceled[generation]) + <-c.release[generation] + return nil, ctx.Err() +} + +func (c *lifecycleTestClient) releaseGeneration(generation int) { + c.releaseOnce[generation].Do(func() { + close(c.release[generation]) + }) +} + +type lifecycleTestHandler struct{} + +func (lifecycleTestHandler) HandleEvent(context.Context, datypes.SubscriptionEvent, bool) error { + return nil +} + +func (lifecycleTestHandler) HandleCatchup(context.Context, uint64) error { + return nil +} + +func waitForLifecycleSignal(t *testing.T, signal <-chan struct{}, description string) { + t.Helper() + select { + case <-signal: + case <-time.After(time.Second): + t.Fatalf("timed out waiting for %s", description) + } +} + func (m *MockSubscriberHandler) HandleEvent(ctx context.Context, ev datypes.SubscriptionEvent, isInline bool) error { args := m.Called(ctx, ev, isInline) return args.Error(0) @@ -29,6 +93,94 @@ func (m *MockSubscriberHandler) HandleCatchup(ctx context.Context, height uint64 return args.Error(0) } +func TestSubscriber_LifecycleSerializesStartAndStop(t *testing.T) { + client := newLifecycleTestClient() + t.Cleanup(func() { + client.releaseGeneration(0) + client.releaseGeneration(1) + }) + + sub := NewSubscriber(SubscriberConfig{ + Client: client, + Logger: zerolog.Nop(), + Handler: lifecycleTestHandler{}, + Namespaces: [][]byte{[]byte("ns")}, + DABlockTime: time.Hour, + }) + + if err := sub.Start(t.Context()); err != nil { + t.Fatalf("start first generation: %v", err) + } + waitForLifecycleSignal(t, client.entered[0], "first generation to start") + + stopDone := make(chan struct{}) + go func() { + sub.Stop() + close(stopDone) + }() + waitForLifecycleSignal(t, client.canceled[0], "first generation cancellation") + + restartDone := make(chan error, 1) + go func() { + restartDone <- sub.Start(t.Context()) + }() + select { + case err := <-restartDone: + if err != nil { + t.Fatalf("start while stopping: %v", err) + } + case <-time.After(time.Second): + t.Fatal("Start did not return while the previous generation was stopping") + } + select { + case <-client.entered[1]: + t.Fatal("Start launched a second generation while the first generation was stopping") + default: + } + select { + case <-stopDone: + t.Fatal("Stop returned before the first generation exited") + default: + } + + concurrentStopDone := make(chan struct{}) + go func() { + sub.Stop() + close(concurrentStopDone) + }() + select { + case <-concurrentStopDone: + t.Fatal("concurrent Stop returned before the first generation exited") + default: + } + + client.releaseGeneration(0) + waitForLifecycleSignal(t, stopDone, "first Stop to return") + waitForLifecycleSignal(t, concurrentStopDone, "concurrent Stop to return") + + if err := sub.Start(t.Context()); err != nil { + t.Fatalf("restart subscriber: %v", err) + } + waitForLifecycleSignal(t, client.entered[1], "second generation to start") + + secondStopDone := make(chan struct{}) + go func() { + sub.Stop() + close(secondStopDone) + }() + waitForLifecycleSignal(t, client.canceled[1], "second generation cancellation") + select { + case <-secondStopDone: + t.Fatal("Stop returned before the second generation exited") + default: + } + client.releaseGeneration(1) + waitForLifecycleSignal(t, secondStopDone, "second Stop to return") + + // Stopping an already stopped subscriber remains safe. + sub.Stop() +} + func TestSubscriber_RunCatchup(t *testing.T) { t.Run("success_sequence", func(t *testing.T) { ctx, cancel := context.WithCancel(t.Context())