fix: lower pipeline sizes (#1437)

Signed-off-by: Chris Gianelloni <wolf31o2@blinklabs.io>

Author: wolf31o2

ledger/chainsync.go

@@ -81,6 +81,19 @@ func (ls *LedgerState) handleEventChainsync(evt event.Event) {
 		}
 	} else if e.BlockHeader != nil {
 		if err := ls.handleEventChainsyncBlockHeader(e); err != nil {
+			// Header queue full is expected during bulk sync when
+			// pipelined headers arrive faster than blockfetch can
+			// drain them. Log at DEBUG to avoid log spam.
+			if errors.Is(err, chain.ErrHeaderQueueFull) {
+				ls.config.Logger.Debug(
+					"failed to handle block header",
+					"component", "ledger",
+					"error", err,
+					"slot", e.Point.Slot,
+					"hash", hex.EncodeToString(e.Point.Hash),
+				)
+				return
+			}
 			ls.config.Logger.Error(
 				"failed to handle block header",
 				"component", "ledger",

ouroboros/chainsync.go

@@ -46,10 +46,14 @@ func (o *Ouroboros) chainsyncClientConnOpts() []ochainsync.ChainSyncOptionFunc {
 	return []ochainsync.ChainSyncOptionFunc{
 		ochainsync.WithRollForwardFunc(o.chainsyncClientRollForward),
 		ochainsync.WithRollBackwardFunc(o.chainsyncClientRollBackward),
-		// Enable pipelining of RequestNext messages to speed up chainsync
-		ochainsync.WithPipelineLimit(50),
-		// Set the recv queue size to 2x our pipeline limit
-		ochainsync.WithRecvQueueSize(100),
+		// Pipeline enough headers to keep one blockfetch batch (500
+		// blocks) ready while the previous batch processes. A depth
+		// of 10 is sufficient; higher values flood the header queue
+		// and waste CPU parsing headers that are immediately dropped.
+		ochainsync.WithPipelineLimit(10),
+		// Recv queue at 2x pipeline limit to absorb bursts without
+		// blocking the protocol goroutine.
+		ochainsync.WithRecvQueueSize(20),
 	}
 }