From 0339f6df93b2dfa289d16d6f3d845467953a404f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Mon, 17 Aug 2026 11:22:10 -0400 Subject: [PATCH] hub: count blocks, not partials, toward the unlinkable limit WithMaxConsecutiveUnlinkableBlocks names a number of blocks, but the counter was advanced by every message. A flash-block chain delivers one block as several partials that all fail the same link check, so a hub configured for 5 gave up after 2 blocks. Intermediate partials no longer count; a plain block and a block's final partial do. --- CHANGELOG.md | 2 ++ hub/hub.go | 16 +++++++-- hub/hub_test.go | 95 +++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 111 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a5e3b27..f783232 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/hub/hub.go b/hub/hub.go index 9f97fe0..e3ce633 100644 --- a/hub/hub.go +++ b/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), diff --git a/hub/hub_test.go b/hub/hub_test.go index b826fe6..57697a5 100644 --- a/hub/hub_test.go +++ b/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) + }) +}