From d8ee1e3a334e2e95d490a28246007f0debbb03d6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:56:28 -0400 Subject: [PATCH 1/5] Keep default SOURCE_CHAN_SIZE on invalid value An unparseable value was assigned anyway, leaving every subscription channel with capacity 0 and disconnecting all clients with ErrSubscriptionChannelFull. --- hub/hub.go | 25 +++++++++++++++------- hub/source_chan_size_test.go | 41 ++++++++++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 8 deletions(-) create mode 100644 hub/source_chan_size_test.go diff --git a/hub/hub.go b/hub/hub.go index 7b223ca..1eec442 100644 --- a/hub/hub.go +++ b/hub/hub.go @@ -68,14 +68,7 @@ func NewForkableHubWithOptions(liveSourceFactory bstream.SourceFactory, keepFina } func newForkableHub(liveSourceFactory bstream.SourceFactory, keepFinalBlocks int, oneBlocksStore dstore.Store, hubOptions []Option, extraForkableOptions ...forkable.Option) *ForkableHub { - sourceChanSize := 100 - if os.Getenv("SOURCE_CHAN_SIZE") != "" { - newSize, err := strconv.Atoi(os.Getenv("SOURCE_CHAN_SIZE")) - if err != nil { - zlog.Warn("invalid SOURCE_CHAN_SIZE, ignoring", zap.Error(err)) - } - sourceChanSize = newSize - } + sourceChanSize := sourceChanSizeFromEnv(100) hub := &ForkableHub{ Shutter: shutter.New(), @@ -116,6 +109,22 @@ func newForkableHub(liveSourceFactory bstream.SourceFactory, keepFinalBlocks int return hub } +// sourceChanSizeFromEnv returns the subscription channel size from the +// SOURCE_CHAN_SIZE environment variable, falling back to defaultSize when the +// variable is unset or invalid. +func sourceChanSizeFromEnv(defaultSize int) int { + value := os.Getenv("SOURCE_CHAN_SIZE") + if value == "" { + return defaultSize + } + newSize, err := strconv.Atoi(value) + if err != nil { + zlog.Warn("invalid SOURCE_CHAN_SIZE, ignoring", zap.Error(err)) + return defaultSize + } + return newSize +} + // Option configures a ForkableHub. type Option func(h *ForkableHub) diff --git a/hub/source_chan_size_test.go b/hub/source_chan_size_test.go new file mode 100644 index 0000000..f107360 --- /dev/null +++ b/hub/source_chan_size_test.go @@ -0,0 +1,41 @@ +package hub + +import ( + "testing" + + "github.com/streamingfast/bstream" + "github.com/stretchr/testify/require" +) + +func TestSourceChanSizeFromEnv(t *testing.T) { + tests := []struct { + name string + envValue string + setEnv bool + expect int + }{ + {"unset keeps default", "", false, 100}, + {"empty keeps default", "", true, 100}, + {"garbage keeps default", "garbage", true, 100}, + {"valid value is used", "42", true, 42}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + if test.setEnv { + t.Setenv("SOURCE_CHAN_SIZE", test.envValue) + } + require.Equal(t, test.expect, sourceChanSizeFromEnv(100)) + }) + } +} + +func TestNewForkableHub_InvalidSourceChanSizeEnv(t *testing.T) { + t.Setenv("SOURCE_CHAN_SIZE", "not-a-number") + + lsf := bstream.NewTestSourceFactory() + fh := NewForkableHub(lsf.NewSource, 0, nil) + + require.Equal(t, 100, fh.sourceChannelSize, + "invalid SOURCE_CHAN_SIZE must keep the default channel size instead of using 0 and disconnecting every subscriber") +} From 619f1c6d164d5e56cc254f9c22ce87205c4961b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:56:39 -0400 Subject: [PATCH 2/5] Protect hub subscribers slice with a mutex subscribe runs under the forkable read lock only (all SourceFrom* paths), so concurrent subscribers raced on the slice and a subscription could be lost. The new mutex is a leaf lock, acquired last and never while calling out. --- hub/hub.go | 25 ++++++-- hub/hub_subscribe_race_test.go | 107 +++++++++++++++++++++++++++++++++ 2 files changed, 128 insertions(+), 4 deletions(-) create mode 100644 hub/hub_subscribe_race_test.go diff --git a/hub/hub.go b/hub/hub.go index 1eec442..c9f67e6 100644 --- a/hub/hub.go +++ b/hub/hub.go @@ -22,6 +22,7 @@ import ( "os" "strconv" "strings" + "sync" "time" "github.com/streamingfast/dstore" @@ -45,6 +46,7 @@ type ForkableHub struct { keepFinalBlocks int + subscribersLock sync.Mutex // leaf lock: never acquire any other lock while holding it subscribers []*Subscription sourceChannelSize int @@ -101,7 +103,12 @@ func newForkableHub(liveSourceFactory bstream.SourceFactory, keepFinalBlocks int ) hub.OnTerminating(func(err error) { - for _, sub := range hub.subscribers { + hub.subscribersLock.Lock() + subscribers := make([]*Subscription, len(hub.subscribers)) + copy(subscribers, hub.subscribers) + hub.subscribersLock.Unlock() + + for _, sub := range subscribers { sub.Shutdown(err) } }) @@ -209,19 +216,26 @@ func (h *ForkableHub) IsReady() bool { } } -// subscribe must be called while hub is locked +// subscribe must be called while the forkable is locked (via forkable.CallWithBlocks*) +// so that no new block can be broadcast between the snapshot of initialBlocks and the +// registration of the subscription. The subscribers slice itself is protected by +// subscribersLock: the forkable lock is not enough because CallWithBlocks* only takes +// a read lock, so multiple subscribe calls can run concurrently. func (h *ForkableHub) subscribe(handler bstream.Handler, initialBlocks []*bstream.PreprocessedBlock, withPartials bool) *Subscription { chanSize := h.sourceChannelSize + len(initialBlocks) sub := NewSubscription(handler, chanSize, withPartials) for _, ppblk := range initialBlocks { _ = sub.push(ppblk) } + h.subscribersLock.Lock() h.subscribers = append(h.subscribers, sub) + h.subscribersLock.Unlock() return sub } -// unsubscribe must be called while hub is locked func (h *ForkableHub) unsubscribe(removeSub *Subscription) { + h.subscribersLock.Lock() + defer h.subscribersLock.Unlock() var newSubscriber []*Subscription for _, sub := range h.subscribers { if sub != removeSub { @@ -527,7 +541,10 @@ func (h *ForkableHub) broadcastBlock(blk *pbbstream.Block, obj any) error { preprocBlock := &bstream.PreprocessedBlock{Block: blk, Obj: obj} - subscribers := h.subscribers // we may remove some from the original slice during the loop + h.subscribersLock.Lock() + subscribers := make([]*Subscription, len(h.subscribers)) + copy(subscribers, h.subscribers) // we may remove some from the original slice during the loop + h.subscribersLock.Unlock() for _, sub := range subscribers { err := sub.push(preprocBlock) diff --git a/hub/hub_subscribe_race_test.go b/hub/hub_subscribe_race_test.go new file mode 100644 index 0000000..c3aba33 --- /dev/null +++ b/hub/hub_subscribe_race_test.go @@ -0,0 +1,107 @@ +package hub + +import ( + "fmt" + "sync" + "testing" + "time" + + "github.com/streamingfast/bstream" + "github.com/streamingfast/bstream/forkable" + pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1" + "github.com/streamingfast/shutter" + "github.com/stretchr/testify/require" +) + +// TestForkableHub_ConcurrentSubscribe exercises concurrent subscriptions while +// blocks are being broadcast. All SourceFrom* paths call subscribe under the +// forkable *read* lock only, so two concurrent subscribers used to race on the +// h.subscribers slice (caught by -race) and could lose a subscription entirely. +func TestForkableHub_ConcurrentSubscribe(t *testing.T) { + fh := &ForkableHub{ + Shutter: shutter.New(), + logger: zlog, + sourceChannelSize: 100, + } + fh.forkable = forkable.New(bstream.HandlerFunc(fh.broadcastBlock), + forkable.HoldBlocksUntilLIB(), + forkable.WithKeptFinalBlocks(100), + forkable.WithFilters(bstream.StepsAllWithPartial), + ) + + // Seed the forkable so that SourceFromBlockNum can serve block 4. + require.NoError(t, fh.forkable.ProcessBlock(bstream.TestBlockWithLIBNum("00000003", "00000002", 2), nil)) + require.NoError(t, fh.forkable.ProcessBlock(bstream.TestBlockWithLIBNum("00000004", "00000003", 3), nil)) + + const subscriberCount = 20 + const lastBroadcastBlock = uint64(40) + + start := make(chan struct{}) + + // Broadcaster: pushes blocks 5..40 through the forkable while subscribers register. + broadcastDone := make(chan struct{}) + go func() { + defer close(broadcastDone) + <-start + prev := "00000004" + for num := uint64(5); num <= lastBroadcastBlock; num++ { + id := fmt.Sprintf("%08x", num) + if err := fh.forkable.ProcessBlock(bstream.TestBlockWithLIBNum(id, prev, 3), nil); err != nil { + panic(err) + } + prev = id + } + }() + + type subscriber struct { + source bstream.Source + received chan uint64 + } + subs := make([]*subscriber, subscriberCount) + + var wg sync.WaitGroup + for i := 0; i < subscriberCount; i++ { + sub := &subscriber{received: make(chan uint64, 256)} + subs[i] = sub + wg.Add(1) + go func() { + defer wg.Done() + <-start + handler := bstream.HandlerFunc(func(blk *pbbstream.Block, _ any) error { + sub.received <- blk.Number + return nil + }) + sub.source = fh.SourceFromBlockNum(4, handler) + }() + } + + close(start) + wg.Wait() + <-broadcastDone + + // Every subscription must exist and receive blocks up to the last one + // broadcast after it registered. A lost subscription (dropped by a racing + // append) would never receive it. + for i, sub := range subs { + require.NotNil(t, sub.source, "subscriber %d did not get a source", i) + go sub.source.Run() + + var last uint64 + timeout := time.After(5 * time.Second) + drain: + for { + select { + case num := <-sub.received: + if num > last { + last = num + } + if last == lastBroadcastBlock { + break drain + } + case <-timeout: + t.Fatalf("subscriber %d timed out waiting for block %d, last received %d (lost subscription?)", i, lastBroadcastBlock, last) + } + } + sub.source.Shutdown(nil) + } +} From 0b18d3c45767d8f10f638ab2824ccf4b47aad623 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:56:50 -0400 Subject: [PATCH 3/5] Start blocksFromNum at first block >= num Chains like Solana or NEAR skip block numbers; requiring an exact match made the hub unable to serve a skipped start block, stalling the consumer on the file source. Requests below the lowest held block still fail so callers fall back to another source. --- forkable/blocks_from_num_test.go | 66 ++++++++++++++++++++++++++++++++ forkable/forkable.go | 12 +++++- 2 files changed, 77 insertions(+), 1 deletion(-) create mode 100644 forkable/blocks_from_num_test.go diff --git a/forkable/blocks_from_num_test.go b/forkable/blocks_from_num_test.go new file mode 100644 index 0000000..1f0578b --- /dev/null +++ b/forkable/blocks_from_num_test.go @@ -0,0 +1,66 @@ +package forkable + +import ( + "testing" + + "github.com/streamingfast/bstream" + pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Some chains (Solana, NEAR, ...) can skip block numbers: requesting a start +// block that was skipped must serve from the first block above it instead of +// failing forever. +func TestCallWithBlocksFromNum_SkippedBlockNums(t *testing.T) { + p := New( + bstream.HandlerFunc(func(blk *pbbstream.Block, obj any) error { return nil }), + HoldBlocksUntilLIB(), + WithKeptFinalBlocks(100), + ) + + blocks := []*pbbstream.Block{ + bstream.TestBlockWithLIBNum("00000003", "00000002", 2), + bstream.TestBlockWithLIBNum("00000004", "00000003", 3), + bstream.TestBlockWithLIBNum("00000006", "00000004", 3), // number 5 skipped by the chain + bstream.TestBlockWithLIBNum("00000007", "00000006", 3), + } + for _, blk := range blocks { + require.NoError(t, p.ProcessBlock(blk, nil)) + } + + blockNumsFrom := func(num uint64) (out []uint64, err error) { + err = p.CallWithBlocksFromNum(num, func(blks []*bstream.PreprocessedBlock) { + for _, b := range blks { + out = append(out, b.Num()) + } + }, false) + return + } + + t.Run("requesting a skipped block num starts at the next available block", func(t *testing.T) { + nums, err := blockNumsFrom(5) + require.NoError(t, err) + assert.Equal(t, []uint64{6, 7}, nums) + }) + + t.Run("requesting an existing block num still starts exactly there", func(t *testing.T) { + nums, err := blockNumsFrom(4) + require.NoError(t, err) + assert.Equal(t, []uint64{4, 6, 7}, nums) + + nums, err = blockNumsFrom(3) + require.NoError(t, err) + assert.Equal(t, []uint64{3, 4, 6, 7}, nums) + }) + + t.Run("requesting below the lowest held block fails so caller can use another source", func(t *testing.T) { + _, err := blockNumsFrom(2) + require.Error(t, err) + }) + + t.Run("requesting above head fails", func(t *testing.T) { + _, err := blockNumsFrom(8) + require.Error(t, err) + }) +} diff --git a/forkable/forkable.go b/forkable/forkable.go index e6c8dc2..38d0c10 100644 --- a/forkable/forkable.go +++ b/forkable/forkable.go @@ -180,6 +180,10 @@ func (p *Forkable) blocksFromNumWithForks(startNum uint64) ([]*bstream.Preproces return out, nil } +// blocksFromNum returns the blocks of the current canonical segment starting +// at the first block whose number is >= num. The '>=' matters: chains like +// Solana or NEAR can skip block numbers, so requiring an exact match would +// make the whole segment unservable when the requested number was skipped. func (p *Forkable) blocksFromNum(num uint64) ([]*bstream.PreprocessedBlock, error) { if !p.forkDB.HasLIB() { return nil, fmt.Errorf("no lib") @@ -196,13 +200,19 @@ func (p *Forkable) blocksFromNum(num uint64) ([]*bstream.PreprocessedBlock, erro return nil, fmt.Errorf("head segment does not reach LIB") } + if len(seg) > 0 && seg[0].AsRef().Num() > num { + // we don't hold the requested block: serving from the lowest block we + // have would silently skip blocks, let the caller use another source + return nil, fmt.Errorf("lowest block in complete segment %d is above requested block num %d", seg[0].AsRef().Num(), num) + } + libNum := p.forkDB.libRef.Num() var out []*bstream.PreprocessedBlock var seenBlock bool for i := range seg { ref := seg[i].AsRef() - if !seenBlock && ref.Num() == num { + if !seenBlock && ref.Num() >= num { seenBlock = true } From e2a0189341c08c3160a4561d154fd264b6577295 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:56:50 -0400 Subject: [PATCH 4/5] Fix double attempt increment in retryable getter The error path bumped attempt on top of the loop post statement, halving the effective retries. Also make 'attempts' mean the total number of tries. --- tracker.go | 3 +-- tracker_test.go | 64 +++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 2 deletions(-) create mode 100644 tracker_test.go diff --git a/tracker.go b/tracker.go index 9e225d6..bd1a689 100644 --- a/tracker.go +++ b/tracker.go @@ -234,7 +234,7 @@ func HighestBlockRefGetter(getters ...BlockRefGetter) BlockRefGetter { func RetryableBlockRefGetter(attempts int, wait time.Duration, next BlockRefGetter) BlockRefGetter { return func(ctx context.Context) (ref BlockRef, err error) { var errs []string - for attempt := 0; attempts == -1 || attempt <= attempts; attempt++ { + for attempt := 0; attempts == -1 || attempt < attempts; attempt++ { if err = ctx.Err(); err != nil { errs = append(errs, err.Error()) break @@ -244,7 +244,6 @@ func RetryableBlockRefGetter(attempts int, wait time.Duration, next BlockRefGett if err != nil { zlog.Debug("got an error from a block ref getter", zap.Error(err)) errs = append(errs, err.Error()) - attempt++ time.Sleep(wait) continue } diff --git a/tracker_test.go b/tracker_test.go new file mode 100644 index 0000000..275963e --- /dev/null +++ b/tracker_test.go @@ -0,0 +1,64 @@ +package bstream + +import ( + "context" + "fmt" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestRetryableBlockRefGetter_AttemptCount(t *testing.T) { + tests := []struct { + name string + attempts int + succeedOnCall int // 0 means never succeed + expectCalls int + expectError bool + }{ + {"always failing consumes each configured attempt", 4, 0, 4, true}, + {"single attempt", 1, 0, 1, true}, + {"succeeds on first call", 4, 1, 1, false}, + {"succeeds after two failures", 4, 3, 3, false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var calls int + getter := BlockRefGetter(func(ctx context.Context) (BlockRef, error) { + calls++ + if test.succeedOnCall != 0 && calls >= test.succeedOnCall { + return NewBlockRef("00000001", 1), nil + } + return nil, fmt.Errorf("failure %d", calls) + }) + + ref, err := RetryableBlockRefGetter(test.attempts, 0, getter)(context.Background()) + + assert.Equal(t, test.expectCalls, calls) + if test.expectError { + require.Error(t, err) + } else { + require.NoError(t, err) + require.NotNil(t, ref) + assert.EqualValues(t, 1, ref.Num()) + } + }) + } +} + +func TestRetryableBlockRefGetter_CanceledContext(t *testing.T) { + var calls int + getter := BlockRefGetter(func(ctx context.Context) (BlockRef, error) { + calls++ + return nil, fmt.Errorf("failure") + }) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := RetryableBlockRefGetter(4, 0, getter)(ctx) + require.Error(t, err) + assert.Equal(t, 0, calls, "canceled context must prevent any call to the wrapped getter") +} From f020ddc800e5ea3f7cc2771bf23d13c9940e302f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ste=CC=81phane=20Duchesneau?= Date: Sun, 5 Jul 2026 19:56:50 -0400 Subject: [PATCH 5/5] Compare Range end blocks by value in Equals endBlock was compared by pointer, so bounded ranges built separately never compared equal and IsNext always returned false. --- range.go | 13 +++++++++---- range_test.go | 46 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 55 insertions(+), 4 deletions(-) diff --git a/range.go b/range.go index 251b045..c8ff9bf 100644 --- a/range.go +++ b/range.go @@ -343,10 +343,15 @@ func (r *Range) IsNext(next *Range, size uint64) bool { } func (r *Range) Equals(other *Range) bool { - return r.startBlock == other.startBlock && - r.endBlock == other.endBlock && - r.exclusiveStartBlock == other.exclusiveStartBlock && - r.exclusiveEndBlock == other.exclusiveEndBlock + if r.startBlock != other.startBlock || + r.exclusiveStartBlock != other.exclusiveStartBlock || + r.exclusiveEndBlock != other.exclusiveEndBlock { + return false + } + if r.endBlock == nil || other.endBlock == nil { + return r.endBlock == nil && other.endBlock == nil + } + return *r.endBlock == *other.endBlock } func (r *Range) Size() (uint64, error) { diff --git a/range_test.go b/range_test.go index 9bd39ed..dcf4ea4 100644 --- a/range_test.go +++ b/range_test.go @@ -357,3 +357,49 @@ func errorEqual(expectedErrString string) require.ErrorAssertionFunc { require.EqualError(t, err, expectedErrString, msgAndArgs...) } } + +func TestRange_Equals(t *testing.T) { + tests := []struct { + name string + left *Range + right *Range + expect bool + }{ + {"bounded ranges with equal values but distinct endBlock pointers", &Range{10, ptr(20), false, false}, &Range{10, ptr(20), false, false}, true}, + {"same range instance", MustParseRange("10-20"), MustParseRange("10-20"), true}, + {"open ended ranges", &Range{10, nil, false, false}, &Range{10, nil, false, false}, true}, + {"open ended vs bounded", &Range{10, nil, false, false}, &Range{10, ptr(20), false, false}, false}, + {"bounded vs open ended", &Range{10, ptr(20), false, false}, &Range{10, nil, false, false}, false}, + {"different end block", &Range{10, ptr(20), false, false}, &Range{10, ptr(30), false, false}, false}, + {"different start block", &Range{10, ptr(20), false, false}, &Range{11, ptr(20), false, false}, false}, + {"different exclusive start", &Range{10, ptr(20), true, false}, &Range{10, ptr(20), false, false}, false}, + {"different exclusive end", &Range{10, ptr(20), false, true}, &Range{10, ptr(20), false, false}, false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.expect, test.left.Equals(test.right)) + }) + } +} + +func TestRange_IsNext(t *testing.T) { + tests := []struct { + name string + in *Range + next *Range + size uint64 + expect bool + }{ + {"contiguous bounded ranges", &Range{10, ptr(20), false, false}, &Range{20, ptr(30), false, false}, 10, true}, + {"non contiguous bounded ranges", &Range{10, ptr(20), false, false}, &Range{30, ptr(40), false, false}, 10, false}, + {"contiguous open ended ranges", &Range{10, nil, false, false}, &Range{20, nil, false, false}, 10, true}, + {"wrong size", &Range{10, ptr(20), false, false}, &Range{20, ptr(35), false, false}, 10, false}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + assert.Equal(t, test.expect, test.in.IsNext(test.next, test.size)) + }) + } +}