From 65fe10e2168d5fd6c38bab06d4dd435618f9eabe Mon Sep 17 00:00:00 2001 From: Reiers Date: Thu, 2 Jul 2026 16:12:06 +0200 Subject: [PATCH] =?UTF-8?q?feat(#91):=20ChainExchange=20client=20=E2=80=94?= =?UTF-8?q?=20gateway-free=20header-chain=20fetch=20over=20libp2p?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the requesting half of /fil/chain/xchg/0.0.1 (the responder landed in #17). A bridge-off Lantern (#76, --no-fallback-rpc) can now backfill a gossip head+N gap by pulling the whole verified header chain from a real Filecoin peer in ONE request, with zero upstream RPC (no Glif, no gateway). This is Stage A of the snapshot-free full-node epic (#87). net/chainxchg/client.go - Client.FetchTipsetChain(head, length): headers-only request, walks parent-ward newest->oldest. Handwritten CBOR both directions (same zero-dependency policy as the responder), cross-checked against Lotus chain/exchange cbor_gen layout. - Trust model: every header CID-verified; level 0 must equal the requested head set, each deeper level must equal the previous level's Parents, heights strictly decreasing. Caller derives head from already-verified data (a gossip block's Parents / the quorum anchor), so a malicious peer can only refuse, never splice. - Partial (101) responses accepted; NotFound (201)/GoAway (202) fail that peer and rotate to the next. Bounded response read (32 MiB), peer cap, trusted-floor-first peer ordering, observable ClientStats. net/blockingest/blockingest.go - ChainFetcher interface + SetChainFetcher: in parent-walk (bridge-off) mode the ingestor prefers ONE ChainExchange request over the per-CID bitswap walk, falling back to the walk on error (or surfacing the error when there is no per-CID source). net/chainxchg/client_test.go - Two-host libp2p harness: happy path, partial, NotFound, spliced-chain rejection, no-peers. Full suite green, build+vet+fmt clean. Stacks on feat/108-no-fallback-rpc (#76 scaffolding). Server-side chain serving remains a separate follow-up. --- net/blockingest/blockingest.go | 69 ++++++ net/chainxchg/client.go | 408 +++++++++++++++++++++++++++++++++ net/chainxchg/client_test.go | 277 ++++++++++++++++++++++ 3 files changed, 754 insertions(+) create mode 100644 net/chainxchg/client.go create mode 100644 net/chainxchg/client_test.go diff --git a/net/blockingest/blockingest.go b/net/blockingest/blockingest.go index 489e979..7ad0de9 100644 --- a/net/blockingest/blockingest.go +++ b/net/blockingest/blockingest.go @@ -90,6 +90,13 @@ type Ingestor struct { // Enabled by the daemon only when there is no upstream RPC source. parentWalk bool + // chainFetcher, when set, is the preferred gap-resolution path in + // parent-walk mode (#91): ONE ChainExchange request pulls the whole + // verified header chain from a libp2p peer, instead of per-CID + // bitswap fetches. Falls back to the per-CID walk when nil or on + // error. + chainFetcher ChainFetcher + incoming chan *ltypes.BlockMsg seen map[cid.Cid]struct{} @@ -450,6 +457,17 @@ func (g *Ingestor) markSeen(c cid.Cid) { // construction time before Run. func (g *Ingestor) SetParentWalkBackfill(on bool) { g.parentWalk = on } +// ChainFetcher pulls a verified header chain (newest->oldest, level 0 = +// the tipset whose block CIDs are head) from the p2p swarm. Implemented +// by net/chainxchg.Client (#91). +type ChainFetcher interface { + FetchTipsetChain(ctx context.Context, head []cid.Cid, length int) ([][]*ltypes.BlockHeader, error) +} + +// SetChainFetcher wires the ChainExchange client as the preferred +// parent-walk gap resolver (#91). Call before Run. +func (g *Ingestor) SetChainFetcher(f ChainFetcher) { g.chainFetcher = f } + // parentWalkBackfill resolves a head+N (N>1) gap without any RPC by walking // the target block's Parents chain via CID-addressed FetchBlock (bitswap), // descending until every CID at a level is already in the store (the seeded @@ -468,6 +486,21 @@ func (g *Ingestor) parentWalkBackfill(ctx context.Context, bh *ltypes.BlockHeade return fmt.Errorf("bridge-off backfill: gossip block @ %d has no parents", bh.Height) } + // Preferred path (#91): ONE ChainExchange request for the whole gap. + // The client CID-verifies the chain against bh.Parents, so a bad peer + // can only refuse, not splice. Fall back to the per-CID bitswap walk + // on any error. + if g.chainFetcher != nil { + if err := g.chainExchangeBackfill(ctx, bh); err == nil { + return nil + } else if g.src == nil { + // No per-CID fallback source either: surface the real error. + return fmt.Errorf("bridge-off chainxchg backfill: %w", err) + } else { + log.Debugw("chainxchg backfill failed; falling back to per-CID walk", "err", err) + } + } + bctx, cancel := context.WithTimeout(ctx, 30*time.Second) defer cancel() @@ -528,6 +561,42 @@ func (g *Ingestor) parentWalkBackfill(ctx context.Context, bh *ltypes.BlockHeade return nil } +// chainExchangeBackfill fills [storeHead+1, bh.Height-1] with one +// ChainExchange request rooted at bh.Parents (#91). The returned chain +// is newest->oldest and already CID-verified; install oldest-first so +// the store head lands on the newest backfilled tipset. +func (g *Ingestor) chainExchangeBackfill(ctx context.Context, bh *ltypes.BlockHeader) error { + curHead := g.store.HeadEpoch() + // Number of tipsets between the store head and the gossip block, + // inclusive of null-round slack: ask for the full epoch gap; the + // response naturally contains only real tipsets. + gap := int(bh.Height - 1 - curHead) + if gap < 1 { + gap = 1 + } + bctx, cancel := context.WithTimeout(ctx, 45*time.Second) + defer cancel() + chain, err := g.chainFetcher.FetchTipsetChain(bctx, bh.Parents, gap) + if err != nil { + return err + } + // Install oldest-first, skipping tipsets at/below the store head. + for i := len(chain) - 1; i >= 0; i-- { + blocks := chain[i] + if blocks[0].Height <= curHead { + continue + } + ts, terr := ltypes.NewTipSet(blocks) + if terr != nil { + return fmt.Errorf("chainxchg backfill tipset @ %d: %w", blocks[0].Height, terr) + } + if serr := g.store.SetHead(ctx, ts); serr != nil { + return fmt.Errorf("chainxchg backfill set head @ %d: %w", blocks[0].Height, serr) + } + } + return nil +} + // dedupBlocksByCID removes duplicate block headers (same CID reached via // multiple parent paths) while preserving order. func dedupBlocksByCID(in []*ltypes.BlockHeader) []*ltypes.BlockHeader { diff --git a/net/chainxchg/client.go b/net/chainxchg/client.go new file mode 100644 index 0000000..efe3244 --- /dev/null +++ b/net/chainxchg/client.go @@ -0,0 +1,408 @@ +// ChainExchange CLIENT (issue #91, Stage A of the full-node epic #87). +// +// The responder half of this package (chainxchg.go, issue #17) made us +// REACHABLE on /fil/chain/xchg/0.0.1. This file adds the requesting +// half: fetch tipset header chains from real Filecoin peers over +// libp2p, so a bridge-off Lantern (#76, --no-fallback-rpc) can seed its +// header store from the quorum anchor and backfill gossip gaps with +// ZERO upstream RPC (no Glif, no gateway). +// +// Scope (per the #91 survey): client-only, headers-only. +// - Request{Head []cid, Length, Options} — Options bit 0x1 = Headers, +// 0x2 = Messages. We only ever set Headers; the response's per- +// tipset Messages slot is then null and is skipped with a Deferred. +// - Response{Status, ErrorMessage, Chain []BSTipSet} where +// BSTipSet{Blocks []BlockHeader, Messages}. Chain[0] is the tipset +// identified by Head; entries walk BACKWARD (newest -> oldest). +// - Status 101 (Partial) is legal: a peer may return fewer tipsets +// than asked. The caller loops if it needs more. +// +// Trust model: every returned header is CID-verified. Chain[0]'s block +// CIDs must equal the requested Head CIDs exactly (set equality), and +// each deeper level must equal the previous level's Parents. The caller +// always derives Head from something already verified (a gossip block's +// Parents, or the multi-source quorum anchor), so a malicious peer +// cannot splice us onto a fabricated chain — it can only refuse. +// +// Wire format cross-checked against Lotus chain/exchange/protocol.go + +// cbor_gen.go (handwritten here to keep the dependency surface at zero, +// same policy as the responder). + +package chainxchg + +import ( + "context" + "fmt" + "io" + "math/rand" + "sync/atomic" + "time" + + "github.com/ipfs/go-cid" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/peer" + cbg "github.com/whyrusleeping/cbor-gen" + + ltypes "github.com/Reiers/lantern/chain/types" +) + +// Options bits (Lotus chain/exchange/protocol.go). +const ( + optHeaders uint64 = 1 << iota + optMessages +) + +// Additional status codes the client must understand (the responder +// side above only ever emits statusNotFound). +const ( + statusPartial uint64 = 101 + statusGoAway uint64 = 202 +) + +// MaxRequestLength is the protocol cap on tipsets per request. +const MaxRequestLength = 900 + +var ( + clientWriteDeadline = 10 * time.Second + clientReadDeadline = 45 * time.Second + // maxResponseBytes bounds a headers-only response read. Worst + // legitimate case (900 tipsets x ~10 blocks x ~2 KiB/header) is + // ~18 MiB; 32 MiB leaves headroom without letting a malicious + // peer feed us unbounded data. + maxResponseBytes int64 = 32 << 20 + // maxPeerTries bounds how many peers one FetchTipsetChain call + // will rotate through before giving up. + maxPeerTries = 8 + // maxErrMsgLen bounds the ErrorMessage string we will read. + maxErrMsgLen uint64 = 1024 + // maxBlocksPerTipset bounds decoded blocks per tipset (Filecoin + // max is 10 wincount-weighted; 32 is generous). + maxBlocksPerTipset uint64 = 32 +) + +// Client is a ChainExchange requester bound to a libp2p host. +type Client struct { + h host.Host + + // preferredPeers, when set, returns peers to try FIRST (e.g. the + // trusted floor / bootstrap quorum peers). Others follow shuffled. + preferredPeers func() []peer.ID + + requests atomic.Uint64 // FetchTipsetChain calls + succeeded atomic.Uint64 // calls that returned >=1 verified tipset + failed atomic.Uint64 // calls that exhausted all peers + peerTries atomic.Uint64 // per-peer attempts (success + failure) +} + +// NewClient constructs a ChainExchange client on h. Register() is NOT +// required; the client only opens outbound streams. +func NewClient(h host.Host) *Client { + return &Client{h: h} +} + +// SetPreferredPeers wires a provider of first-choice peers (trusted +// floor / quorum peers). Optional. +func (c *Client) SetPreferredPeers(f func() []peer.ID) { c.preferredPeers = f } + +// ClientStats reports observable activity for the dashboard. +type ClientStats struct { + Requests uint64 `json:"requests"` + Succeeded uint64 `json:"succeeded"` + Failed uint64 `json:"failed"` + PeerTries uint64 `json:"peer_tries"` +} + +// ClientStats returns a snapshot. +func (c *Client) ClientStats() ClientStats { + return ClientStats{ + Requests: c.requests.Load(), + Succeeded: c.succeeded.Load(), + Failed: c.failed.Load(), + PeerTries: c.peerTries.Load(), + } +} + +// FetchTipsetChain requests up to length tipsets of HEADERS starting at +// the tipset whose block CIDs are head, walking parent-ward. The result +// is ordered as received: index 0 = the head tipset, increasing index = +// older. Every header is CID-verified and every level is verified to be +// the parent set of the level above; the first level must match head +// exactly. Returns at least one tipset on success (partial responses +// are legal and returned as-is). +func (c *Client) FetchTipsetChain(ctx context.Context, head []cid.Cid, length int) ([][]*ltypes.BlockHeader, error) { + if len(head) == 0 { + return nil, fmt.Errorf("chainxchg client: empty head") + } + if length < 1 { + length = 1 + } + if length > MaxRequestLength { + length = MaxRequestLength + } + c.requests.Add(1) + + peers := c.candidatePeers() + if len(peers) == 0 { + c.failed.Add(1) + return nil, fmt.Errorf("chainxchg client: no connected peers support %s", ProtocolID) + } + if len(peers) > maxPeerTries { + peers = peers[:maxPeerTries] + } + + var lastErr error + for _, p := range peers { + if ctx.Err() != nil { + c.failed.Add(1) + return nil, ctx.Err() + } + c.peerTries.Add(1) + chain, err := c.fetchFromPeer(ctx, p, head, length) + if err != nil { + lastErr = err + log.Debugw("chainxchg client: peer failed", "peer", p, "err", err) + continue + } + c.succeeded.Add(1) + return chain, nil + } + c.failed.Add(1) + return nil, fmt.Errorf("chainxchg client: all %d peers failed, last: %w", len(peers), lastErr) +} + +// candidatePeers returns connected peers that advertise the protocol, +// preferred peers first, the rest shuffled. +func (c *Client) candidatePeers() []peer.ID { + connected := c.h.Network().Peers() + supports := make(map[peer.ID]bool, len(connected)) + var pool []peer.ID + for _, p := range connected { + if p == c.h.ID() { + continue + } + protos, err := c.h.Peerstore().SupportsProtocols(p, ProtocolID) + if err != nil || len(protos) == 0 { + continue + } + supports[p] = true + pool = append(pool, p) + } + rand.Shuffle(len(pool), func(i, j int) { pool[i], pool[j] = pool[j], pool[i] }) + + if c.preferredPeers == nil { + return pool + } + seen := make(map[peer.ID]bool, len(pool)) + var out []peer.ID + for _, p := range c.preferredPeers() { + if supports[p] && !seen[p] { + out = append(out, p) + seen[p] = true + } + } + for _, p := range pool { + if !seen[p] { + out = append(out, p) + seen[p] = true + } + } + return out +} + +// fetchFromPeer performs one request/response exchange with one peer +// and fully verifies the returned chain. +func (c *Client) fetchFromPeer(ctx context.Context, p peer.ID, head []cid.Cid, length int) ([][]*ltypes.BlockHeader, error) { + str, err := c.h.NewStream(ctx, p, ProtocolID) + if err != nil { + return nil, fmt.Errorf("open stream: %w", err) + } + defer func() { _ = str.Close() }() + + _ = str.SetWriteDeadline(time.Now().Add(clientWriteDeadline)) + if err := writeRequest(str, head, uint64(length), optHeaders); err != nil { + _ = str.Reset() + return nil, fmt.Errorf("write request: %w", err) + } + // Half-close so the remote sees EOF on its read (standard Lotus + // client behaviour; our own responder relies on it too). + _ = str.CloseWrite() + _ = str.SetWriteDeadline(time.Time{}) + + _ = str.SetReadDeadline(time.Now().Add(clientReadDeadline)) + defer func() { _ = str.SetReadDeadline(time.Time{}) }() + + chain, status, errMsg, err := readResponse(io.LimitReader(str, maxResponseBytes)) + if err != nil { + _ = str.Reset() + return nil, fmt.Errorf("read response: %w", err) + } + switch status { + case statusOK, statusPartial: + // fine + case statusNotFound: + return nil, fmt.Errorf("peer does not have tipset (NotFound): %s", errMsg) + case statusGoAway: + return nil, fmt.Errorf("peer sent GoAway: %s", errMsg) + default: + return nil, fmt.Errorf("peer returned status %d: %s", status, errMsg) + } + if len(chain) == 0 { + return nil, fmt.Errorf("peer returned OK with empty chain") + } + if err := verifyChain(head, chain); err != nil { + return nil, fmt.Errorf("chain verification: %w", err) + } + return chain, nil +} + +// verifyChain enforces the trust model: level 0 == requested head CIDs, +// level i+1 == level i's Parents, heights strictly decreasing, and all +// blocks within a level share the same Parents. +func verifyChain(head []cid.Cid, chain [][]*ltypes.BlockHeader) error { + expected := cidSet(head) + var prevHeight int64 = -1 + for i, blocks := range chain { + if len(blocks) == 0 { + return fmt.Errorf("level %d: empty tipset", i) + } + got := make(map[cid.Cid]bool, len(blocks)) + for _, b := range blocks { + got[b.Cid()] = true + } + if len(got) != len(expected) { + return fmt.Errorf("level %d: got %d distinct blocks, expected %d", i, len(got), len(expected)) + } + for c := range expected { + if !got[c] { + return fmt.Errorf("level %d: missing expected block %s", i, c) + } + } + h := int64(blocks[0].Height) + if prevHeight >= 0 && h >= prevHeight { + return fmt.Errorf("level %d: height %d not below previous %d", i, h, prevHeight) + } + prevHeight = h + parents := cidSet(blocks[0].Parents) + for _, b := range blocks[1:] { + if !sameCidSet(parents, cidSet(b.Parents)) { + return fmt.Errorf("level %d: blocks disagree on parents", i) + } + } + expected = parents + } + return nil +} + +func cidSet(cids []cid.Cid) map[cid.Cid]bool { + m := make(map[cid.Cid]bool, len(cids)) + for _, c := range cids { + m[c] = true + } + return m +} + +func sameCidSet(a, b map[cid.Cid]bool) bool { + if len(a) != len(b) { + return false + } + for c := range a { + if !b[c] { + return false + } + } + return true +} + +// writeRequest emits Request{Head, Length, Options} as CBOR array(3), +// matching Lotus exchange cbor_gen.go layout. +func writeRequest(w io.Writer, head []cid.Cid, length, options uint64) error { + cw := cbg.NewCborWriter(w) + if err := cw.WriteMajorTypeHeader(cbg.MajArray, 3); err != nil { + return err + } + if err := cw.WriteMajorTypeHeader(cbg.MajArray, uint64(len(head))); err != nil { + return err + } + for _, c := range head { + if err := cbg.WriteCid(cw, c); err != nil { + return err + } + } + if err := cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, length); err != nil { + return err + } + return cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, options) +} + +// readResponse decodes Response{Status, ErrorMessage, Chain []BSTipSet} +// keeping only headers. Each BSTipSet is array(2){Blocks, Messages}; +// Messages is consumed with a Deferred (null for headers-only, but a +// peer that sends messages anyway is tolerated and skipped). +func readResponse(r io.Reader) (chain [][]*ltypes.BlockHeader, status uint64, errMsg string, err error) { + cr := cbg.NewCborReader(r) + + maj, n, err := cr.ReadHeader() + if err != nil { + return nil, 0, "", fmt.Errorf("response envelope: %w", err) + } + if maj != cbg.MajArray || n != 3 { + return nil, 0, "", fmt.Errorf("response envelope: want array(3), got maj=%d n=%d", maj, n) + } + + maj, status, err = cr.ReadHeader() + if err != nil || maj != cbg.MajUnsignedInt { + return nil, 0, "", fmt.Errorf("response status: maj=%d err=%v", maj, err) + } + + maj, slen, err := cr.ReadHeader() + if err != nil || maj != cbg.MajTextString { + return nil, 0, "", fmt.Errorf("response errmsg header: maj=%d err=%v", maj, err) + } + if slen > maxErrMsgLen { + return nil, 0, "", fmt.Errorf("response errmsg too long: %d", slen) + } + if slen > 0 { + buf := make([]byte, slen) + if _, err := io.ReadFull(cr, buf); err != nil { + return nil, 0, "", fmt.Errorf("response errmsg body: %w", err) + } + errMsg = string(buf) + } + + maj, tn, err := cr.ReadHeader() + if err != nil || maj != cbg.MajArray { + return nil, 0, "", fmt.Errorf("response chain header: maj=%d err=%v", maj, err) + } + if tn > MaxRequestLength { + return nil, 0, "", fmt.Errorf("response chain too long: %d", tn) + } + chain = make([][]*ltypes.BlockHeader, 0, tn) + for i := uint64(0); i < tn; i++ { + maj, fn, err := cr.ReadHeader() + if err != nil || maj != cbg.MajArray || fn != 2 { + return nil, 0, "", fmt.Errorf("bstipset %d: want array(2), got maj=%d n=%d err=%v", i, maj, fn, err) + } + maj, bn, err := cr.ReadHeader() + if err != nil || maj != cbg.MajArray { + return nil, 0, "", fmt.Errorf("bstipset %d blocks header: maj=%d err=%v", i, maj, err) + } + if bn == 0 || bn > maxBlocksPerTipset { + return nil, 0, "", fmt.Errorf("bstipset %d: bad block count %d", i, bn) + } + blocks := make([]*ltypes.BlockHeader, 0, bn) + for j := uint64(0); j < bn; j++ { + bh := new(ltypes.BlockHeader) + if err := bh.UnmarshalCBOR(cr); err != nil { + return nil, 0, "", fmt.Errorf("bstipset %d block %d: %w", i, j, err) + } + blocks = append(blocks, bh) + } + // Messages: null for headers-only; tolerate anything and skip. + var d cbg.Deferred + if err := d.UnmarshalCBOR(cr); err != nil { + return nil, 0, "", fmt.Errorf("bstipset %d messages skip: %w", i, err) + } + chain = append(chain, blocks) + } + return chain, status, errMsg, nil +} diff --git a/net/chainxchg/client_test.go b/net/chainxchg/client_test.go new file mode 100644 index 0000000..e7d89d5 --- /dev/null +++ b/net/chainxchg/client_test.go @@ -0,0 +1,277 @@ +// Tests for the ChainExchange client (issue #91). +// +// A pair of real libp2p hosts: the "server" host runs a handwritten +// responder that returns canned Response CBOR built from real +// ltypes.BlockHeader values; the client fetches and must CID-verify the +// chain. Covers: happy path, partial responses, NotFound, and a +// malicious peer returning a spliced (wrong-head) chain. + +package chainxchg + +import ( + "bytes" + "context" + "io" + "testing" + "time" + + "github.com/filecoin-project/go-address" + abi "github.com/filecoin-project/go-state-types/abi" + "github.com/ipfs/go-cid" + "github.com/libp2p/go-libp2p" + "github.com/libp2p/go-libp2p/core/host" + "github.com/libp2p/go-libp2p/core/network" + "github.com/libp2p/go-libp2p/core/peer" + mh "github.com/multiformats/go-multihash" + "github.com/stretchr/testify/require" + cbg "github.com/whyrusleeping/cbor-gen" + + ltypes "github.com/Reiers/lantern/chain/types" +) + +func mkTestCID(t *testing.T, s string) cid.Cid { + t.Helper() + hash, err := mh.Sum([]byte(s), mh.SHA2_256, -1) + require.NoError(t, err) + return cid.NewCidV1(cid.DagCBOR, hash) +} + +func mkTestBlock(t *testing.T, h abi.ChainEpoch, parents []cid.Cid, tag string) *ltypes.BlockHeader { + t.Helper() + a, err := address.NewIDAddress(1000) + require.NoError(t, err) + return <ypes.BlockHeader{ + Miner: a, + Ticket: <ypes.Ticket{VRFProof: []byte("t-" + tag)}, + ElectionProof: <ypes.ElectionProof{WinCount: 1, VRFProof: []byte("e-" + tag)}, + Parents: parents, + ParentWeight: ltypes.NewInt(uint64(h)), + Height: h, + ParentStateRoot: mkTestCID(t, "state-"+tag), + ParentMessageReceipts: mkTestCID(t, "rcpt-"+tag), + Messages: mkTestCID(t, "msgs-"+tag), + Timestamp: 1_700_000_000 + uint64(h)*30, + ParentBaseFee: ltypes.NewInt(100), + } +} + +// mkChain builds n single-block tipsets, newest-first (chain[0] highest), +// linked via Parents. +func mkChain(t *testing.T, n int, topHeight abi.ChainEpoch) [][]*ltypes.BlockHeader { + t.Helper() + blocks := make([]*ltypes.BlockHeader, n) + // Build oldest -> newest so parents link. + prevParents := []cid.Cid{mkTestCID(t, "genesis")} + for i := n - 1; i >= 0; i-- { + h := topHeight - abi.ChainEpoch(i) + b := mkTestBlock(t, h, prevParents, "c"+string(rune('a'+i))) + blocks[i] = b + prevParents = []cid.Cid{b.Cid()} + } + chain := make([][]*ltypes.BlockHeader, n) + for i := 0; i < n; i++ { + chain[i] = []*ltypes.BlockHeader{blocks[i]} + } + return chain +} + +// encodeResponse writes Response{status, errMsg, chain} the way Lotus +// cbor_gen would (Messages slot = null). +func encodeResponse(t *testing.T, status uint64, errMsg string, chain [][]*ltypes.BlockHeader) []byte { + t.Helper() + var buf bytes.Buffer + cw := cbg.NewCborWriter(&buf) + require.NoError(t, cw.WriteMajorTypeHeader(cbg.MajArray, 3)) + require.NoError(t, cw.WriteMajorTypeHeader(cbg.MajUnsignedInt, status)) + require.NoError(t, cw.WriteMajorTypeHeader(cbg.MajTextString, uint64(len(errMsg)))) + if errMsg != "" { + _, err := cw.Write([]byte(errMsg)) + require.NoError(t, err) + } + require.NoError(t, cw.WriteMajorTypeHeader(cbg.MajArray, uint64(len(chain)))) + for _, blocks := range chain { + require.NoError(t, cw.WriteMajorTypeHeader(cbg.MajArray, 2)) + require.NoError(t, cw.WriteMajorTypeHeader(cbg.MajArray, uint64(len(blocks)))) + for _, b := range blocks { + require.NoError(t, b.MarshalCBOR(cw)) + } + _, err := cw.Write(cbg.CborNull) + require.NoError(t, err) + } + return buf.Bytes() +} + +// serveCanned registers a handler on h that drains the request and +// writes raw. It also captures the decoded request for assertions. +type cannedServer struct { + raw []byte + gotHead []cid.Cid + gotLen uint64 + gotOpts uint64 +} + +func (cs *cannedServer) register(t *testing.T, h host.Host) { + t.Helper() + h.SetStreamHandler(ProtocolID, func(str network.Stream) { + defer func() { _ = str.Close() }() + // Decode the request so tests can assert on it. + cr := cbg.NewCborReader(str) + maj, n, err := cr.ReadHeader() + if err != nil || maj != cbg.MajArray || n != 3 { + _ = str.Reset() + return + } + maj, hn, err := cr.ReadHeader() + if err != nil || maj != cbg.MajArray { + _ = str.Reset() + return + } + for i := uint64(0); i < hn; i++ { + c, err := cbg.ReadCid(cr) + if err != nil { + _ = str.Reset() + return + } + cs.gotHead = append(cs.gotHead, c) + } + _, cs.gotLen, _ = cr.ReadHeader() + _, cs.gotOpts, _ = cr.ReadHeader() + // Drain until EOF (client half-closes). + _, _ = io.Copy(io.Discard, str) + _, _ = str.Write(cs.raw) + }) +} + +// hostPair spins up two connected hosts and waits until the client side +// sees the server advertise the protocol (identify propagation). +func hostPair(t *testing.T) (client host.Host, server host.Host) { + t.Helper() + c, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + require.NoError(t, err) + t.Cleanup(func() { _ = c.Close() }) + s, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + require.NoError(t, c.Connect(context.Background(), peer.AddrInfo{ID: s.ID(), Addrs: s.Addrs()})) + // Wait for identify to propagate protocol support. + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + protos, err := c.Peerstore().SupportsProtocols(s.ID(), ProtocolID) + if err == nil && len(protos) > 0 { + return c, s + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("identify never propagated protocol support") + return nil, nil +} + +func TestClient_FetchChain_HappyPath(t *testing.T) { + chain := mkChain(t, 3, 100) + head := []cid.Cid{chain[0][0].Cid()} + + // Register the responder on the server host FIRST so identify + // advertises the protocol. + cs := &cannedServer{raw: encodeResponse(t, statusOK, "", chain)} + chost, shost := func() (host.Host, host.Host) { + c, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + require.NoError(t, err) + t.Cleanup(func() { _ = c.Close() }) + s, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + return c, s + }() + cs.register(t, shost) + require.NoError(t, chost.Connect(context.Background(), peer.AddrInfo{ID: shost.ID(), Addrs: shost.Addrs()})) + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + protos, err := chost.Peerstore().SupportsProtocols(shost.ID(), ProtocolID) + if err == nil && len(protos) > 0 { + break + } + time.Sleep(20 * time.Millisecond) + } + + cl := NewClient(chost) + got, err := cl.FetchTipsetChain(context.Background(), head, 3) + require.NoError(t, err) + require.Len(t, got, 3) + require.Equal(t, chain[0][0].Cid(), got[0][0].Cid()) + require.Equal(t, chain[2][0].Cid(), got[2][0].Cid()) + // Request wire assertions. + require.Equal(t, head, cs.gotHead) + require.Equal(t, uint64(3), cs.gotLen) + require.Equal(t, optHeaders, cs.gotOpts) + st := cl.ClientStats() + require.Equal(t, uint64(1), st.Succeeded) + require.Equal(t, uint64(0), st.Failed) +} + +func TestClient_PartialAccepted(t *testing.T) { + chain := mkChain(t, 3, 200) + head := []cid.Cid{chain[0][0].Cid()} + cs := &cannedServer{raw: encodeResponse(t, statusPartial, "", chain[:1])} + chost, shost := hostPairWith(t, cs) + cl := NewClient(chost) + got, err := cl.FetchTipsetChain(context.Background(), head, 3) + require.NoError(t, err) + require.Len(t, got, 1) + _ = shost +} + +func TestClient_NotFoundFails(t *testing.T) { + cs := &cannedServer{raw: encodeResponse(t, statusNotFound, "nope", nil)} + chost, _ := hostPairWith(t, cs) + cl := NewClient(chost) + _, err := cl.FetchTipsetChain(context.Background(), []cid.Cid{mkTestCID(t, "whatever")}, 2) + require.Error(t, err) + require.Contains(t, err.Error(), "NotFound") + require.Equal(t, uint64(1), cl.ClientStats().Failed) +} + +func TestClient_SplicedChainRejected(t *testing.T) { + // Server returns a valid-looking chain that does NOT match the + // requested head — must be rejected by verification. + chain := mkChain(t, 2, 300) + cs := &cannedServer{raw: encodeResponse(t, statusOK, "", chain)} + chost, _ := hostPairWith(t, cs) + cl := NewClient(chost) + _, err := cl.FetchTipsetChain(context.Background(), []cid.Cid{mkTestCID(t, "different-head")}, 2) + require.Error(t, err) + require.Contains(t, err.Error(), "verification") +} + +func TestClient_NoPeers(t *testing.T) { + h, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + require.NoError(t, err) + t.Cleanup(func() { _ = h.Close() }) + cl := NewClient(h) + _, err = cl.FetchTipsetChain(context.Background(), []cid.Cid{mkTestCID(t, "x")}, 1) + require.Error(t, err) + require.Contains(t, err.Error(), "no connected peers") +} + +// hostPairWith wires a canned server + connected client host, waiting +// for identify. +func hostPairWith(t *testing.T, cs *cannedServer) (host.Host, host.Host) { + t.Helper() + c, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + require.NoError(t, err) + t.Cleanup(func() { _ = c.Close() }) + s, err := libp2p.New(libp2p.ListenAddrStrings("/ip4/127.0.0.1/tcp/0")) + require.NoError(t, err) + t.Cleanup(func() { _ = s.Close() }) + cs.register(t, s) + require.NoError(t, c.Connect(context.Background(), peer.AddrInfo{ID: s.ID(), Addrs: s.Addrs()})) + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + protos, err := c.Peerstore().SupportsProtocols(s.ID(), ProtocolID) + if err == nil && len(protos) > 0 { + return c, s + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("identify never propagated protocol support") + return nil, nil +}