feat(#80 part 2, #97): head-source corroboration + per-topic gossipsub scoring

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

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),

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.

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,13 +98,29 @@ 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
 	skipped          atomic.Uint64
 	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()),

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, &ltypes.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 := &ltypes.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 := &ltypes.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, &ltypes.BlockMsg{Header: next})
+	require.Equal(t, uint64(1), ing.installed.Load(), "nil gate must adopt")
+}