test branch

headertest/store.go

@@ -20,7 +20,7 @@ type Store[H header.Header[H]] struct {
 	TailHeight uint64
 
 	onDeleteMu sync.Mutex
-	onDelete   []func(context.Context, []H) error
+	onDelete   []func(context.Context, uint64) error
 }
 
 // NewDummyStore creates a store for DummyHeader.
@@ -80,26 +80,26 @@ func (m *Store[H]) GetByHeight(_ context.Context, height uint64) (H, error) {
 }
 
 func (m *Store[H]) DeleteTo(ctx context.Context, to uint64) error {
-	var deleted []H
 	for h := m.TailHeight; h < to; h++ {
-		hdr, ok := m.Headers[h]
-		if ok {
-			delete(m.Headers, h)
-			deleted = append(deleted, hdr)
+		_, ok := m.Headers[h]
+		if !ok {
+			continue
 		}
-	}
 
-	m.TailHeight = to
-	for _, deleteFn := range m.onDelete {
-		err := deleteFn(ctx, deleted)
-		if err != nil {
-			return err
+		for _, deleteFn := range m.onDelete {
+			err := deleteFn(ctx, h)
+			if err != nil {
+				return err
+			}
 		}
+		delete(m.Headers, h) // must be after deleteFn
 	}
+
+	m.TailHeight = to
 	return nil
 }
 
-func (m *Store[H]) OnDelete(fn func(context.Context, []H) error) {
+func (m *Store[H]) OnDelete(fn func(context.Context, uint64) error) {
 	m.onDeleteMu.Lock()
 	defer m.onDeleteMu.Unlock()
 

interface.go

@@ -88,8 +88,10 @@ type Store[H Header[H]] interface {
 	// DeleteTo deletes the range [Tail():to).
 	DeleteTo(ctx context.Context, to uint64) error
 
-	// OnDelete registers given handler to be called whenever headers are removed from the Store.
-	OnDelete(func(context.Context, []H) error)
+	// OnDelete registers given handler to be called whenever a header with the height is being removed.
+	// OnDelete guarantees that the header is accessible for the handler with GetByHeight and is removed
+	// only after the handler terminates with nil error.
+	OnDelete(handler func(ctx context.Context, height uint64) error)
 }
 
 // Getter contains the behavior necessary for a component to retrieve

p2p/server_test.go

@@ -197,4 +197,4 @@ func (timeoutStore[H]) DeleteTo(ctx context.Context, _ uint64) error {
 	return ctx.Err()
 }
 
-func (timeoutStore[H]) OnDelete(fn func(context.Context, []H) error) {}
+func (timeoutStore[H]) OnDelete(fn func(context.Context, uint64) error) {}

p2p/subscriber.go

@@ -5,6 +5,7 @@
 	"errors"
 	"fmt"
 	"sync"
+	"time"
 
 	pubsub "github.com/libp2p/go-libp2p-pubsub"
 	"github.com/libp2p/go-libp2p/core/peer"
@@ -85,7 +86,7 @@
 // be called separately to ensure a validator is mounted on the topic.
 func (s *Subscriber[H]) Start(context.Context) (err error) {
 	log.Debugw("joining topic", "topic ID", s.pubsubTopicID)
-	err = s.pubsub.RegisterTopicValidator(s.pubsubTopicID, s.verifyMessage)
+	err = s.pubsub.RegisterTopicValidator(s.pubsubTopicID, s.verifyMessage, pubsub.WithValidatorTimeout(time.Minute))
 	if err != nil {
 		return err
 	}

store/height_indexer.go

@@ -34,7 +34,11 @@ func newHeightIndexer[H header.Header[H]](
 }
 
 // HashByHeight loads a header hash corresponding to the given height.
-func (hi *heightIndexer[H]) HashByHeight(ctx context.Context, h uint64) (header.Hash, error) {
+func (hi *heightIndexer[H]) HashByHeight(
+	ctx context.Context,
+	h uint64,
+	cache bool,
+) (header.Hash, error) {
 	if v, ok := hi.cache.Get(h); ok {
 		return v, nil
 	}
@@ -44,6 +48,8 @@ func (hi *heightIndexer[H]) HashByHeight(ctx context.Context, h uint64) (header.
 		return nil, err
 	}
 
-	hi.cache.Add(h, header.Hash(val))
+	if cache {
+		hi.cache.Add(h, header.Hash(val))
+	}
 	return val, nil
 }