fix hub subscriptions with_partials=false: they should receive 'lastPartial' blocks as if they were full blocks

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- 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).
 
 ### Changed

hub/subscription.go

@@ -18,6 +18,7 @@ import (
 	"fmt"
 
 	"github.com/streamingfast/bstream"
+	pbbstream "github.com/streamingfast/bstream/pb/sf/bstream/v1"
 	"github.com/streamingfast/shutter"
 )
 
@@ -46,7 +47,15 @@ func NewSubscription(handler bstream.Handler, chanSize int, withPartial bool) *S
 
 func (s *Subscription) push(ppblk *bstream.PreprocessedBlock) error {
 	if !s.withPartial && ppblk.Block.PartialIndex != 0 {
-		return nil
+		// A no-partial subscriber only wants full blocks. Intermediate partials are
+		// dropped, but the LastPartial carries the complete block, so it is delivered
+		// as if it were a full block. The block is broadcast to every subscriber (and
+		// shared with the forkdb), so we must not mutate it: deliver a thin copy with
+		// the partial markers cleared, sharing the (immutable) payload.
+		if !ppblk.Block.LastPartial {
+			return nil
+		}
+		ppblk = &bstream.PreprocessedBlock{Block: asFullBlock(ppblk.Block), Obj: ppblk.Obj}
 	}
 	if len(s.blocks) == cap(s.blocks) {
 		return ErrSubscriptionChannelFull
@@ -55,6 +64,26 @@ func (s *Subscription) push(ppblk *bstream.PreprocessedBlock) error {
 	return nil
 }
 
+// asFullBlock returns a shallow copy of blk with the partial markers cleared, so a
+// LastPartial can be delivered to a no-partial subscriber as if it were a full block.
+// Only the block metadata is copied; the payload pointer is shared (it is never mutated).
+func asFullBlock(blk *pbbstream.Block) *pbbstream.Block {
+	return &pbbstream.Block{
+		Number:         blk.Number,
+		Id:             blk.Id,
+		ParentId:       blk.ParentId,
+		Timestamp:      blk.Timestamp,
+		LibNum:         blk.LibNum,
+		PayloadKind:    blk.PayloadKind,
+		PayloadVersion: blk.PayloadVersion,
+		PayloadBuffer:  blk.PayloadBuffer,
+		HeadNum:        blk.HeadNum,
+		ParentNum:      blk.ParentNum,
+		Payload:        blk.Payload,
+		// PartialIndex and LastPartial intentionally left at zero: this is a full block.
+	}
+}
+
 func lookAhead(ch chan *bstream.PreprocessedBlock) *bstream.PreprocessedBlock {
 	select {
 	case ppblk := <-ch:

hub/subscription_repro_test.go

@@ -78,3 +78,59 @@ func TestSubscriptionEatsLastPartial(t *testing.T) {
 	}
 	t.Fatalf("closing last-partial block 0004a was never delivered to the subscriber")
 }
+
+// A no-partial subscriber must not receive intermediate partials, but must receive the
+// closing "last partial" of each block as if it were a full block (partial markers
+// cleared), so it never stalls waiting for a separate PartialIndex==0 block. The
+// original blocks (shared with other subscribers and the forkdb) must not be mutated.
+func TestSubscriptionNoPartialPromotesLastPartial(t *testing.T) {
+	var received []*pbbstream.Block
+	handler := bstream.HandlerFunc(func(blk *pbbstream.Block, obj interface{}) error {
+		received = append(received, blk)
+		return nil
+	})
+
+	sub := NewSubscription(handler, 100, false) // withPartial=false
+
+	lastPartial4 := reproBlock("0004a", 4, 3, true)
+	full5 := reproBlock("00000005", 5, 0, false) // regular full block
+
+	push := func(step bstream.StepType, blk *pbbstream.Block) {
+		if err := sub.push(&bstream.PreprocessedBlock{Block: blk, Obj: stepObj{step}}); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	push(bstream.StepPartial, reproBlock("0004p1", 4, 1, false)) // intermediate -> dropped
+	push(bstream.StepPartial, reproBlock("0004p2", 4, 2, false)) // intermediate -> dropped
+	push(bstream.StepNewPartial, lastPartial4)                   // closing -> delivered as full block
+	push(bstream.StepNew, full5)                                 // full block -> delivered as-is
+
+	go func() {
+		time.Sleep(200 * time.Millisecond)
+		sub.Shutdown(nil)
+	}()
+	sub.Run()
+
+	if len(received) != 2 {
+		var ids []string
+		for _, b := range received {
+			ids = append(ids, fmt.Sprintf("%s(idx=%d)", b.Id, b.PartialIndex))
+		}
+		t.Fatalf("expected 2 blocks delivered, got %d: %v", len(received), ids)
+	}
+
+	// block 4: the last-partial, delivered with partial markers cleared
+	if received[0].Id != "0004a" || received[0].PartialIndex != 0 || received[0].LastPartial {
+		t.Fatalf("last-partial not promoted to full block: %+v", received[0])
+	}
+	// block 5: full block untouched
+	if received[1].Id != "00000005" || received[1].PartialIndex != 0 {
+		t.Fatalf("full block altered: %+v", received[1])
+	}
+
+	// the shared original block must be untouched (thin copy, no mutation)
+	if lastPartial4.PartialIndex != 3 || !lastPartial4.LastPartial {
+		t.Fatalf("original last-partial block was mutated: idx=%d last=%v", lastPartial4.PartialIndex, lastPartial4.LastPartial)
+	}
+}