feat(#50): prefetch-on-send (UN-DEPLOYED — needs supervised bridge-off verify)

pkg/daemon/daemon.go

@@ -279,6 +279,12 @@ type Daemon struct {
 	mpool    *mpool.Pool     // gossipsub mempool publisher (#45 Stage 4)
 	bitswap  *bitswap.Client // libp2p block source on the embedded fetcher (#50)
 
+	// sendWarmer pre-warms a sent tx's message/receipt blocks into the
+	// Bitswap cache in the background so the receipt poll resolves locally
+	// (lantern#50 prefetch-on-send). Wired to chainAPI.OnSentTx in
+	// startInternal; best-effort, read-only.
+	sendWarmer *sendWarmer
+
 	// Internal cancellation: derived from caller's ctx in Start.
 	cancel context.CancelFunc
 }

pkg/daemon/sendwarm.go

@@ -0,0 +1,185 @@
+package daemon
+
+// lantern#50 prefetch-on-send.
+//
+// When eth_sendRawTransaction publishes a tx locally (#45 Stage 4), the
+// receipt poll that follows (eth_getTransactionReceipt -> StateSearchMsg)
+// has to fetch the freshly-produced message block, its message AMTs, and
+// the receipt block over Bitswap. Bridge-off, those blocks live only on the
+// peer set, and a just-produced block isn't reliably served by random
+// gossip peers inside the client's poll window. The result was the residual
+// "net/bitswap: context canceled" miss documented in #50.
+//
+// The warmer closes that gap proactively: the moment we publish a tx, we
+// start polling StateSearchMsg for its message CID in the BACKGROUND, on a
+// generous standalone context that isn't tied to any client RPC deadline.
+// That background search drives Bitswap to pull the message + AMT + receipt
+// blocks into the local blockstore as soon as the tx lands. By the time the
+// client's own (short-deadline) receipt poll runs, the blocks are warm and
+// the search resolves from cache instead of racing a cold cross-peer fetch.
+//
+// This is purely additive and best-effort:
+//   - It never affects the send result (fired non-blocking from the handler).
+//   - It only READS (StateSearchMsg); it never publishes or mutates state.
+//   - A nil warmer (hook unset) means the old behavior, unchanged.
+//   - It self-bounds: each warm goroutine stops on success, on the message
+//     becoming un-findable past its lookback, or after maxWarmDuration.
+
+import (
+	"context"
+	"sync"
+	"time"
+
+	"github.com/ipfs/go-cid"
+
+	"github.com/Reiers/lantern/chain/types"
+	"github.com/Reiers/lantern/rpc/handlers"
+)
+
+const (
+	// warmPollInterval is how often the background warmer re-runs
+	// StateSearchMsg for an in-flight tx. Matches the chain's ~30s block
+	// cadence loosely; a tx typically lands within 1-3 polls.
+	warmPollInterval = 5 * time.Second
+
+	// maxWarmDuration bounds a single tx's warming loop. A calibration/
+	// mainnet tx that hasn't landed in this window is either dropped or
+	// stuck (an #47 concern), not a Bitswap-availability problem, so we
+	// stop warming and let the normal poll/bridge path handle it.
+	maxWarmDuration = 10 * time.Minute
+
+	// maxConcurrentWarms caps in-flight warm goroutines so a burst of
+	// sends can't spawn unbounded work. The SP write->confirm loop only
+	// has a handful of txs in flight at once; well past that we simply
+	// skip warming (the receipt path still works, just without the warm
+	// assist).
+	maxConcurrentWarms = 64
+)
+
+// searchFn resolves a message CID, returning (found, err). It is the seam
+// the warmer drives in a loop; production wires it to ChainAPI.StateSearchMsg
+// (which pulls the needed blocks over Bitswap as a side effect). Tests inject
+// a deterministic stand-in.
+type searchFn func(ctx context.Context, msgCID cid.Cid) (found bool, err error)
+
+// sendWarmer runs background StateSearchMsg polls for recently-sent txs to
+// pre-warm their message/receipt blocks into the Bitswap cache (#50).
+type sendWarmer struct {
+	ctx    context.Context
+	search searchFn
+
+	// pollInterval / maxDuration / maxConcurrent are fields (not the bare
+	// consts) so tests can drive the loop fast. Production uses the const
+	// defaults via newSendWarmer.
+	pollInterval  time.Duration
+	maxDuration   time.Duration
+	maxConcurrent int
+
+	mu       sync.Mutex
+	inFlight map[cid.Cid]struct{}
+	active   int
+}
+
+// newSendWarmer builds a warmer bound to the daemon's root context. The
+// chainAPI must be the same handler the RPC server uses (so StateSearchMsg
+// reads through the embedded Bitswap-backed BlockGetter).
+func newSendWarmer(ctx context.Context, chain *handlers.ChainAPI) *sendWarmer {
+	w := newSendWarmerWithSearch(ctx, func(sctx context.Context, msgCID cid.Cid) (bool, error) {
+		lookup, err := chain.StateSearchMsg(sctx, types.TipSetKey{}, msgCID, 0, false)
+		return lookup != nil, err
+	})
+	// A nil chain means "no real handler"; mark the warmer inert so Warm
+	// is a no-op (mirrors the old `w.chain == nil` guard).
+	if chain == nil {
+		w.search = nil
+	}
+	return w
+}
+
+// newSendWarmerWithSearch is the testable constructor: it takes the search
+// seam directly.
+func newSendWarmerWithSearch(ctx context.Context, search searchFn) *sendWarmer {
+	return &sendWarmer{
+		ctx:           ctx,
+		search:        search,
+		pollInterval:  warmPollInterval,
+		maxDuration:   maxWarmDuration,
+		maxConcurrent: maxConcurrentWarms,
+		inFlight:      make(map[cid.Cid]struct{}),
+	}
+}
+
+// Warm starts a background warming loop for msgCID. Safe to call from the
+// send hot path: it returns immediately and dedups concurrent calls for the
+// same message. A no-op if the warmer or its chain handle is nil.
+func (w *sendWarmer) Warm(msgCID cid.Cid) {
+	if w == nil || w.search == nil || !msgCID.Defined() {
+		return
+	}
+
+	w.mu.Lock()
+	if _, dup := w.inFlight[msgCID]; dup {
+		w.mu.Unlock()
+		return
+	}
+	if w.active >= w.maxConcurrent {
+		// Over budget: skip warming. The receipt path still resolves the
+		// tx (just without the warm assist), so this only loses the
+		// optimization under heavy burst, never correctness.
+		w.mu.Unlock()
+		return
+	}
+	w.inFlight[msgCID] = struct{}{}
+	w.active++
+	w.mu.Unlock()
+
+	go w.run(msgCID)
+}
+
+func (w *sendWarmer) run(msgCID cid.Cid) {
+	defer func() {
+		w.mu.Lock()
+		delete(w.inFlight, msgCID)
+		w.active--
+		w.mu.Unlock()
+	}()
+
+	deadline := time.Now().Add(w.maxDuration)
+	ticker := time.NewTicker(w.pollInterval)
+	defer ticker.Stop()
+
+	for {
+		// Each search gets a bounded child context so a single hung
+		// Bitswap round can't pin the goroutine; the standalone budget
+		// (well above the StateSearchMsg internal 18s retry window) is
+		// what makes this a real warm rather than a client-deadline race.
+		sctx, cancel := context.WithTimeout(w.ctx, warmSearchBudget)
+		found, err := w.search(sctx, msgCID)
+		cancel()
+
+		switch {
+		case err != nil:
+			// Transient (e.g. a block still uncached). Keep polling; the
+			// whole point is to drive Bitswap to fetch it.
+		case found:
+			// Landed and resolved locally: blocks are now warm, the
+			// client's receipt poll will hit cache. Done.
+			return
+		}
+
+		if time.Now().After(deadline) {
+			return
+		}
+		select {
+		case <-w.ctx.Done():
+			return
+		case <-ticker.C:
+		}
+	}
+}
+
+// warmSearchBudget is the per-attempt ceiling for a background warm search.
+// It must comfortably exceed StateSearchMsg's internal retry window (18s)
+// so a warm attempt contains a full set of Bitswap rounds rather than being
+// cut short the way a client's short receipt-poll deadline would be.
+const warmSearchBudget = 25 * time.Second

pkg/daemon/sendwarm_test.go

@@ -0,0 +1,178 @@
+package daemon
+
+import (
+	"context"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/ipfs/go-cid"
+	mh "github.com/multiformats/go-multihash"
+)
+
+// testCID returns a distinct, defined CID for test n.
+func testCID(t *testing.T, n byte) cid.Cid {
+	t.Helper()
+	h, err := mh.Sum([]byte{n, n, n, n}, mh.SHA2_256, -1)
+	if err != nil {
+		t.Fatalf("multihash: %v", err)
+	}
+	return cid.NewCidV1(cid.DagCBOR, h)
+}
+
+// newTestWarmer builds a warmer with fast timings around an injected search.
+func newTestWarmer(ctx context.Context, search searchFn) *sendWarmer {
+	w := newSendWarmerWithSearch(ctx, search)
+	w.pollInterval = 1 * time.Millisecond
+	w.maxDuration = 2 * time.Second
+	return w
+}
+
+// TestWarm_StopsOnFound: the loop ends as soon as the search reports found,
+// and does not keep polling afterward.
+func TestWarm_StopsOnFound(t *testing.T) {
+	var calls int32
+	foundAfter := int32(3)
+	done := make(chan struct{})
+
+	w := newTestWarmer(context.Background(), func(_ context.Context, _ cid.Cid) (bool, error) {
+		n := atomic.AddInt32(&calls, 1)
+		if n >= foundAfter {
+			close(done)
+			return true, nil
+		}
+		return false, nil // not yet on-chain
+	})
+
+	w.Warm(testCID(t, 1))
+
+	select {
+	case <-done:
+	case <-time.After(2 * time.Second):
+		t.Fatal("warm loop never reported found")
+	}
+
+	// Give the goroutine a beat to exit, then confirm it stopped polling.
+	time.Sleep(20 * time.Millisecond)
+	stable := atomic.LoadInt32(&calls)
+	time.Sleep(30 * time.Millisecond)
+	if got := atomic.LoadInt32(&calls); got != stable {
+		t.Fatalf("warmer kept polling after found: %d -> %d", stable, got)
+	}
+	if w.activeCount() != 0 {
+		t.Fatalf("active count not drained: %d", w.activeCount())
+	}
+}
+
+// TestWarm_Dedup: concurrent Warm calls for the SAME cid run a single loop.
+func TestWarm_Dedup(t *testing.T) {
+	release := make(chan struct{})
+	var distinct sync.Map
+	var running int32
+
+	w := newTestWarmer(context.Background(), func(_ context.Context, c cid.Cid) (bool, error) {
+		distinct.Store(c, true)
+		atomic.AddInt32(&running, 1)
+		<-release // hold the loop open so dedup is observable
+		return true, nil
+	})
+
+	c := testCID(t, 2)
+	for i := 0; i < 8; i++ {
+		w.Warm(c)
+	}
+	// Let the single goroutine enter search.
+	time.Sleep(30 * time.Millisecond)
+
+	if w.activeCount() != 1 {
+		t.Fatalf("expected exactly 1 active warm for a deduped cid, got %d", w.activeCount())
+	}
+	if got := atomic.LoadInt32(&running); got != 1 {
+		t.Fatalf("expected 1 concurrent search, got %d", got)
+	}
+	close(release)
+}
+
+// TestWarm_ConcurrencyCap: at most maxConcurrent loops run at once; extra
+// sends are dropped (no goroutine), preserving correctness without unbounded
+// fan-out.
+func TestWarm_ConcurrencyCap(t *testing.T) {
+	release := make(chan struct{})
+	w := newTestWarmer(context.Background(), func(_ context.Context, _ cid.Cid) (bool, error) {
+		<-release
+		return true, nil
+	})
+	w.maxConcurrent = 4
+
+	for i := 0; i < 10; i++ {
+		w.Warm(testCID(t, byte(100+i)))
+	}
+	time.Sleep(30 * time.Millisecond)
+
+	if got := w.activeCount(); got != 4 {
+		t.Fatalf("expected active capped at 4, got %d", got)
+	}
+	close(release)
+}
+
+// TestWarm_ContextCancelStops: cancelling the root context ends the loop even
+// if the search never reports found.
+func TestWarm_ContextCancelStops(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	exited := make(chan struct{})
+
+	w := newSendWarmerWithSearch(ctx, func(sctx context.Context, _ cid.Cid) (bool, error) {
+		return false, nil // never lands
+	})
+	w.pollInterval = 1 * time.Millisecond
+	w.maxDuration = time.Hour // would otherwise run a long time
+
+	// Wrap run completion detection via a tiny shim cid.
+	go func() {
+		w.Warm(testCID(t, 9))
+		// Warm spawns its own goroutine; poll activeCount to detect exit.
+		for {
+			if w.activeCount() == 0 {
+				close(exited)
+				return
+			}
+			time.Sleep(2 * time.Millisecond)
+		}
+	}()
+
+	time.Sleep(20 * time.Millisecond)
+	cancel()
+
+	select {
+	case <-exited:
+	case <-time.After(2 * time.Second):
+		t.Fatal("warm loop did not stop on context cancel")
+	}
+}
+
+// TestWarm_NilSafe: a nil warmer and a nil-search warmer are both no-ops.
+func TestWarm_NilSafe(t *testing.T) {
+	var nilW *sendWarmer
+	nilW.Warm(testCID(t, 1)) // must not panic
+
+	inert := newSendWarmer(context.Background(), nil) // nil chain -> inert
+	inert.Warm(testCID(t, 1))
+	if inert.activeCount() != 0 {
+		t.Fatalf("inert warmer started a loop: active=%d", inert.activeCount())
+	}
+
+	// Undefined CID is ignored.
+	w := newTestWarmer(context.Background(), func(_ context.Context, _ cid.Cid) (bool, error) {
+		t.Fatal("search called for undefined cid")
+		return false, nil
+	})
+	w.Warm(cid.Undef)
+}
+
+// activeCount is a test helper exposing the live warm count.
+func (w *sendWarmer) activeCount() int {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.active
+}

pkg/daemon/start.go

@@ -248,6 +248,14 @@ func (d *Daemon) startInternal(ctx context.Context) error {
 		chainAPI.OnLocalMiss = pf.AddAddr
 	}
 
+	// lantern#50 prefetch-on-send: when eth_sendRawTransaction publishes a
+	// tx locally, warm its message/receipt blocks into the Bitswap cache in
+	// the background so the follow-up receipt poll resolves locally instead
+	// of racing a cold cross-peer fetch. Bound to the daemon root context
+	// (cancelled on Stop). Best-effort and read-only; nil hook = unchanged.
+	d.sendWarmer = newSendWarmer(ctx, chainAPI)
+	chainAPI.OnSentTx = d.sendWarmer.Warm
+
 	// Stash for accessor-style reads (LocalEthCallStats, etc.).
 	d.mu.Lock()
 	d.chainAPI = chainAPI