hub: count blocks, not partials, toward the unlinkable limit

CHANGELOG.md

@@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- `hub.WithMaxConsecutiveUnlinkableBlocks` now counts blocks rather than messages: an intermediate flash block — `PartialIndex != 0` without `LastPartial` — no longer advances the counter. Every partial of a block fails the same link check, so on a chain delivering four per block the hub gave up after a quarter of the blocks the limit names. A plain block and a block's final partial still count, and any linkable block still resets the count.
+
 - Hub subscriptions with `with_partials=false` no longer stall on flash/partial-block chains. Previously every block with `PartialIndex != 0` (including the closing `LastPartial`) was dropped, so a no-partial subscriber only advanced on separate `PartialIndex==0` full blocks, which can lag the sealed head by tens of seconds. The subscription now drops only intermediate partials and delivers each `LastPartial` as a full block (partial markers cleared on a thin copy that shares the payload; the shared original block is never mutated).
 - `FileSource` with `FileSourceErrorOnMissingMergedBlocksFile` no longer truncates its output: on a missing file it now drains every already-read block through the ordered stream before surfacing the error, instead of calling `Shutdown()` immediately (which aborted in-flight reader goroutines and discarded blocks).
 

hub/hub.go

@@ -364,8 +364,15 @@ func (h *ForkableHub) Run() {
 	liveSource.Run()
 
 }
