perf(litebe-sync): bound the gRPC block channel with backpressure

Each block range is streamed over one gRPC connection into a channel
that a worker thread drains, encodes, and scans. That channel was
unbounded, so when the server delivered blocks faster than the
scanner consumed them the backlog piled up in memory with no ceiling
- on a fast link against a large range it could grow without limit.

- The channel is now a bounded synchronous channel holding at most
  2048 blocks. When it fills, the fetch side waits, applying
  backpressure up the stream instead of buffering endlessly, so peak
  memory is capped no matter how far ahead the server runs.
- The batch size is untouched: it still streams a whole range over a
  single connection, which is what keeps the connection count low
  enough to avoid socket-buffer exhaustion on Windows. Only the
  in-flight depth between fetch and scan is now bounded.

The same blocks are fetched and scanned in the same order, so the
resulting wallet is byte-identical, verified against the differential
harness. The wallet file format is unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-10-14 08:50:12 -05:00
co-authored by Claude Opus 4.8
parent be69c9bafe
commit c045e56d4e
2 changed files with 5 additions and 4 deletions
@@ -11,7 +11,7 @@ use tokio_rustls::{rustls::ClientConfig};
use tonic::{Request};
use threadpool::ThreadPool;
use std::sync::mpsc::channel;
use std::sync::mpsc::{channel, sync_channel};
use crate::PubCertificate;
use crate::grpc_client::compact_tx_streamer_client::CompactTxStreamerClient;
@@ -113,7 +113,7 @@ where F : Fn(&[u8], u64) {
let request = Request::new(BlockRange { start: Some(bs), end: Some(be) });
let (tx, rx) = channel::<Option<CompactBlock>>();
let (tx, rx) = sync_channel::<Option<CompactBlock>>(2048);
let (ftx, frx) = channel();
pool.execute(move || {
@@ -1660,8 +1660,9 @@ pub fn start_mempool_monitor(lc: Arc<LightClient>) -> Result<(), String> {
// get_block_range streams an ENTIRE range over ONE gRPC connection, so the batch size is
// the dominant factor in how many TCP connections a sync opens. The old 1000 meant ~1300
// connections for a full sync, which on Windows exhausted socket buffers (WSAENOBUFS / os
// error 10055). A large batch (25k) cuts that ~50x (~52 connections for a full sync) with
// no extra memory — blocks are still streamed + scanned incrementally, not buffered.
// error 10055). A large batch (25k) cuts that ~50x (~52 connections for a full sync). Blocks
// are now buffered up to 2048 deep (bounded backpressure, C240), still one connection per
// 25k batch; scan_batch_size must NOT shrink or Windows WSAENOBUFS returns.
let scan_batch_size = 25000;
let mut end_height = std::cmp::min(last_scanned_height + scan_batch_size, latest_block);