fix(store): stop long receding/advancing on Stop (#326)

Ensures Store stops receding/advancing if the take long time.
Apparently, badger, pebble, and any other major datastore
implementations do not handle contexts, so we need to do it ourselves.
Also adds better debug logging coverage

Author: Wondertan

go.mod

@@ -17,6 +17,7 @@ require (
 	go.opentelemetry.io/otel v1.36.0
 	go.opentelemetry.io/otel/metric v1.36.0
 	go.opentelemetry.io/otel/trace v1.36.0
+	go.uber.org/zap v1.27.0
 	golang.org/x/crypto v0.39.0
 )
 
@@ -101,7 +102,6 @@ require (
 	go.uber.org/automaxprocs v1.6.0 // indirect
 	go.uber.org/mock v0.5.2 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/exp v0.0.0-20250606033433-dcc06ee1d476 // indirect
 	golang.org/x/mod v0.25.0 // indirect
 	golang.org/x/net v0.41.0 // indirect

store/store.go

@@ -12,6 +12,7 @@ import (
 	"github.com/ipfs/go-datastore"
 	"github.com/ipfs/go-datastore/namespace"
 	logging "github.com/ipfs/go-log/v2"
+	"go.uber.org/zap/zapcore"
 
 	"github.com/celestiaorg/go-header"
 )
@@ -58,6 +59,7 @@ type Store[H header.Header[H]] struct {
 	pending *batch[H]
 	// syncCh is a channel used to synchronize writes
 	syncCh chan chan struct{}
+	cancel context.CancelFunc
 
 	onDeleteMu sync.Mutex
 	onDelete   []func(context.Context, uint64) error
@@ -132,7 +134,9 @@ func (s *Store[H]) Start(ctx context.Context) error {
 		return err
 	}
 
-	go s.flushLoop()
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancel = cancel
+	go s.flushLoop(ctx)
 	return nil
 }
 
@@ -430,9 +434,8 @@ func (s *Store[H]) Append(ctx context.Context, headers ...H) error {
 // This way writes are controlled and manageable from one place allowing
 // (1) Appends not to be blocked on long disk IO writes and underlying DB compactions
 // (2) Batching header writes
-func (s *Store[H]) flushLoop() {
+func (s *Store[H]) flushLoop(ctx context.Context) {
 	defer close(s.writesDn)
-	ctx := context.Background()
 
 	flush := func(headers []H) {
 		s.ensureInit(headers)
@@ -608,37 +611,72 @@ func (s *Store[H]) recedeTail(ctx context.Context) {
 func (s *Store[H]) nextHead(ctx context.Context) (head H, changed bool) {
 	head, err := s.Head(ctx)
 	if err != nil {
-		log.Errorw("cannot load head", "err", err)
+		log.Errorw("cannot load head while advancing", "err", err)
 		return head, false
 	}
 
-	for {
+	for ctx.Err() == nil {
 		h, err := s.getByHeight(ctx, head.Height()+1)
 		if err != nil {
+			log.Debugw("next head error", "current", head.Height(), "err", err)
 			return head, changed
 		}
+
+		if !changed && log.Level() == zapcore.DebugLevel {
+			now := time.Now()
+			log.Debugw("advancing head", "start_height", head.Height())
+			defer func() {
+				log.Debugw(
+					"finished advancing head",
+					"end_height",
+					head.Height(),
+					"took(s)",
+					time.Since(now),
+				)
+			}()
+		}
+
 		head = h
 		changed = true
 	}
+
+	return head, changed
 }
 
 // nextTail finds the new contiguous Tail by iterating the current Tail down until the older height Tail is found.
 // Returns true if the older one was found.
 func (s *Store[H]) nextTail(ctx context.Context) (tail H, changed bool) {
 	tail, err := s.Tail(ctx)
 	if err != nil {
-		log.Errorw("cannot load tail", "err", err)
+		log.Errorw("cannot load tail while receding", "err", err)
 		return tail, false
 	}
 
-	for {
+	for ctx.Err() == nil {
 		h, err := s.getByHeight(ctx, tail.Height()-1)
 		if err != nil {
 			return tail, changed
 		}
+
+		if !changed && log.Level() == zapcore.DebugLevel {
+			now := time.Now()
+			log.Debugw("receding tail", "start_height", tail.Height())
+			defer func() {
+				log.Debugw(
+					"finished receding tail",
+					"end_height",
+					tail.Height(),
+					"took(s)",
+					time.Since(now),
+				)
+			}()
+		}
+
 		tail = h
 		changed = true
 	}
+
+	return tail, changed
 }
 
 func (s *Store[H]) loadHeadAndTail(ctx context.Context) error {