Discovery hardening, persistence, fork behavior and counter fixes

bootnode/clconfig/admit_test.go

@@ -0,0 +1,187 @@
+package clconfig
+
+import (
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethpandaops/bootnodoor/enr"
+)
+
+func admitTestFilter(t *testing.T) *ForkDigestFilter {
+	t.Helper()
+
+	cfg := &Config{
+		SecondsPerSlot:      12,
+		customSlotsPerEpoch: 32,
+		genesisForkVersion:  [4]byte{0x00, 0x00, 0x00, 0x01},
+		forks: []forkDefinition{
+			{name: "Altair", epoch: 0, parsedVersion: [4]byte{0x01, 0x00, 0x00, 0x00}},
+		},
+	}
+	cfg.SetGenesisTime(uint64(time.Now().Unix()) - 60)
+	return NewForkDigestFilter(cfg, time.Hour)
+}
+
+func recordWithEth2(t *testing.T, eth2 []byte) *enr.Record {
+	t.Helper()
+
+	key, err := crypto.GenerateKey()
+	if err != nil {
+		t.Fatalf("generate key: %v", err)
+	}
+	rec := enr.New()
+	if err := rec.Set("ip", net.IPv4(1, 2, 3, 4)); err != nil {
+		t.Fatalf("set ip: %v", err)
+	}
+	if eth2 != nil {
+		if err := rec.Set("eth2", eth2); err != nil {
+			t.Fatalf("set eth2: %v", err)
+		}
+	}
+	rec.SetSeq(1)
+	if err := rec.Sign(key); err != nil {
+		t.Fatalf("sign: %v", err)
+	}
+	return rec
+}
+
+// Matches is the per-packet classification entry point, so it must decide
+// without moving any counter — those numbers report admissions, and packet
+// traffic dwarfs admissions by orders of magnitude.
+func TestMatchesRecordsNoStats(t *testing.T) {
+	filter := admitTestFilter(t)
+	current := filter.GetCurrentForkDigest()
+
+	cases := []struct {
+		name string
+		eth2 []byte
+		want bool
+	}{
+		{"no eth2", nil, false},
+		{"malformed", []byte{0x01, 0x02}, false},
+		{"current digest", EncodeETH2Field(current, [4]byte{0x01, 0x00, 0x00, 0x00}, ^uint64(0)), true},
+		{"unknown digest", EncodeETH2Field(ForkDigest{0xde, 0xad, 0xbe, 0xef}, [4]byte{}, 0), false},
+	}
+
+	for _, tc := range cases {
+		rec := recordWithEth2(t, tc.eth2)
+		if got := filter.Matches(rec); got != tc.want {
+			t.Errorf("Matches(%s) = %v, want %v", tc.name, got, tc.want)
+		}
+	}
+
+	stats := filter.GetStats()
+	if stats.TotalChecks != 0 || stats.AcceptedCurrent != 0 || stats.AcceptedOld != 0 ||
+		stats.AcceptedHistorical != 0 || stats.RejectedInvalid != 0 {
+		t.Fatalf("Matches moved counters: %+v", stats)
+	}
+}
+
+// Admit is the only counting entry point, and TotalChecks must always equal the
+// sum of the buckets — the web UI renders them in one table, so a reader must
+// never see rows that do not add up.
+func TestAdmitRecordsOneBucketPerCall(t *testing.T) {
+	filter := admitTestFilter(t)
+	current := filter.GetCurrentForkDigest()
+
+	cases := []struct {
+		name        string
+		eth2        []byte
+		wantAccept  bool
+		wantChecks  int
+		wantCurrent int
+		wantInvalid int
+	}{
+		{"no eth2 is uncounted", nil, false, 0, 0, 0},
+		{"current digest", EncodeETH2Field(current, [4]byte{0x01, 0x00, 0x00, 0x00}, ^uint64(0)), true, 1, 1, 0},
+		{"unknown digest", EncodeETH2Field(ForkDigest{0xde, 0xad, 0xbe, 0xef}, [4]byte{}, 0), false, 2, 1, 1},
+		{"malformed", []byte{0x01, 0x02}, false, 3, 1, 2},
+	}
+
+	for _, tc := range cases {
+		rec := recordWithEth2(t, tc.eth2)
+		if got := filter.Admit(rec); got != tc.wantAccept {
+			t.Errorf("Admit(%s) = %v, want %v", tc.name, got, tc.wantAccept)
+		}
+
+		stats := filter.GetStats()
+		if stats.TotalChecks != tc.wantChecks {
+			t.Errorf("after %s: TotalChecks = %d, want %d", tc.name, stats.TotalChecks, tc.wantChecks)
+		}
+		if stats.AcceptedCurrent != tc.wantCurrent {
+			t.Errorf("after %s: AcceptedCurrent = %d, want %d", tc.name, stats.AcceptedCurrent, tc.wantCurrent)
+		}
+		if stats.RejectedInvalid != tc.wantInvalid {
+			t.Errorf("after %s: RejectedInvalid = %d, want %d", tc.name, stats.RejectedInvalid, tc.wantInvalid)
+		}
+
+		sum := stats.AcceptedCurrent + stats.AcceptedOld + stats.AcceptedHistorical + stats.RejectedInvalid
+		if stats.TotalChecks != sum {
+			t.Errorf("after %s: TotalChecks = %d but buckets sum to %d", tc.name, stats.TotalChecks, sum)
+		}
+	}
+}
+
+// Update mutates oldForkDigests while packets are being filtered, so the digest
+// lookups must happen under the lock. Publishing the map reference and indexing
+// it afterwards is a concurrent map read/write, which aborts the process.
+func TestAdmitConcurrentWithUpdate(t *testing.T) {
+	filter := admitTestFilter(t)
+
+	rec := recordWithEth2(t, EncodeETH2Field(ForkDigest{0x11, 0x22, 0x33, 0x44}, [4]byte{}, 0))
+
+	var wg sync.WaitGroup
+	stop := make(chan struct{})
+
+	// Update only writes oldForkDigests on a fork activation or a grace expiry,
+	// so seed an already-expired entry each round to make its cleanup loop
+	// delete — the same map write, just at test frequency.
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+				filter.mu.Lock()
+				filter.oldForkDigests[ForkDigest{0x11, 0x22, 0x33, 0x44}] = time.Now().Add(-2 * time.Hour)
+				filter.mu.Unlock()
+				filter.Update()
+			}
+		}
+	}()
+
+	for i := 0; i < 8; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			for j := 0; j < 200; j++ {
+				if i%2 == 0 {
+					filter.Admit(rec)
+				} else {
+					filter.Matches(rec)
+				}
+			}
+		}(i)
+	}
+
+	// The Update loop runs until the readers finish, then is joined separately.
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+	time.Sleep(200 * time.Millisecond)
+	close(stop)
+	<-done
+
+	stats := filter.GetStats()
+	sum := stats.AcceptedCurrent + stats.AcceptedOld + stats.AcceptedHistorical + stats.RejectedInvalid
+	if stats.TotalChecks != sum {
+		t.Fatalf("TotalChecks = %d but buckets sum to %d under concurrency", stats.TotalChecks, sum)
+	}
+}

