fix(local pub sub): fix must subscribe to handle context cancelled

block/block.go

@@ -121,12 +121,12 @@ func (m *Manager) attemptApplyCachedBlocks() error {
  for {
  expectedHeight := m.State.NextHeight()
 
- cachedBlock, blockExists := m.blockCache.GetBlockFromCache(expectedHeight)
+ cachedBlock, blockExists := m.blockCache.Get(expectedHeight)
  if !blockExists {
  break
  }
  if err := m.validateBlock(cachedBlock.Block, cachedBlock.Commit); err != nil {
- m.blockCache.DeleteBlockFromCache(cachedBlock.Block.Header.Height)
+ m.blockCache.Delete(cachedBlock.Block.Header.Height)
  // TODO: can we take an action here such as dropping the peer / reducing their reputation?
  return fmt.Errorf("block not valid at height %d, dropping it: err:%w", cachedBlock.Block.Header.Height, err)
  }
@@ -137,7 +137,7 @@ func (m *Manager) attemptApplyCachedBlocks() error {
  }
  m.logger.Info("Block applied", "height", expectedHeight)
 
- m.blockCache.DeleteBlockFromCache(cachedBlock.Block.Header.Height)
+ m.blockCache.Delete(cachedBlock.Block.Header.Height)
  }
 
  return nil

block/block_cache.go

@@ -9,23 +9,23 @@ type Cache struct {
  cache map[uint64]types.CachedBlock
 }
 
-func (m *Cache) AddBlockToCache(h uint64, b *types.Block, c *types.Commit, source types.BlockSource) {
+func (m *Cache) Add(h uint64, b *types.Block, c *types.Commit, source types.BlockSource) {
  m.cache[h] = types.CachedBlock{Block: b, Commit: c, Source: source}
  types.BlockCacheSizeGauge.Set(float64(m.Size()))
 }
 
-func (m *Cache) DeleteBlockFromCache(h uint64) {
+func (m *Cache) Delete(h uint64) {
  delete(m.cache, h)
  types.BlockCacheSizeGauge.Set(float64(m.Size()))
 }
 
-func (m *Cache) GetBlockFromCache(h uint64) (types.CachedBlock, bool) {
+func (m *Cache) Get(h uint64) (types.CachedBlock, bool) {
  ret, found := m.cache[h]
  return ret, found
 }
 
-func (m *Cache) HasBlockInCache(h uint64) bool {
- _, found := m.GetBlockFromCache(h)
+func (m *Cache) Has(h uint64) bool {
+ _, found := m.Get(h)
  return found
 }
 

block/p2p.go

@@ -39,7 +39,7 @@ func (m *Manager) onReceivedBlock(event pubsub.Message) {
  m.retrieverMu.Lock() // needed to protect blockCache access
 
  // It is not strictly necessary to return early, for correctness, but doing so helps us avoid mutex pressure and unnecessary repeated attempts to apply cached blocks
- if m.blockCache.HasBlockInCache(height) {
+ if m.blockCache.Has(height) {
  m.retrieverMu.Unlock()
  return
  }
@@ -51,7 +51,7 @@ func (m *Manager) onReceivedBlock(event pubsub.Message) {
 
  nextHeight := m.State.NextHeight()
  if height >= nextHeight {
- m.blockCache.AddBlockToCache(height, &block, &commit, source)
+ m.blockCache.Add(height, &block, &commit, source)
  }
  m.retrieverMu.Unlock() // have to give this up as it's locked again in attempt apply, and we're not re-entrant
 

block/retriever.go

@@ -136,7 +136,7 @@ func (m *Manager) ProcessNextDABatch(daMetaData *da.DASubmitMetaData) error {
 
  lastAppliedHeight = float64(block.Header.Height)
 
- m.blockCache.DeleteBlockFromCache(block.Header.Height)
+ m.blockCache.Delete(block.Header.Height)
  }
  }
  types.LastReceivedDAHeightGauge.Set(lastAppliedHeight)

block/submit.go

@@ -176,14 +176,14 @@ func (m *Manager) CreateBatch(maxBatchSize uint64, startHeight uint64, endHeight
  Commits: make([]*types.Commit, 0, batchSize),
  }
 
- for height := startHeight; height <= endHeightInclusive; height++ {
- block, err := m.Store.LoadBlock(height)
+ for h := startHeight; h <= endHeightInclusive; h++ {
+ block, err := m.Store.LoadBlock(h)
  if err != nil {
- return nil, fmt.Errorf("load block: height: %d: %w", height, err)
+ return nil, fmt.Errorf("load block: h: %d: %w", h, err)
  }
- commit, err := m.Store.LoadCommit(height)
+ commit, err := m.Store.LoadCommit(h)
  if err != nil {
- return nil, fmt.Errorf("load commit: height: %d: %w", height, err)
+ return nil, fmt.Errorf("load commit: h: %d: %w", h, err)
  }
 
  batch.Blocks = append(batch.Blocks, block)
@@ -196,8 +196,8 @@ func (m *Manager) CreateBatch(maxBatchSize uint64, startHeight uint64, endHeight
  batch.Blocks = batch.Blocks[:len(batch.Blocks)-1]
  batch.Commits = batch.Commits[:len(batch.Commits)-1]
 
- if height == startHeight {
- return nil, fmt.Errorf("block size exceeds max batch size: height %d: size: %d: %w", height, totalSize, gerrc.ErrOutOfRange)
+ if h == startHeight {
+ return nil, fmt.Errorf("block size exceeds max batch size: h %d: size: %d: %w", h, totalSize, gerrc.ErrOutOfRange)
  }
  break
  }

utils/event/funcs.go

@@ -24,9 +24,13 @@ func MustSubscribe(
  logger types.Logger,
 ) {
  subscription, err := pubsubServer.SubscribeUnbuffered(ctx, clientID, eventQuery)
- if err != nil && !errors.Is(err, context.Canceled) {
- logger.Error("subscribe to events")
- panic(err)
+ if err != nil {
+ err = fmt.Errorf("subscribe unbuffered: %w", err)
+ if !errors.Is(err, context.Canceled) {
+ logger.Error("Must subscribe.", "err", err)
+ panic(err)
+ }
+ return
  }
 
  for {