From 3dced0821a4a31795d02f08af70872900a123f20 Mon Sep 17 00:00:00 2001 From: Reiers Date: Mon, 22 Jun 2026 01:03:10 +0200 Subject: [PATCH 1/2] =?UTF-8?q?feat(#50):=20prefetch-on-send=20=E2=80=94?= =?UTF-8?q?=20warm=20sent-tx=20blocks=20into=20bitswap=20cache=20(UN-DEPLO?= =?UTF-8?q?YED,=20needs=20supervised=20bridge-off=20verify)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When eth_sendRawTransaction publishes a tx locally (#45 Stage 4), start a background StateSearchMsg poll for its message CID on a standalone, generous-deadline context. That background search drives the embedded Bitswap source to pull the freshly-produced message + AMT + receipt blocks into cache as soon as the tx lands. By the time the client's own short- deadline receipt poll runs, the blocks are warm and the receipt resolves locally instead of racing (and losing to) a cold cross-peer Bitswap fetch inside the poll window — the residual that kept #50 open. - new ChainAPI.OnSentTx hook, fired non-blocking after a successful local MpoolPush (mirrors the #44 OnLocalMiss adaptive-warming pattern) - pkg/daemon sendWarmer: per-tx background warm loop, dedup, concurrency cap (64), self-bounded (stop on found / 10m max), bound to daemon root ctx - purely additive + read-only: nil hook = unchanged behavior; never affects the send result; never publishes or mutates state - 5 unit tests (stop-on-found, dedup, concurrency cap, ctx-cancel, nil-safe), pass under -race; full pkg/daemon + rpc/handlers suites green NOT deployed to cc-smoke. Needs live bridge-off verification in a supervised daytime window before merge (the engine builds + unit-tests clean, but the real proof is a bridge-off write-confirm cycle showing zero 'net/bitswap: context canceled' receipt misses). --- pkg/daemon/daemon.go | 6 + pkg/daemon/sendwarm.go | 185 +++++++++++++++++++++++++++++ pkg/daemon/sendwarm_test.go | 178 +++++++++++++++++++++++++++ pkg/daemon/start.go | 8 ++ rpc/handlers/chain_api.go | 13 ++ rpc/handlers/extra_writepath_tx.go | 7 ++ 6 files changed, 397 insertions(+) create mode 100644 pkg/daemon/sendwarm.go create mode 100644 pkg/daemon/sendwarm_test.go diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index 2973976..1787fa2 100644 --- a/pkg/daemon/daemon.go +++ b/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 } diff --git a/pkg/daemon/sendwarm.go b/pkg/daemon/sendwarm.go new file mode 100644 index 0000000..676760b --- /dev/null +++ b/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 diff --git a/pkg/daemon/sendwarm_test.go b/pkg/daemon/sendwarm_test.go new file mode 100644 index 0000000..d29217e --- /dev/null +++ b/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 +} diff --git a/pkg/daemon/start.go b/pkg/daemon/start.go index c77bbdd..ea80f74 100644 --- a/pkg/daemon/start.go +++ b/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 diff --git a/rpc/handlers/chain_api.go b/rpc/handlers/chain_api.go index 17e236b..4b24688 100644 --- a/rpc/handlers/chain_api.go +++ b/rpc/handlers/chain_api.go @@ -102,6 +102,19 @@ type ChainAPI struct { // cheap + non-blocking; the handler calls it on the hot path. OnLocalMiss func(addr string) + // OnSentTx, when set, is invoked (non-blocking, on the send hot path) + // with the Filecoin message CID every time eth_sendRawTransaction + // publishes a tx locally (lantern#50 prefetch-on-send). curio-core / + // the daemon wires this to a background warmer that polls + // StateSearchMsg for the message until it lands, which drives the + // embedded Bitswap source to pull the freshly-produced message + AMT + + // receipt blocks into cache. By the time the client's own receipt poll + // runs, those blocks are warm, so the receipt resolves locally instead + // of racing (and losing to) a cold cross-peer Bitswap fetch inside the + // poll window — the residual that kept #50 open. Must be cheap + + // non-blocking; the handler fires it in a goroutine-friendly way. + OnSentTx func(msgCID cid.Cid) + // BeaconParams is the drand-round mapping for the active network. // Defaults to mainnet quicknet if zero-value. BeaconParams lbeacon.QuicknetParams diff --git a/rpc/handlers/extra_writepath_tx.go b/rpc/handlers/extra_writepath_tx.go index 49f1cac..1a3ec6b 100644 --- a/rpc/handlers/extra_writepath_tx.go +++ b/rpc/handlers/extra_writepath_tx.go @@ -153,6 +153,13 @@ func (c *ChainAPI) EthSendRawTransaction(ctx context.Context, signedTxHex string c.sentTx().put(strings.ToLower(hashHex), sentTxRecord{msgCID: msgCID, tx: tx, from: sender}) log.Infow("eth_sendRawTransaction: published locally", "ethHash", hashHex, "msgCID", msgCID, "from", smsg.Message.From) + // lantern#50: kick off background warming of this message's blocks so + // the receipt poll hits a warm Bitswap cache instead of racing a cold + // cross-peer fetch. Non-blocking and best-effort: a nil hook (or a + // slow warmer) never affects the send result. + if c.OnSentTx != nil { + c.OnSentTx(msgCID) + } return hashHex, nil } From 36c2b0e55910057bd4a1862af545bb337dd9c5ef Mon Sep 17 00:00:00 2001 From: Reiers Date: Wed, 24 Jun 2026 20:09:01 +0200 Subject: [PATCH 2/2] fmt: gofmt sendwarm.go struct field alignment (fixes PR #52 CI) Formatting-only; no behavioral change. The branch CI failed on gofmt drift (struct field alignment in sendWarmer). No logic touched. --- pkg/daemon/sendwarm.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/daemon/sendwarm.go b/pkg/daemon/sendwarm.go index 676760b..8ab7b53 100644 --- a/pkg/daemon/sendwarm.go +++ b/pkg/daemon/sendwarm.go @@ -71,8 +71,8 @@ type sendWarmer struct { // 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 + pollInterval time.Duration + maxDuration time.Duration maxConcurrent int mu sync.Mutex