bootnode/clconfig/config.go

@@ -13,6 +13,7 @@ import (
 	"fmt"
 	"math"
 	"os"
+	"slices"
 	"sort"
 	"strings"
 	"time"
@@ -76,6 +77,29 @@ func (c *Config) getForks() []forkDefinition {
 	return c.forks
 }
 
+// ForkEpochs returns every scheduled fork epoch in ascending order, including
+// BPO entries and excluding far-future placeholders.
+//
+// Distinct from GetAllForkDigestInfos, which deduplicates by digest: the eth2
+// next-fork tuple changes at a boundary even when the current digest does not,
+// so a caller scheduling work per boundary needs the raw epochs.
+func (c *Config) ForkEpochs() []uint64 {
+	forks := c.getForks()
+
+	epochs := make([]uint64, 0, len(forks))
+	for i := range forks {
+		epoch := forks[i].epoch
+		if epoch == math.MaxUint64 {
+			continue
+		}
+		if len(epochs) > 0 && epochs[len(epochs)-1] == epoch {
+			continue
+		}
+		epochs = append(epochs, epoch)
+	}
+	return epochs
+}
+
 // GetForkEpoch returns the epoch for a given fork name.
 // Returns nil if the fork is not defined.
 func (c *Config) GetForkEpoch(forkName string) *uint64 {
@@ -549,22 +573,19 @@ func (c *Config) currentEpochNow() (uint64, bool) {
 // genesis, each registered fork (including BPO pseudo-forks), and each blob
 // schedule boundary. Sorted ascending, deduplicated.
 func (c *Config) forkBoundaryEpochs() []uint64 {
-	seen := map[uint64]bool{0: true}
 	epochs := []uint64{0}
-	add := func(epoch uint64) {
-		if epoch != math.MaxUint64 && !seen[epoch] {
-			seen[epoch] = true
-			epochs = append(epochs, epoch)
-		}
-	}
 	for _, fork := range c.getForks() {
-		add(fork.epoch)
+		if fork.epoch != math.MaxUint64 {
+			epochs = append(epochs, fork.epoch)
+		}
 	}
 	for _, entry := range c.BlobSchedule {
-		add(entry.Epoch)
+		if entry.Epoch != math.MaxUint64 {
+			epochs = append(epochs, entry.Epoch)
+		}
 	}
-	sort.Slice(epochs, func(i, j int) bool { return epochs[i] < epochs[j] })
-	return epochs
+	slices.Sort(epochs)
+	return slices.Compact(epochs)
 }
 
 // GetCurrentForkDigest returns the fork digest for the current epoch.
@@ -659,14 +680,13 @@ type ForkDigestInfo struct {
 // for same-epoch intermediate forks (never current on the wire) are
 // intentionally not included.
 func (c *Config) GetAllForkDigests() []ForkDigest {
-	var digests []ForkDigest
-	seen := make(map[ForkDigest]bool)
-	for _, epoch := range c.forkBoundaryEpochs() {
-		digest := c.GetForkDigestForEpoch(epoch)
-		if !seen[digest] {
-			seen[digest] = true
-			digests = append(digests, digest)
-		}
+	infos := c.GetAllForkDigestInfos()
+	if len(infos) == 0 {
+		return nil
+	}
+	digests := make([]ForkDigest, 0, len(infos))
+	for _, info := range infos {
+		digests = append(digests, info.Digest)
 	}
 	return digests
 }