+
+// isNonFinalPartial reports whether the block is a flash block that is not the last one of
+// its block number, i.e. a preview of a block that has not been completed yet.
+func isNonFinalPartial(blk *pbbstream.Block) bool {
+	return blk.PartialIndex != 0 && !blk.LastPartial
+}
+
 func (h *ForkableHub) ProcessBlock(blk *pbbstream.Block, obj any) error {
-	if !h.IsReady() && blk.PartialIndex != 0 && !blk.LastPartial {
+	if !h.IsReady() && isNonFinalPartial(blk) {
 		return nil // we don't get ready with partial blocks...
 	}
 
@@ -388,7 +395,12 @@ func (h *ForkableHub) ProcessBlock(blk *pbbstream.Block, obj any) error {
 	}
 
 	if !h.forkable.Linkable(blk) {
-		if h.maxConsecutiveUnlinkableBlocks != 0 {
+		// A non-final flash block is not counted. Every partial of a block fails this same
+		// check for the same reason, so on a chain delivering four of them per block the
+		// limit would be reached after a quarter of the blocks it names — and the check is
+		// there to catch a gap the one-block store can no longer bridge, which the block's
+		// final partial reports just as well. A real block and a final flash block count.
+		if h.maxConsecutiveUnlinkableBlocks != 0 && !isNonFinalPartial(blk) {
 			h.consecutiveUnlinkableBlocks++
 			h.logger.Warn("block not linkable after one-block lookup",
 				zap.Uint64("block_num", blk.Number),

hub/hub_test.go

@@ -980,3 +980,98 @@ func TestForkableHub_SourceThroughCursor(t *testing.T) {
 		})
 	}
 }
+
+func TestForkableHub_ProcessBlock_UnlinkableCountSkipsNonFinalFlashBlocks(t *testing.T) {
+	// The limit is expressed in blocks, so it has to be reached after that many blocks fail
+	// to link. A flash-block chain delivers each block as several messages that all fail the
+	// same check, and counting every one of them would restart the hub after a quarter of the
+	// blocks the limit names — while the gap it exists to catch is reported just as well by
+	// each block's final message.
+	newReadyHub := func(t *testing.T) *ForkableHub {
+		t.Helper()
+
+		lsf := bstream.NewTestSourceFactory()
+		oneBlockStore := dstore.NewMockStore(nil)
+
+		fh := NewForkableHubWithOptions(lsf.NewSource, 0, oneBlockStore, []Option{
+			WithMaxConsecutiveUnlinkableBlocks(3),
+		})
+
+		AddToMockStore(t, oneBlockStore,
+			bstream.TestBlockWithLIBNum("00000003", "00000002", 2),
+			bstream.TestBlockWithLIBNum("00000004", "00000003", 2),
+			bstream.TestBlockWithLIBNum("00000005", "00000004", 2),
+			bstream.TestBlockWithLIBNum("00000008", "00000005", 3),
+			bstream.TestBlockWithLIBNum("00000009", "00000008", 3),
+		)
+		require.NoError(t, fh.bootstrap())
+		require.Equal(t, uint64(3), fh.forkable.LowestBlockNum())
+
+		// The gap the counter is about is one the one-block store cannot bridge, which is what
+		// an empty walk stands for. Run() is what closes Ready on a real hub, and the check
+		// only applies past readiness.
+		oneBlockStore.WalkFunc = func(ctx context.Context, prefix string, f func(filename string) error) error {
+			return nil
+		}
+		close(fh.Ready)
+
+		return fh
+	}
+
+	// unlinkable builds a block whose parent the forkable has never seen.
+	unlinkable := func(num uint64, partialIndex int32, lastPartial bool) *pbbstream.Block {
+		blk := bstream.TestBlockWithLIBNum(fmt.Sprintf("%08d", num), fmt.Sprintf("%08d", num-1), 3)
+		blk.PartialIndex = partialIndex
+		blk.LastPartial = lastPartial
+
+		return blk
+	}
+
+	t.Run("non-final flash blocks never trip it", func(t *testing.T) {
+		fh := newReadyHub(t)
+
+		for num := uint64(20); num < 30; num++ {
+			for idx := int32(1); idx <= 3; idx++ {
+				require.NoError(t, fh.ProcessBlock(unlinkable(num, idx, false), nil))
+			}
+		}
+
+		assert.Equal(t, 0, fh.consecutiveUnlinkableBlocks)
+	})
+
+	t.Run("final flash blocks trip it, partials in between do not", func(t *testing.T) {
+		fh := newReadyHub(t)
+
+		var err error
+		for num := uint64(20); err == nil && num < 30; num++ {
+			for idx := int32(1); idx <= 3; idx++ {
+				require.NoError(t, fh.ProcessBlock(unlinkable(num, idx, false), nil))
+			}
+			err = fh.ProcessBlock(unlinkable(num, 4, true), nil)
+		}
+
+		require.ErrorIs(t, err, errRestartRequired)
+		assert.Equal(t, 3, fh.consecutiveUnlinkableBlocks, "one count per block, not per message")
+	})
+
+	t.Run("plain blocks trip it", func(t *testing.T) {
+		fh := newReadyHub(t)
+
+		require.NoError(t, fh.ProcessBlock(unlinkable(20, 0, false), nil))
+		require.NoError(t, fh.ProcessBlock(unlinkable(21, 0, false), nil))
+		require.ErrorIs(t, fh.ProcessBlock(unlinkable(22, 0, false), nil), errRestartRequired)
+	})
+
+	t.Run("a linkable block resets the count", func(t *testing.T) {
+		fh := newReadyHub(t)
+
+		require.NoError(t, fh.ProcessBlock(unlinkable(20, 0, false), nil))
+		require.NoError(t, fh.ProcessBlock(unlinkable(21, 0, false), nil))
+		require.NoError(t, fh.ProcessBlock(bstream.TestBlockWithLIBNum("00000010", "00000009", 3), nil))
+		require.Equal(t, 0, fh.consecutiveUnlinkableBlocks)
+
+		require.NoError(t, fh.ProcessBlock(unlinkable(22, 0, false), nil))
+		require.NoError(t, fh.ProcessBlock(unlinkable(23, 0, false), nil))
+		require.ErrorIs(t, fh.ProcessBlock(unlinkable(24, 0, false), nil), errRestartRequired)
+	})
+}