From 058f14c7830985813fe9cd7bf2f3e2d4772bb9fe Mon Sep 17 00:00:00 2001 From: Reiers Date: Wed, 1 Jul 2026 22:27:48 +0200 Subject: [PATCH 1/2] feat(#97): Lotus per-topic gossipsub score params + RawTracer/trusted-floor plumbing The per-topic score map was empty: a peer that never usefully delivered anything scored the same as one that consistently delivered blocks first. /fil/blocks + /fil/msgs (mainnet + calibration) now carry Lotus's exact TopicScoreParams (first-delivery credit, invalid-message penalties), which is what makes #80's distinct-SCORED-peers corroboration bite against Sybil swarms. Also plumbs (for #80 part 2): - HostConfig.PubSubTracer -> pubsub.WithRawTracer at construction - Host.IsTrustedPeer: is p in the connmgr-protected trusted floor - trustedFloorTag const replaces the inline tag string --- net/libp2p/host.go | 27 ++++++++++++++-- net/libp2p/pubsub.go | 75 ++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 98 insertions(+), 4 deletions(-) diff --git a/net/libp2p/host.go b/net/libp2p/host.go index 87cee2f..5cd8790 100644 --- a/net/libp2p/host.go +++ b/net/libp2p/host.go @@ -68,6 +68,10 @@ type HostConfig struct { // new heads propagate at the same speed they do for those nodes. // Defaults to BootstrapPeers when empty. GossipSubDirectPeers []string + // PubSubTracer, when set, is registered as gossipsub's RawTracer at + // construction. Used by the head-corroboration tracker (#80 part 2) + // so duplicate block deliveries are counted per source peer. + PubSubTracer pubsub.RawTracer } // Host wraps a libp2p Host + GossipSub PubSub instance. @@ -214,7 +218,7 @@ func New(ctx context.Context, cfg HostConfig) (*Host, error) { if len(directPeers) == 0 { directPeers = cfg.BootstrapPeers } - ps, err := newFilecoinPubSub(ctx, h, directPeers) + ps, err := newFilecoinPubSub(ctx, h, directPeers, cfg.PubSubTracer) if err != nil { _ = h.Close() return nil, fmt.Errorf("pubsub.NewGossipSub: %w", err) @@ -250,7 +254,7 @@ func New(ctx context.Context, cfg HostConfig) (*Host, error) { if len(floor) == 0 { floor = cfg.BootstrapPeers } - out.ProtectPeers(floor, "lantern-trusted-floor") + out.ProtectPeers(floor, trustedFloorTag) // Background dial of bootstrap peers (non-blocking). if len(cfg.BootstrapPeers) > 0 { @@ -260,6 +264,10 @@ func New(ctx context.Context, cfg HostConfig) (*Host, error) { return out, nil } +// trustedFloorTag is the connmgr protection tag for the trusted +// bootstrap/beacon/direct-peer floor (#80). +const trustedFloorTag = "lantern-trusted-floor" + // ProtectPeers marks the given multiaddr peers as connmgr-protected under // tag, making them an un-evictable floor the connection-manager trim path // will never drop (#80). Used for the trusted bootstrap/beacon/direct-peer @@ -290,6 +298,21 @@ func (h *Host) ProtectPeers(peers []string, tag string) int { return n } +// IsTrustedPeer reports whether p is part of the protected trusted floor +// (bootstrap/beacon/direct peers pinned via ProtectPeers, #80). The head +// corroboration gate uses this as the super-vote: a head forwarded by a +// trusted floor peer is corroborated regardless of source count. +func (h *Host) IsTrustedPeer(p peer.ID) bool { + if h == nil || h.H == nil || p == "" { + return false + } + cm := h.H.ConnManager() + if cm == nil { + return false + } + return cm.IsProtected(p, trustedFloorTag) +} + // consumeReachability mirrors EvtLocalReachabilityChanged into the cached // atomic so NetAutoNatStatus is a lock-free read. func (h *Host) consumeReachability(ctx context.Context, sub event.Subscription) { diff --git a/net/libp2p/pubsub.go b/net/libp2p/pubsub.go index 458030f..35b1fc3 100644 --- a/net/libp2p/pubsub.go +++ b/net/libp2p/pubsub.go @@ -20,6 +20,8 @@ import ( pubsub "github.com/libp2p/go-libp2p-pubsub" pubsub_pb "github.com/libp2p/go-libp2p-pubsub/pb" + + "github.com/Reiers/lantern/build" "github.com/libp2p/go-libp2p/core/host" "github.com/libp2p/go-libp2p/core/peer" "github.com/multiformats/go-multiaddr" @@ -85,7 +87,71 @@ func filecoinPeerScoreParams() *pubsub.PeerScoreParams { RetainScore: 6 * time.Hour, - Topics: make(map[string]*pubsub.TopicScoreParams), + Topics: filecoinTopicScoreParams(), + } +} + +// filecoinTopicScoreParams returns Lotus's per-topic score params for the +// blocks + msgs topics (#97). Without these, a peer that never usefully +// delivers anything scores the same as one that consistently delivers +// blocks first; WITH them, first-delivery history accumulates positive +// score for honest long-lived peers while a freshly-dialed Sybil swarm +// scores ~0 and decays out of the mesh. This is what makes the #80 +// "distinct scored peers" head corroboration meaningful. +// +// Keys must be exact topic strings, so we register params for every +// network's topics; only joined topics are consulted by gossipsub, so +// the unused network's entries are inert. +// +// Values copied from lotus/node/modules/lp2p/pubsub.go (mesh-delivery +// failure penalties stay off there too - the network is too small for +// meaningful incoming-edge distribution; revisit when Lotus does). +func filecoinTopicScoreParams() map[string]*pubsub.TopicScoreParams { + blocks := func() *pubsub.TopicScoreParams { + return &pubsub.TopicScoreParams{ + // expected 10 blocks/min + TopicWeight: 0.1, // max cap is 50, single invalid message is -100 + + // 1 tick per second, maxes at 1 after 1 hour + TimeInMeshWeight: 0.00027, // ~1/3600 + TimeInMeshQuantum: time.Second, + TimeInMeshCap: 1, + + // deliveries decay after 1 hour, cap at 100 blocks + FirstMessageDeliveriesWeight: 5, // max value is 500 + FirstMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour), + FirstMessageDeliveriesCap: 100, // 100 blocks in an hour + + // invalid messages decay after 1 hour + InvalidMessageDeliveriesWeight: -1000, + InvalidMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour), + } + } + msgs := func() *pubsub.TopicScoreParams { + return &pubsub.TopicScoreParams{ + // expected > 1 tx/second + TopicWeight: 0.1, // max cap is 5, single invalid message is -100 + + // 1 tick per second, maxes at 1 hour + TimeInMeshWeight: 0.0002778, // ~1/3600 + TimeInMeshQuantum: time.Second, + TimeInMeshCap: 1, + + // deliveries decay after 10min, cap at 100 tx + FirstMessageDeliveriesWeight: 0.5, // max value is 50 + FirstMessageDeliveriesDecay: pubsub.ScoreParameterDecay(10 * time.Minute), + FirstMessageDeliveriesCap: 100, // 100 messages in 10 minutes + + // invalid messages decay after 1 hour + InvalidMessageDeliveriesWeight: -1000, + InvalidMessageDeliveriesDecay: pubsub.ScoreParameterDecay(time.Hour), + } + } + return map[string]*pubsub.TopicScoreParams{ + build.MainnetGossipTopicBlocks: blocks(), + build.CalibnetGossipTopicBlocks: blocks(), + build.MainnetGossipTopicMessages: msgs(), + build.CalibnetGossipTopicMessages: msgs(), } } @@ -107,7 +173,7 @@ func filecoinPeerScoreThresholds() *pubsub.PeerScoreThresholds { // Direct peers must already be connected at the libp2p layer; we read // their addresses from the host's peerstore where possible, and fall back // to multiaddr parsing for any that haven't been seen. -func newFilecoinPubSub(ctx context.Context, h host.Host, directPeerAddrs []string) (*pubsub.PubSub, error) { +func newFilecoinPubSub(ctx context.Context, h host.Host, directPeerAddrs []string, tracer pubsub.RawTracer) (*pubsub.PubSub, error) { applyFilecoinGossipSubGlobals() // Resolve direct peer multiaddrs to peer.AddrInfo. @@ -134,6 +200,11 @@ func newFilecoinPubSub(ctx context.Context, h host.Host, directPeerAddrs []strin if len(direct) > 0 { opts = append(opts, pubsub.WithDirectPeers(direct)) } + if tracer != nil { + // #80 part 2: the corroboration tracker rides gossipsub's raw + // tracer so head-source counting sees every duplicate delivery. + opts = append(opts, pubsub.WithRawTracer(tracer)) + } return pubsub.NewGossipSub(ctx, h, opts...) } From 795ab7e2e60b59e9260f08274ff37bb6b11ac183 Mon Sep 17 00:00:00 2001 From: Reiers Date: Wed, 1 Jul 2026 22:27:48 +0200 Subject: [PATCH 2/2] feat(#80): head-source corroboration - N distinct scored peers before head adoption Part 2 of #80 (part 1 = un-evictable trusted floor, shipped v1.8.5). Gossipsub dedups messages before subscribers see them, so corroboration is counted where every copy IS visible: blockpub.CorroborationTracker rides the pubsub RawTracer, recording the forwarding peer of the first delivery + every duplicate, per block header CID (CID binds the header bytes, so a peer cannot vote for block H without sending H). Adoption rule (blockpub.CorroborationGate, wired into the ingestor as SetHeadCorroboration at the same chokepoint as the #79 divergence gate): adopt head H only when - forwarders(H) >= N distinct peers, OR - a trusted floor peer forwarded H (super-vote). The requirement clamps to the connected-peer count (floor 1) so a small node never wedges. A held block is retried 3x1.5s (duplicates arrive within ~1s on a healthy mesh) then dropped - the polling Sync stays the safety net, so corroboration can delay adoption but never wedge it. Defaults: Light/PDP off; Full tier 2 distinct sources. Standalone: --head-corroboration-peers (-1 auto). Embedded: Config. HeadCorroborationPeers. New held_uncorrob stat (+ rejected_lighter / held_diverged now exposed) on the dashboard dev page. Honesty boundary (unchanged): peer IDs are Sybil-cheap; this bites in combination with #97 topic scoring + the trusted floor. Not a finality guarantee - that is F3's job. Tests: tracker distinct/dedup/topic-filter/ eviction, gate thresholds/super-vote/clamp/disabled, ingestor hold/ retry-adopt/exhaustion/nil-gate. --- CHANGELOG.md | 30 ++++ cmd/lantern/dashboard.go | 3 + cmd/lantern/main.go | 34 +++++ net/blockingest/blockingest.go | 99 +++++++++++-- net/blockingest/blockingest_test.go | 94 +++++++++++++ net/blockpub/corroboration.go | 206 ++++++++++++++++++++++++++++ net/blockpub/corroboration_test.go | 198 ++++++++++++++++++++++++++ pkg/daemon/daemon.go | 9 ++ pkg/daemon/start.go | 20 +++ 9 files changed, 685 insertions(+), 8 deletions(-) create mode 100644 net/blockpub/corroboration.go create mode 100644 net/blockpub/corroboration_test.go diff --git a/CHANGELOG.md b/CHANGELOG.md index 667daae..6c601b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,36 @@ All notable changes to Lantern. +## Unreleased (v1.9.0) + +### Security / head-path hardening + +- **Head-source corroboration** ([#80](https://github.com/Reiers/lantern/issues/80) part 2). + The ingestor now (optionally) requires a candidate head to have been + forwarded by **N distinct gossip peers** before adopting it. Gossipsub + deduplicates messages before subscribers see them, so the corroboration + votes are counted where every copy IS visible: a pubsub `RawTracer` + records the forwarding peer of the first delivery and of every duplicate, + per block CID (`blockpub.CorroborationTracker`). A trusted floor peer + (the connmgr-protected bootstrap/beacon/direct set from #80 part 1) + counts as a super-vote, and the requirement clamps to the connected-peer + count so a small node never wedges; an uncorroborated block is retried + briefly (duplicates arrive within ~1s on a healthy mesh), then left to + the polling Sync safety net. Off by default on Light/PDP; the Full tier + defaults to 2 distinct sources (`--head-corroboration-peers`, embedded: + `Config.HeadCorroborationPeers`). New `held_uncorrob` ingestor stat on + the dashboard dev page (plus previously-invisible `rejected_lighter` / + `held_diverged`). +- **Per-topic gossipsub score parameters** ([#97](https://github.com/Reiers/lantern/issues/97)). + Lantern set Lotus's global peer-score params (incl. IP-colocation + penalties) but left the per-topic map empty, so a peer that never + usefully delivered anything scored the same as one that consistently + delivered blocks first. The `/fil/blocks` + `/fil/msgs` topics now carry + Lotus's exact `TopicScoreParams` (first-delivery credit, invalid-message + penalties). This is what makes #80's "distinct **scored** peers" + meaningful against Sybil swarms: freshly-dialed attacker peers have no + delivery history, score ~0, and decay out of the mesh. + ## v1.8.5 (2026-06-30) **Head catch-up + bridge-off PDP proving robustness, and head fork-choice diff --git a/cmd/lantern/dashboard.go b/cmd/lantern/dashboard.go index f8f6d0a..f947272 100644 --- a/cmd/lantern/dashboard.go +++ b/cmd/lantern/dashboard.go @@ -359,6 +359,9 @@ func (d *dashboardDeps) syncSnapshot() map[string]any { "dedup": s.Dedup, "skipped": s.Skipped, "rejected": s.Rejected, + "rejected_lighter": s.RejectedLighter, + "held_diverged": s.HeldDiverged, + "held_uncorrob": s.HeldUncorrob, "backfilled": s.Backfilled, "backfill_failed": s.BackfillFailed, "last_install_epoch": int64(s.LastInstallEpoch), diff --git a/cmd/lantern/main.go b/cmd/lantern/main.go index 43f7b64..f09220d 100644 --- a/cmd/lantern/main.go +++ b/cmd/lantern/main.go @@ -52,6 +52,7 @@ import ( "github.com/Reiers/lantern/chain/types" "github.com/Reiers/lantern/internal/buildinfo" lbitswap "github.com/Reiers/lantern/net/bitswap" + "github.com/Reiers/lantern/net/blockpub" "github.com/Reiers/lantern/net/chainxchg" "github.com/Reiers/lantern/net/combined" "github.com/Reiers/lantern/net/glif" @@ -710,6 +711,10 @@ func cmdDaemon(args []string) error { vmBridgeToken := fs.String("vm-bridge-token", "", "Optional Bearer token for the VM bridge upstream (defaults to env LANTERN_VM_BRIDGE_TOKEN when empty).") vmBridgeTimeout := fs.Duration("vm-bridge-timeout", 30*time.Second, "Per-request timeout for VM bridge RPC calls.") allowBlockSubmit := fs.Bool("allow-block-submit", false, "Allow SyncSubmitBlock to publish to gossipsub. Requires --vm-bridge-rpc.") + // #80 part 2: head-source corroboration. -1 = auto (Full tier: 2, + // Light/PDP: off). 0 = off. N>0 = require N distinct forwarding peers + // (or one trusted floor peer) before adopting a gossip head. + headCorroPeers := fs.Int("head-corroboration-peers", -1, "Distinct gossip peers required to corroborate a head advance before adoption. -1=auto (Full tier 2, others off), 0=off.") fs.Parse(args) // #58: --allow-empty-passphrase is sugar for the env the keystore guard @@ -976,11 +981,28 @@ func cmdDaemon(args []string) error { // nodes need 50+ because they serve chain to others, participate // in F3 voting, and make deals. Lantern does none of those. The // honest floor for a light client is ~20. + // #80 part 2: resolve the corroboration requirement. Auto: the + // Full tier follows head with full trust weight on it, so it + // requires 2 distinct sources; Light/PDP stay off (their head + // safety comes from the divergence monitor + fork choice). + effCorroPeers := *headCorroPeers + if effCorroPeers < 0 { + if profile.FullValidation() { + effCorroPeers = 2 + } else { + effCorroPeers = 0 + } + } + var corroTracker *blockpub.CorroborationTracker + if effCorroPeers > 0 { + corroTracker = blockpub.NewCorroborationTracker(network.GossipTopicBlocks()) + } p2pHost, err = llibp2p.New(ctx, llibp2p.HostConfig{ ListenAddrs: listeners, BootstrapPeers: network.BootstrapPeers(), MinPeers: 20, MaxPeers: 200, + PubSubTracer: corroTracker.Tracer(), }) if err != nil { return fmt.Errorf("start libp2p host: %w", err) @@ -1054,6 +1076,18 @@ func cmdDaemon(args []string) error { fmt.Printf(" gossipsub-blocks: failed to start: %v (continuing without)\n", gerr) } else { gossipIngestor = ing + // #80 part 2: head adoption requires corroboration from + // distinct scored peers (trusted floor peers super-vote; + // requirement clamps to connected-peer count so a small + // node never wedges). + if corroTracker != nil { + hostRef := p2pHost + ing.SetHeadCorroboration(blockpub.CorroborationGate( + corroTracker, effCorroPeers, + hostRef.IsTrustedPeer, + func() int { return hostRef.PeerCount() })) + fmt.Printf(" head-corroboration: on (min distinct sources: %d, trusted floor super-vote)\n", effCorroPeers) + } // Wire the /fil/blocks publisher so SyncSubmitBlock can // actually publish (PDP/backup tier). SyncSubmitBlock still // gates on AllowBlockSubmit, so this is safe on any tier. diff --git a/net/blockingest/blockingest.go b/net/blockingest/blockingest.go index 203cc2c..61221a7 100644 --- a/net/blockingest/blockingest.go +++ b/net/blockingest/blockingest.go @@ -48,6 +48,18 @@ var log = logging.Logger("lantern/blockingest") // gossipsub event before deferring to the polling Sync agent. const DefaultBackfillCap abi.ChainEpoch = 3 +// Corroboration retry tuning (#80 part 2). Gossipsub delivers a block to +// the subscriber on FIRST receipt; the duplicate copies that serve as +// corroboration votes arrive over the following ~1s from the rest of the +// mesh. Three retries 1.5s apart give the mesh ~4.5s to corroborate +// before the block is dropped (an epoch is 30s, so a corroborated block +// still adopts with seconds to spare; an uncorroborated one is left to +// the polling Sync safety net). +const ( + corroMaxRetries = 3 + corroRetryDelay = 1500 * time.Millisecond +) + // BackfillSource is the minimal RPC surface the ingestor uses for inline // backfill when a gossipsub block arrives at head+N with N>1. The // Lantern glif client, gateway client, and combined fetcher all satisfy @@ -86,6 +98,21 @@ type Ingestor struct { // (Light/PDP, or no corroborating sources). (#79 item 2) headAdoptionGate atomic.Pointer[func() bool] + // headCorroboration, when set, is consulted with the candidate head's + // CID before adoption (#80 part 2). It returns true once the block has + // been forwarded by enough distinct scored peers (or a trusted floor + // peer). While false the ingestor holds head and retries briefly: + // corroborating duplicates typically arrive within a second of first + // delivery, so a held head is usually adopted on the first retry. + // nil = always corroborated (Light/PDP default). + headCorroboration atomic.Pointer[func(cid.Cid) bool] + + // Retry bookkeeping for corroboration holds. Touched ONLY from the + // processor goroutine (Run loop); AfterFunc callbacks merely re-send + // on the incoming channel. + corroRetries map[cid.Cid]int + corroPending map[cid.Cid]struct{} + received atomic.Uint64 dedup atomic.Uint64 installed atomic.Uint64 @@ -93,6 +120,7 @@ type Ingestor struct { rejected atomic.Uint64 rejectedLighter atomic.Uint64 // #79: candidates rejected by heaviest-weight fork choice heldDiverged atomic.Uint64 // #79: head adoptions held while the divergence gate was closed + heldUncorrob atomic.Uint64 // #80: head adoptions held awaiting multi-peer corroboration backfilled atomic.Uint64 backfillFailed atomic.Uint64 lastInstallEpoch atomic.Int64 @@ -113,17 +141,37 @@ func (g *Ingestor) SetHeadAdoptionGate(fn func() bool) { g.headAdoptionGate.Store(&fn) } +// SetHeadCorroboration wires the per-candidate corroboration predicate +// (#80 part 2): before adopting a block as the new head, the ingestor +// asks whether the block has been forwarded by enough distinct peers +// (see blockpub.CorroborationGate). While the predicate returns false +// the ingestor holds head where it is and schedules bounded retries +// (corroborating duplicates arrive within ~1s of first delivery on a +// healthy mesh). After the retries are exhausted the block is dropped; +// the polling Sync remains the safety net for head progress, so an +// uncorroboratable block can delay adoption but never wedge the node. +// Passing nil disables the check. Safe to call before or after Run. +func (g *Ingestor) SetHeadCorroboration(fn func(cid.Cid) bool) { + if fn == nil { + g.headCorroboration.Store(nil) + return + } + g.headCorroboration.Store(&fn) +} + // New builds an ingestor wired to the header store. src may be nil; when // nil, blocks at head+N>1 are skipped and the polling Sync's backfill // path handles them on its next cycle. func New(store *hstore.Store, src BackfillSource) *Ingestor { return &Ingestor{ - store: store, - src: src, - backfillCap: DefaultBackfillCap, - incoming: make(chan *ltypes.BlockMsg, 64), - seen: make(map[cid.Cid]struct{}, 256), - seenCap: 512, + store: store, + src: src, + backfillCap: DefaultBackfillCap, + incoming: make(chan *ltypes.BlockMsg, 64), + seen: make(map[cid.Cid]struct{}, 256), + seenCap: 512, + corroRetries: make(map[cid.Cid]int, 8), + corroPending: make(map[cid.Cid]struct{}, 8), } } @@ -161,8 +209,13 @@ func (g *Ingestor) process(ctx context.Context, blk *ltypes.BlockMsg) { headerCID := bh.Cid() if _, ok := g.seen[headerCID]; ok { - g.dedup.Add(1) - return + // A corroboration retry legitimately re-enters process for a + // block we've already marked seen; everything else is a dup. + if _, retry := g.corroPending[headerCID]; !retry { + g.dedup.Add(1) + return + } + delete(g.corroPending, headerCID) } g.markSeen(headerCID) @@ -244,6 +297,34 @@ func (g *Ingestor) process(ctx context.Context, blk *ltypes.BlockMsg) { } } + // #80 part 2: head-source corroboration. Gossipsub dedups messages, + // so first delivery may reach us before other mesh peers' copies have + // been counted. If the block is not yet corroborated by enough + // distinct sources, hold head and retry shortly: the duplicates that + // serve as corroboration votes typically land within a second. Give + // up after corroMaxRetries - the polling Sync is the safety net, so + // this can delay head adoption but never wedge it. + if cp := g.headCorroboration.Load(); cp != nil { + if !(*cp)(headerCID) { + g.heldUncorrob.Add(1) + if g.corroRetries[headerCID] < corroMaxRetries { + g.corroRetries[headerCID]++ + g.corroPending[headerCID] = struct{}{} + time.AfterFunc(corroRetryDelay, func() { + select { + case g.incoming <- blk: + default: + } + }) + } else { + delete(g.corroRetries, headerCID) + } + return + } + delete(g.corroRetries, headerCID) + delete(g.corroPending, headerCID) + } + if err := g.store.SetHead(ctx, ts); err != nil { g.rejected.Add(1) return @@ -368,6 +449,7 @@ type Stats struct { Rejected uint64 RejectedLighter uint64 // #79: rejected by heaviest-ParentWeight fork choice HeldDiverged uint64 // #79: head adoptions held while divergence gate closed + HeldUncorrob uint64 // #80: head adoptions held awaiting multi-peer corroboration Backfilled uint64 BackfillFailed uint64 LastInstallEpoch abi.ChainEpoch @@ -383,6 +465,7 @@ func (g *Ingestor) Stats() Stats { Rejected: g.rejected.Load(), RejectedLighter: g.rejectedLighter.Load(), HeldDiverged: g.heldDiverged.Load(), + HeldUncorrob: g.heldUncorrob.Load(), Backfilled: g.backfilled.Load(), BackfillFailed: g.backfillFailed.Load(), LastInstallEpoch: abi.ChainEpoch(g.lastInstallEpoch.Load()), diff --git a/net/blockingest/blockingest_test.go b/net/blockingest/blockingest_test.go index b777132..b58149e 100644 --- a/net/blockingest/blockingest_test.go +++ b/net/blockingest/blockingest_test.go @@ -390,3 +390,97 @@ func TestIngestor_ForkChoiceAcceptsHeavierAdvance(t *testing.T) { require.Equal(t, uint64(0), ing.Stats().RejectedLighter) require.Equal(t, head.Height()+1, s.HeadEpoch()) } + +// TestIngestor_CorroborationHoldsUncorroborated covers #80 part 2: a valid +// heavier next block whose corroboration predicate returns false must NOT +// be adopted - the ingestor holds and counts the hold. +func TestIngestor_CorroborationHoldsUncorroborated(t *testing.T) { + s, head := withStore(t, 10) + ing := New(s, nil) + ctx := context.Background() + + ing.SetHeadCorroboration(func(cid.Cid) bool { return false }) + + next := mkBlock(t, head.Height()+1, []cid.Cid{head.Blocks()[0].Cid()}, 1000, "next") + ing.process(ctx, <ypes.BlockMsg{Header: next}) + + require.Equal(t, uint64(0), ing.installed.Load(), + "uncorroborated head must be held") + require.Equal(t, uint64(1), ing.Stats().HeldUncorrob, + "hold must be counted") + require.Equal(t, head.Height(), s.HeadEpoch(), + "head must stay put while uncorroborated") +} + +// TestIngestor_CorroborationRetryAdopts: a block held for lack of +// corroboration is re-enqueued by the retry path; when corroboration +// arrives (duplicates counted) the retry adopts it. The retry re-enters +// process for a seen CID via corroPending, so this also covers the +// seen-bypass. +func TestIngestor_CorroborationRetryAdopts(t *testing.T) { + s, head := withStore(t, 10) + ing := New(s, nil) + ctx := context.Background() + + corroborated := false + ing.SetHeadCorroboration(func(cid.Cid) bool { return corroborated }) + + next := mkBlock(t, head.Height()+1, []cid.Cid{head.Blocks()[0].Cid()}, 1000, "next") + msg := <ypes.BlockMsg{Header: next} + + // First delivery: held, retry scheduled (corroPending set). + ing.process(ctx, msg) + require.Equal(t, uint64(0), ing.installed.Load(), "held on first pass") + require.Equal(t, uint64(1), ing.Stats().HeldUncorrob) + + // Corroboration arrives, then the retry re-enters process with the + // same (already seen) CID. Simulate the retry synchronously. + corroborated = true + ing.process(ctx, msg) + require.Equal(t, uint64(1), ing.installed.Load(), + "retry with corroboration must adopt") + require.Equal(t, next.Height, s.HeadEpoch(), "head must advance") +} + +// TestIngestor_CorroborationRetriesExhausted: an uncorroboratable block +// is retried corroMaxRetries times, then dropped for good; head progress +// is left to the polling Sync safety net (no wedge, no unbounded retry). +func TestIngestor_CorroborationRetriesExhausted(t *testing.T) { + s, head := withStore(t, 10) + ing := New(s, nil) + ctx := context.Background() + + ing.SetHeadCorroboration(func(cid.Cid) bool { return false }) + + next := mkBlock(t, head.Height()+1, []cid.Cid{head.Blocks()[0].Cid()}, 1000, "next") + msg := <ypes.BlockMsg{Header: next} + + // First pass + simulated retries. + for i := 0; i < corroMaxRetries+1; i++ { + ing.process(ctx, msg) + } + require.Equal(t, uint64(0), ing.installed.Load(), "never adopted") + require.Equal(t, uint64(corroMaxRetries)+1, ing.Stats().HeldUncorrob, + "each pass counts a hold") + // After exhaustion the pending flag is gone: one more process call is + // a plain dedup, not another hold. + before := ing.Stats().HeldUncorrob + ing.process(ctx, msg) + require.Equal(t, before, ing.Stats().HeldUncorrob, + "post-exhaustion re-delivery must dedup, not hold") + require.Equal(t, uint64(1), ing.dedup.Load()) +} + +// TestIngestor_CorroborationNilAdoptsNormally: nil predicate = feature +// off, adoption proceeds exactly as before (Light/PDP default). +func TestIngestor_CorroborationNilAdoptsNormally(t *testing.T) { + s, head := withStore(t, 10) + ing := New(s, nil) + ctx := context.Background() + + ing.SetHeadCorroboration(nil) + + next := mkBlock(t, head.Height()+1, []cid.Cid{head.Blocks()[0].Cid()}, 1000, "next") + ing.process(ctx, <ypes.BlockMsg{Header: next}) + require.Equal(t, uint64(1), ing.installed.Load(), "nil gate must adopt") +} diff --git a/net/blockpub/corroboration.go b/net/blockpub/corroboration.go new file mode 100644 index 0000000..d55582d --- /dev/null +++ b/net/blockpub/corroboration.go @@ -0,0 +1,206 @@ +// Corroboration tracking for the head path (#80 part 2). +// +// Gossipsub deduplicates messages before they reach subscribers, so the +// ingestor only ever sees each block once and cannot tell how many +// distinct peers forwarded it. But gossipsub's OWN plumbing sees every +// copy: the first delivery surfaces via the RawTracer's DeliverMessage +// hook and every subsequent copy via DuplicateMessage, each carrying the +// forwarding peer's ID. That makes the tracer a free corroboration +// counter: every copy of the same block CID from a distinct peer is one +// vote that this block really is circulating on the network mesh, not +// something a single (possibly hostile) peer fabricated for us. +// +// CorroborationTracker records those votes per header CID. The daemon +// wires CorroborationGate over it as the ingestor's head-corroboration +// predicate: a head advance is adopted only once >=N distinct peers have +// forwarded the block (or a trusted anchor/beacon peer has - the +// un-evictable floor from #80 part 1 counts as a super-vote). +// +// Honest boundary (TRUST-MODEL.md): peer IDs are Sybil-cheap. Distinct- +// source counting bites only in combination with gossipsub peer scoring +// (first-delivery history, IP-colocation penalties - see net/libp2p) and +// the protected trusted-peer floor. This raises the cost of a head-path +// eclipse; it is not a finality guarantee. Closing the unfinalized-tip +// fork against a powerful adversary is F3's job. + +package blockpub + +import ( + "bytes" + "sync" + + "github.com/ipfs/go-cid" + pubsub "github.com/libp2p/go-libp2p-pubsub" + "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/core/protocol" + + ltypes "github.com/Reiers/lantern/chain/types" +) + +// corroCap bounds how many distinct block CIDs the tracker retains. +// Blocks arrive ~5/epoch (30s) on Filecoin; 128 entries is ~13 minutes +// of history, far beyond the couple of seconds corroboration needs. +const corroCap = 128 + +// CorroborationTracker counts distinct forwarding peers per block header +// CID, fed by gossipsub tracer events. Safe for concurrent use: pubsub +// invokes tracer hooks from its own goroutines while the ingestor's gate +// reads counts from the processor goroutine. +type CorroborationTracker struct { + topic string + + mu sync.Mutex + sources map[cid.Cid]map[peer.ID]struct{} + order []cid.Cid // FIFO eviction + + recorded uint64 // lifetime votes recorded (incl. repeat peers) +} + +// NewCorroborationTracker builds a tracker that only counts messages on +// the given gossipsub topic (the /fil/blocks/ topic). +func NewCorroborationTracker(topic string) *CorroborationTracker { + return &CorroborationTracker{ + topic: topic, + sources: make(map[cid.Cid]map[peer.ID]struct{}, corroCap), + } +} + +// record decodes the block message and files the forwarding peer as a +// source for that header CID. Undecodable payloads are ignored: the CID +// is derived from the header bytes themselves, so a peer cannot vote for +// block H without actually sending H's header. +func (t *CorroborationTracker) record(msg *pubsub.Message) { + if msg == nil || len(msg.Data) == 0 { + return + } + if msg.GetTopic() != t.topic { + return + } + blk := new(ltypes.BlockMsg) + if err := blk.UnmarshalCBOR(bytes.NewReader(msg.Data)); err != nil || blk.Header == nil { + return + } + c := blk.Header.Cid() + if !c.Defined() { + return + } + from := msg.ReceivedFrom + if from == "" { + return + } + + t.mu.Lock() + defer t.mu.Unlock() + set, ok := t.sources[c] + if !ok { + set = make(map[peer.ID]struct{}, 4) + t.sources[c] = set + t.order = append(t.order, c) + if len(t.order) > corroCap { + evict := t.order[0] + t.order = t.order[1:] + delete(t.sources, evict) + } + } + set[from] = struct{}{} + t.recorded++ +} + +// SourceCount returns how many distinct peers have forwarded the block. +func (t *CorroborationTracker) SourceCount(c cid.Cid) int { + t.mu.Lock() + defer t.mu.Unlock() + return len(t.sources[c]) +} + +// Sources returns the distinct peers that forwarded the block. +func (t *CorroborationTracker) Sources(c cid.Cid) []peer.ID { + t.mu.Lock() + defer t.mu.Unlock() + set := t.sources[c] + out := make([]peer.ID, 0, len(set)) + for p := range set { + out = append(out, p) + } + return out +} + +// Stats returns (distinct blocks tracked, lifetime votes recorded). +func (t *CorroborationTracker) Stats() (tracked int, recorded uint64) { + t.mu.Lock() + defer t.mu.Unlock() + return len(t.sources), t.recorded +} + +// Tracer returns the pubsub.RawTracer to register at gossipsub +// construction (net/libp2p HostConfig.PubSubTracer). Nil-safe: a nil +// tracker returns a nil tracer, so callers can wire +// `PubSubTracer: maybeNilTracker.Tracer()` unconditionally. +func (t *CorroborationTracker) Tracer() pubsub.RawTracer { + if t == nil { + return nil + } + return corroTracer{t: t} +} + +// corroTracer adapts CorroborationTracker to pubsub.RawTracer. Only +// DeliverMessage (first copy, post-validation) and DuplicateMessage +// (every further copy) matter; the rest are no-ops. +type corroTracer struct { + t *CorroborationTracker +} + +var _ pubsub.RawTracer = corroTracer{} + +func (ct corroTracer) DeliverMessage(msg *pubsub.Message) { ct.t.record(msg) } +func (ct corroTracer) DuplicateMessage(msg *pubsub.Message) { ct.t.record(msg) } +func (corroTracer) AddPeer(peer.ID, protocol.ID) {} +func (corroTracer) RemovePeer(peer.ID) {} +func (corroTracer) Join(string) {} +func (corroTracer) Leave(string) {} +func (corroTracer) Graft(peer.ID, string) {} +func (corroTracer) Prune(peer.ID, string) {} +func (corroTracer) ValidateMessage(*pubsub.Message) {} +func (corroTracer) RejectMessage(*pubsub.Message, string) {} +func (corroTracer) ThrottlePeer(peer.ID) {} +func (corroTracer) RecvRPC(*pubsub.RPC) {} +func (corroTracer) SendRPC(*pubsub.RPC, peer.ID) {} +func (corroTracer) DropRPC(*pubsub.RPC, peer.ID) {} +func (corroTracer) UndeliverableMessage(*pubsub.Message) {} + +// CorroborationGate builds the head-adoption corroboration predicate the +// ingestor consults before walking head onto a new tip (#80 part 2). +// +// - minSources <= 0 disables the gate (returns nil: always adopt). +// - trusted, when non-nil, is the super-vote: if ANY forwarding peer is +// trusted (the #80 part 1 protected floor / configured beacons), the +// head is corroborated regardless of count. +// - connectedPeers, when non-nil, is the graceful-degradation input: a +// node with fewer connected peers than minSources cannot possibly +// meet the bar, so the requirement clamps to the peer count (floor 1). +// A 1-peer node must never wedge (#79 StatusInsufficient philosophy). +func CorroborationGate(t *CorroborationTracker, minSources int, trusted func(peer.ID) bool, connectedPeers func() int) func(cid.Cid) bool { + if t == nil || minSources <= 0 { + return nil + } + return func(c cid.Cid) bool { + srcs := t.Sources(c) + if trusted != nil { + for _, p := range srcs { + if trusted(p) { + return true + } + } + } + need := minSources + if connectedPeers != nil { + if n := connectedPeers(); n < need { + need = n + if need < 1 { + need = 1 + } + } + } + return len(srcs) >= need + } +} diff --git a/net/blockpub/corroboration_test.go b/net/blockpub/corroboration_test.go new file mode 100644 index 0000000..1a1b95e --- /dev/null +++ b/net/blockpub/corroboration_test.go @@ -0,0 +1,198 @@ +package blockpub + +import ( + "bytes" + "fmt" + "testing" + + "github.com/filecoin-project/go-state-types/abi" + "github.com/ipfs/go-cid" + pubsub "github.com/libp2p/go-libp2p-pubsub" + pubsubpb "github.com/libp2p/go-libp2p-pubsub/pb" + "github.com/libp2p/go-libp2p/core/peer" + + ltypes "github.com/Reiers/lantern/chain/types" +) + +const testTopic = "/fil/blocks/testnetnet" + +// corroMsg wraps valid block bytes in a pubsub.Message from a given peer +// on a given topic. +func corroMsg(t *testing.T, data []byte, from peer.ID, topic string) *pubsub.Message { + t.Helper() + return &pubsub.Message{ + Message: &pubsubpb.Message{Data: data, Topic: &topic}, + ReceivedFrom: from, + } +} + +// validBlockBytesAtHeight is validBlockBytes with a distinct height so +// tests can mint many distinct block CIDs. +func validBlockBytesAtHeight(t *testing.T, h int64) []byte { + t.Helper() + blk := new(ltypes.BlockMsg) + if err := blk.UnmarshalCBOR(bytes.NewReader(validBlockBytes(t))); err != nil { + t.Fatalf("unmarshal: %v", err) + } + blk.Header.Height = abi.ChainEpoch(h) + var buf bytes.Buffer + if err := blk.MarshalCBOR(&buf); err != nil { + t.Fatalf("marshal: %v", err) + } + return buf.Bytes() +} + +func blockCID(t *testing.T, data []byte) cid.Cid { + t.Helper() + blk := new(ltypes.BlockMsg) + if err := blk.UnmarshalCBOR(bytes.NewReader(data)); err != nil { + t.Fatalf("unmarshal test block: %v", err) + } + return blk.Header.Cid() +} + +// TestCorroborationTracker_CountsDistinctSources: the same block +// forwarded by three distinct peers counts 3; a repeat forward from an +// already-counted peer does not inflate the count. +func TestCorroborationTracker_CountsDistinctSources(t *testing.T) { + data := validBlockBytes(t) + c := blockCID(t, data) + tr := NewCorroborationTracker(testTopic) + tracer := tr.Tracer() + + tracer.DeliverMessage(corroMsg(t, data, peer.ID("peerA"), testTopic)) + tracer.DuplicateMessage(corroMsg(t, data, peer.ID("peerB"), testTopic)) + tracer.DuplicateMessage(corroMsg(t, data, peer.ID("peerC"), testTopic)) + // Repeat from peerA must not double count. + tracer.DuplicateMessage(corroMsg(t, data, peer.ID("peerA"), testTopic)) + + if got := tr.SourceCount(c); got != 3 { + t.Fatalf("SourceCount = %d, want 3", got) + } + tracked, recorded := tr.Stats() + if tracked != 1 { + t.Fatalf("tracked = %d, want 1", tracked) + } + if recorded != 4 { + t.Fatalf("recorded = %d, want 4", recorded) + } +} + +// TestCorroborationTracker_IgnoresOtherTopicsAndGarbage: messages on a +// different topic or with undecodable payloads must not register votes. +func TestCorroborationTracker_IgnoresOtherTopicsAndGarbage(t *testing.T) { + data := validBlockBytes(t) + c := blockCID(t, data) + tr := NewCorroborationTracker(testTopic) + tracer := tr.Tracer() + + // Wrong topic. + tracer.DeliverMessage(corroMsg(t, data, peer.ID("peerA"), "/fil/msgs/testnetnet")) + // Garbage payload on the right topic. + tracer.DeliverMessage(corroMsg(t, []byte{0xde, 0xad}, peer.ID("peerB"), testTopic)) + // Empty ReceivedFrom. + tracer.DeliverMessage(corroMsg(t, data, peer.ID(""), testTopic)) + + if got := tr.SourceCount(c); got != 0 { + t.Fatalf("SourceCount = %d, want 0", got) + } +} + +// TestCorroborationTracker_EvictsFIFO: the tracker holds at most corroCap +// distinct block CIDs; the oldest entry is evicted when the cap is hit. +func TestCorroborationTracker_EvictsFIFO(t *testing.T) { + tr := NewCorroborationTracker(testTopic) + // Directly exercise record's eviction via synthetic entries: build + // corroCap+1 distinct blocks by varying the header height. + first := cid.Undef + for i := 0; i <= corroCap; i++ { + data := validBlockBytesAtHeight(t, int64(1000+i)) + if i == 0 { + first = blockCID(t, data) + } + tr.Tracer().DeliverMessage(corroMsg(t, data, peer.ID(fmt.Sprintf("p%d", i)), testTopic)) + } + tracked, _ := tr.Stats() + if tracked != corroCap { + t.Fatalf("tracked = %d, want cap %d", tracked, corroCap) + } + if got := tr.SourceCount(first); got != 0 { + t.Fatalf("oldest entry should be evicted, SourceCount = %d", got) + } +} + +// TestCorroborationGate_Thresholds: below-N is held, at-N adopts. +func TestCorroborationGate_Thresholds(t *testing.T) { + data := validBlockBytes(t) + c := blockCID(t, data) + tr := NewCorroborationTracker(testTopic) + + gate := CorroborationGate(tr, 2, nil, nil) + if gate == nil { + t.Fatal("gate must not be nil for minSources=2") + } + + // Zero sources: held. + if gate(c) { + t.Fatal("uncorroborated block must be held") + } + tr.Tracer().DeliverMessage(corroMsg(t, data, peer.ID("peerA"), testTopic)) + if gate(c) { + t.Fatal("single-source block must be held at minSources=2") + } + tr.Tracer().DuplicateMessage(corroMsg(t, data, peer.ID("peerB"), testTopic)) + if !gate(c) { + t.Fatal("two-source block must be corroborated at minSources=2") + } +} + +// TestCorroborationGate_TrustedSuperVote: a single forward from a trusted +// floor peer corroborates regardless of count. +func TestCorroborationGate_TrustedSuperVote(t *testing.T) { + data := validBlockBytes(t) + c := blockCID(t, data) + tr := NewCorroborationTracker(testTopic) + + trusted := func(p peer.ID) bool { return p == peer.ID("anchor") } + gate := CorroborationGate(tr, 3, trusted, nil) + + tr.Tracer().DeliverMessage(corroMsg(t, data, peer.ID("anchor"), testTopic)) + if !gate(c) { + t.Fatal("trusted-peer forward must corroborate as a super-vote") + } +} + +// TestCorroborationGate_ClampsToPeerCount: a node with fewer connected +// peers than minSources must not wedge - the requirement clamps down. +func TestCorroborationGate_ClampsToPeerCount(t *testing.T) { + data := validBlockBytes(t) + c := blockCID(t, data) + tr := NewCorroborationTracker(testTopic) + + gate := CorroborationGate(tr, 3, nil, func() int { return 1 }) + tr.Tracer().DeliverMessage(corroMsg(t, data, peer.ID("onlyPeer"), testTopic)) + if !gate(c) { + t.Fatal("1-peer node must clamp the requirement and adopt") + } + // Even a pathological 0-peer report keeps the floor at 1. + gate0 := CorroborationGate(tr, 3, nil, func() int { return 0 }) + if !gate0(c) { + t.Fatal("0-peer clamp must floor at 1, not wedge") + } +} + +// TestCorroborationGate_DisabledReturnsNil: minSources<=0 or nil tracker +// disables the gate entirely. +func TestCorroborationGate_DisabledReturnsNil(t *testing.T) { + tr := NewCorroborationTracker(testTopic) + if CorroborationGate(tr, 0, nil, nil) != nil { + t.Fatal("minSources=0 must return a nil gate") + } + if CorroborationGate(nil, 2, nil, nil) != nil { + t.Fatal("nil tracker must return a nil gate") + } + var nilTracker *CorroborationTracker + if nilTracker.Tracer() != nil { + t.Fatal("nil tracker must return a nil tracer") + } +} diff --git a/pkg/daemon/daemon.go b/pkg/daemon/daemon.go index dc63279..6e5eaff 100644 --- a/pkg/daemon/daemon.go +++ b/pkg/daemon/daemon.go @@ -126,6 +126,15 @@ type Config struct { // before the pipeline is trusted to gate ingest. FullValidationFatal bool + // HeadCorroborationPeers (#80 part 2) is the number of distinct gossip + // peers that must forward a block before the ingestor adopts it as the + // new head. A trusted floor peer (protected bootstrap/beacon/direct + // peer) counts as a super-vote, and the requirement clamps to the + // connected-peer count so a small node never wedges. 0 disables (the + // Light/PDP default; their head safety comes from the divergence + // monitor + heaviest-weight fork choice). + HeadCorroborationPeers int + // StaleResetThreshold is the number of epochs behind live head past // which a persisted header store re-anchors near live head instead of // trying (and failing) to backfill an un-connectable gap (#51). This is diff --git a/pkg/daemon/start.go b/pkg/daemon/start.go index 1c77ba8..b8e7302 100644 --- a/pkg/daemon/start.go +++ b/pkg/daemon/start.go @@ -56,6 +56,7 @@ import ( "github.com/Reiers/lantern/chain/types" "github.com/Reiers/lantern/net/bitswap" "github.com/Reiers/lantern/net/blockingest" + "github.com/Reiers/lantern/net/blockpub" "github.com/Reiers/lantern/net/combined" "github.com/Reiers/lantern/net/glif" "github.com/Reiers/lantern/net/hsync" @@ -501,11 +502,19 @@ func (d *Daemon) stopInternal(ctx context.Context) error { // caller logs and continues on the polling Sync (best-effort). func (d *Daemon) startGossipHead(ctx context.Context, store *hstore.Store, src blockingest.BackfillSource, network build.Network, chainAPI *handlers.ChainAPI, fetcher *combined.Fetcher) error { listeners := splitCSV(d.cfg.P2PListen) + // #80 part 2: when head corroboration is enabled, the tracker rides + // gossipsub's raw tracer so every duplicate block delivery is counted + // per source peer. Nil tracker => nil tracer => zero overhead. + var corroTracker *blockpub.CorroborationTracker + if d.cfg.HeadCorroborationPeers > 0 { + corroTracker = blockpub.NewCorroborationTracker(network.GossipTopicBlocks()) + } host, err := llibp2p.New(ctx, llibp2p.HostConfig{ ListenAddrs: listeners, BootstrapPeers: network.BootstrapPeers(), MinPeers: 20, MaxPeers: 200, + PubSubTracer: corroTracker.Tracer(), }) if err != nil { return fmt.Errorf("start libp2p host: %w", err) @@ -536,6 +545,17 @@ func (d *Daemon) startGossipHead(ctx context.Context, store *hstore.Store, src b _ = host.Close() return fmt.Errorf("start gossipsub block ingestor: %w", err) } + // #80 part 2: head adoption requires corroboration from distinct + // scored peers. Trusted floor peers super-vote; the requirement + // clamps to the connected-peer count so a small embedded node (a + // curio-core box with a handful of peers) never wedges. + if corroTracker != nil { + ing.SetHeadCorroboration(blockpub.CorroborationGate( + corroTracker, d.cfg.HeadCorroborationPeers, + host.IsTrustedPeer, + func() int { return host.PeerCount() })) + log.Infow("head-source corroboration enabled", "minSources", d.cfg.HeadCorroborationPeers) + } // Capture the /fil/blocks publisher so a PDP/backup-tier daemon can // actually SUBMIT blocks (SyncSubmitBlock -> BlockPublisher). The CLI // path already wired this; the embedded daemon previously discarded it,