From f84b3e71d1472b9de18864acdf7ec33adf160615 Mon Sep 17 00:00:00 2001 From: Matthew Date: Sat, 24 Oct 2026 17:28:01 -0400 Subject: [PATCH] hub: link a live block from one-block files up to that block only linkLiveUsingOneBlocks walked the one-block store from the LIB to its end and decoded every unknown file into the forkable. Files above the live block can't help link it: when a reader restarts below blocks already in the store (for example blocks stranded above a gap the merger can't pass), the hub decoded all of them, payloads included. On a chain with blocks of up to ~190 MB, with a few thousand one-block files stranded above a gap, this took the process from a completed bootstrap to OOM within about 90 seconds of its first live block. The walk now stops at the live block's number. The live block's LIB and every block between it and the live block are still read, so the reconnection check (blockFromFile.Number == blk.LibNum) is unchanged. Co-Authored-By: Claude Opus 5.5 (1M context) --- hub/hub.go | 30 ++++++++++++++++++++++- hub/hub_test.go | 63 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 1 deletion(-) diff --git a/hub/hub.go b/hub/hub.go index 730b62d..e22ffb8 100644 --- a/hub/hub.go +++ b/hub/hub.go @@ -19,6 +19,7 @@ import ( "errors" "fmt" "io" + "strconv" "strings" "time" @@ -534,7 +535,10 @@ func (h *ForkableHub) linkLiveUsingOneBlocks(ctx context.Context, blk *pbbstream h.logger.Info("linking live block using one blocks", zap.Uint64("processed_block", blk.Number), zap.Uint64("last_know_lib", lastKnownLib)) - sortedOneBlocksFiles, err := h.WalkOneBlocksStoreFrom(ctx, lastKnownLib) + // Only blocks up to the live block can link it. Files above it (a reader that got + // ahead before a restart, blocks stranded above a gap) are left for later, otherwise + // they would all be decoded and kept in the forkable for this one lookup. + sortedOneBlocksFiles, err := h.WalkOneBlocksStoreRange(ctx, lastKnownLib, blk.Number) if err != nil { return fmt.Errorf("walking through one blocks files: %w", err) } @@ -656,6 +660,30 @@ func (h *ForkableHub) WalkOneBlocksStore(ctx context.Context) ([]string, error) }) return sortedOneBlocksFiles, err } + +// WalkOneBlocksStoreRange lists the one-block files from startingBlock to stopBlock, +// inclusively, in order. +func (h *ForkableHub) WalkOneBlocksStoreRange(ctx context.Context, startingBlock, stopBlock uint64) ([]string, error) { + startingPoint := fmt.Sprintf("%010d", startingBlock) + sortedOneBlocksFiles := make([]string, 0) + err := h.oneBlocksStore.WalkFrom( + ctx, + "", + startingPoint, + func(filename string) error { + num, err := strconv.ParseUint(strings.SplitN(filename, "-", 2)[0], 10, 64) + if err != nil { + return fmt.Errorf("parsing block number of one-block file %q: %w", filename, err) + } + if num > stopBlock { + return dstore.StopIteration + } + sortedOneBlocksFiles = append(sortedOneBlocksFiles, filename) + return nil + }) + return sortedOneBlocksFiles, err +} + func (h *ForkableHub) WalkOneBlocksStoreFrom(ctx context.Context, startingBlock uint64) ([]string, error) { startingPoint := fmt.Sprintf("%010d", startingBlock) sortedOneBlocksFiles := make([]string, 0) diff --git a/hub/hub_test.go b/hub/hub_test.go index 6517192..09bd58f 100644 --- a/hub/hub_test.go +++ b/hub/hub_test.go @@ -5,6 +5,9 @@ import ( "context" "errors" "fmt" + "io" + "strconv" + "strings" "testing" "time" @@ -346,6 +349,66 @@ func TestForkableHub_ProcessBlock_OutOfOrder(t *testing.T) { "block 6 should be the forkable head: all previous blocks were delivered, just out of order") } +func TestForkableHub_LinkLiveUsingOneBlocks_StopsAtLiveBlock(t *testing.T) { + // A hub without LIB (its bootstrap failed) links a live block from the one-block + // store, from the block's LIB. Blocks above the live block can't help: here they are + // stranded above a gap (7 to 19 missing), as when a reader was restarted from an + // earlier block. They must not be decoded into the forkable. + + lsf := bstream.NewTestSourceFactory() + testOneBlockStore := dstore.NewMockStore(nil) + + fh := NewForkableHub(lsf.NewSource, 0, testOneBlockStore) + + var stored []*pbbstream.Block + for num := uint64(1); num <= 6; num++ { + stored = append(stored, bstream.TestBlockWithLIBNum(fmt.Sprintf("%08x", num), fmt.Sprintf("%08x", num-1), num-1)) + } + for num := uint64(20); num <= 30; num++ { + stored = append(stored, bstream.TestBlockWithLIBNum(fmt.Sprintf("%08x", num), fmt.Sprintf("%08x", num-1), num-1)) + } + AddToMockStore(t, testOneBlockStore, stored...) + + var opened []string + testOneBlockStore.OpenObjectFunc = func(ctx context.Context, name string) (io.ReadCloser, error) { + opened = append(opened, name) + return io.NopCloser(bytes.NewReader(testOneBlockStore.Files[name])), nil + } + + require.False(t, fh.forkable.ForkDBHasLib()) + require.NoError(t, fh.ProcessBlock(bstream.TestBlockWithLIBNum("00000005", "00000004", 4), nil)) + + for _, name := range opened { + num, err := strconv.ParseUint(name[:10], 10, 64) + require.NoError(t, err) + assert.LessOrEqual(t, num, uint64(5), "decoded one-block file %s, above the live block", name) + } + assert.NotEmpty(t, opened, "the blocks from the LIB to the live block are read") + assert.Empty(t, fh.forkable.AllBlocksAt(25), "blocks above the live block stay out of the forkable") +} + +func TestForkableHub_WalkOneBlocksStoreRange(t *testing.T) { + testOneBlockStore := dstore.NewMockStore(nil) + fh := NewForkableHub(bstream.NewTestSourceFactory().NewSource, 0, testOneBlockStore) + + AddToMockStore(t, testOneBlockStore, + bstream.TestBlockWithLIBNum("00000003", "00000002", 2), + bstream.TestBlockWithLIBNum("00000004", "00000003", 3), + bstream.TestBlockWithLIBNum("00000005", "00000004", 4), + bstream.TestBlockWithLIBNum("00000009", "00000008", 8), + ) + + files, err := fh.WalkOneBlocksStoreRange(context.Background(), 4, 5) + require.NoError(t, err) + require.Len(t, files, 2) + assert.True(t, strings.HasPrefix(files[0], "0000000004-")) + assert.True(t, strings.HasPrefix(files[1], "0000000005-")) + + files, err = fh.WalkOneBlocksStoreRange(context.Background(), 6, 8) + require.NoError(t, err) + assert.Empty(t, files) +} + func TestForkableHub_Run(t *testing.T) { t.Helper()