turbine and leader: accelerate ingestion and block production

cmd/mithril/configcmd/configcmd.go

@@ -261,7 +261,12 @@ max_rps = 8               # Verifier's own RPC budget (never shares the block-fe
 # ── Replay tuning ────────────────────────────────────────────────────────
 [tuning]
 txpar = 24                # Validator auto-defaults to 2x CPU cores only when unset; explicit 0 = sequential
-sigverify_backend = "auto" # auto|r51|generic|stdlib; stdlib uses Go's crypto/ed25519 impl after strict checks.
+
+[sigverify]
+backend = "auto"          # auto|r51|generic|stdlib
+workers = 0               # 0 = min(2, GOMAXPROCS); explicit value overrides the shared transaction pool
+batch_target = 8          # 4 or 8 signature lanes; available work runs immediately
+disable_shred_overlap = false # Diagnostic fallback: verify after complete block assembly
 
 # ── Mithril's RPC server ─────────────────────────────────────────────────
 [rpc]

cmd/mithril/configcmd/configcmd_test.go

@@ -0,0 +1,24 @@
+package configcmd
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/spf13/viper"
+	"github.com/stretchr/testify/require"
+)
+
+func TestStarterConfigSignatureVerification(t *testing.T) {
+	for _, validator := range []bool{false, true} {
+		v := viper.New()
+		v.SetConfigType("toml")
+		require.NoError(t, v.ReadConfig(strings.NewReader(generateStarterConfig(validator))))
+		require.Equal(t, "auto", v.GetString("sigverify.backend"))
+		require.True(t, v.IsSet("sigverify.workers"))
+		require.Zero(t, v.GetInt("sigverify.workers"))
+		require.Equal(t, 8, v.GetInt("sigverify.batch_target"))
+		require.True(t, v.IsSet("sigverify.disable_shred_overlap"))
+		require.False(t, v.GetBool("sigverify.disable_shred_overlap"))
+		require.False(t, v.IsSet("tuning.sigverify_backend"))
+	}
+}

cmd/mithril/node/node.go

@@ -74,35 +74,37 @@ var (
 		},
 	}
 
-	bootstrapMode                   string // "auto", "snapshot", "new-snapshot", "new-incremental", or "accountsdb"
-	snapshotArchivePath             string
-	incrementalSnapshotFilename     string
-	accountsPath                    string
-	scratchDirectory                string
-	rpcEndpoints                    []string
-	cluster                         string // "alpenglow", "mainnet-beta", "testnet", or "devnet"
-	legacyGenesisHash               string // explicit lineage for pre-binding AccountsDB/ledger artifacts
-	blockSource                     string // "turbine", "rpc", or "lightbringer"
-	lightbringerEndpoint            string
-	repairCatchupMaxGapSlots        int    // Resume gaps up to this fill via turbine repair instead of RPC (0 = off)
-	repairMaxRequestsPerSecond      int    // Repair request-rate ceiling override (0 = adaptive default)
-	blockRPCFallback                bool   // Allow RPC block fetch when > repairCatchupMaxGapSlots behind (default false: shreds only)
-	blockMaxRPS                     int    // Rate limit for block fetching
-	blockMaxInflight                int    // Max concurrent block fetch workers
-	blockTipPollIntervalMs          int    // Tip poll interval in milliseconds
-	blockTipSafetyMargin            int    // Don't fetch within N slots of tip
-	consensusModeFlag               string // raw --consensus-mode value (cobra binding)
-	consensusMode                   string // resolved: "verifying" (default) or "validator"
-	alpenglowObserverBindAddr       string
-	alpenglowMaxMessageBytes        int64
-	alpenglowBLSDST                 string
-	validatorIdentityKeypair        string
-	validatorVoteAccountKeypair     string
-	validatorAuthorizedVoterKeypair string
-	validatorWithdrawerKeypair      string
-	validatorTPUQUICBind            string
-	validatorAdvertisedIP           string
-	validatorSigverifyWorkers       int
+	bootstrapMode                    string // "auto", "snapshot", "new-snapshot", "new-incremental", or "accountsdb"
+	snapshotArchivePath              string
+	incrementalSnapshotFilename      string
+	accountsPath                     string
+	scratchDirectory                 string
+	rpcEndpoints                     []string
+	cluster                          string // "alpenglow", "mainnet-beta", "testnet", or "devnet"
+	legacyGenesisHash                string // explicit lineage for pre-binding AccountsDB/ledger artifacts
+	blockSource                      string // "turbine", "rpc", or "lightbringer"
+	lightbringerEndpoint             string
+	repairCatchupMaxGapSlots         int    // Resume gaps up to this fill via turbine repair instead of RPC (0 = off)
+	repairMaxRequestsPerSecond       int    // Repair request-rate ceiling override (0 = adaptive default)
+	blockRPCFallback                 bool   // Allow RPC block fetch when > repairCatchupMaxGapSlots behind (default false: shreds only)
+	blockMaxRPS                      int    // Rate limit for block fetching
+	blockMaxInflight                 int    // Max concurrent block fetch workers
+	blockTipPollIntervalMs           int    // Tip poll interval in milliseconds
+	blockTipSafetyMargin             int    // Don't fetch within N slots of tip
+	consensusModeFlag                string // raw --consensus-mode value (cobra binding)
+	consensusMode                    string // resolved: "verifying" (default) or "validator"
+	alpenglowObserverBindAddr        string
+	alpenglowMaxMessageBytes         int64
+	alpenglowBLSDST                  string
+	validatorIdentityKeypair         string
+	validatorVoteAccountKeypair      string
+	validatorAuthorizedVoterKeypair  string
+	validatorWithdrawerKeypair       string
+	validatorTPUQUICBind             string
+	validatorAdvertisedIP            string
+	validatorSigverifyWorkers        int
+	validatorCompletionReserveMs     int
+	validatorMaxBufferedTransactions int
 
 	// Mode thresholds
 	blockNearTipThreshold        int // Enter near-tip when gap <= this
@@ -547,6 +549,8 @@ func init() {
 	Run.Flags().StringVar(&validatorTPUQUICBind, "tpu-quic-bind-addr", "", "Validator TPU QUIC listen address (default 0.0.0.0:8004)")
 	Run.Flags().StringVar(&validatorAdvertisedIP, "validator-advertised-ip", "", "Public IP advertised for validator TPU QUIC")
 	Run.Flags().IntVar(&validatorSigverifyWorkers, "tpu-sigverify-workers", 0, "TPU signature verification workers (0 = GOMAXPROCS)")
+	Run.Flags().IntVar(&validatorCompletionReserveMs, "leader-completion-reserve-ms", 0, "Time reserved for leader finalization and broadcast (0 = 75ms default; tune from measured completion times)")
+	Run.Flags().IntVar(&validatorMaxBufferedTransactions, "tpu-max-buffered-transactions", 0, "Maximum queued TPU transactions (0 = 131072 default)")
 
 	// [tuning] section flags
 	Run.Flags().Uint64Var(&paramArenaSizeMB, "param-arena-size-mb", 512, "Size in MB for serialized parameter arena (0 to disable)")
@@ -560,6 +564,12 @@ func init() {
 	Run.Flags().StringVar(&snapshot.SnapshotIndexTempDir, "snapshot-index-temp-dir", "", "Optional directory for snapshot index shard logs/SST staging")
 	Run.Flags().StringVar(&sigverify.Cfg.Backend, "sigverify-backend", sigverify.Defaults().Backend,
 		"ed25519 verification backend: auto|r51|generic|stdlib")
+	Run.Flags().IntVar(&sigverify.Cfg.Workers, "sigverify-workers", 0,
+		"Turbine transaction signature verification workers (0 = min(2, GOMAXPROCS))")
+	Run.Flags().IntVar(&sigverify.Cfg.BatchTarget, "sigverify-batch-target", sigverify.Defaults().BatchTarget,
+		"Turbine transaction signature batch target: 4 or 8 (available short batches run immediately)")
+	Run.Flags().BoolVar(&sigverify.Cfg.DisableShredOverlap, "sigverify-disable-shred-overlap", false,
+		"Defer Turbine transaction decoding and signature verification until all block shreds arrive")
 	Run.Flags().BoolVar(&sbpf.UsePool, "use-pool", true, "Disable to allocate fresh slices")
 	Run.Flags().IntVar(&accountsdb.StoreAccountsWorkers, "store-accounts-workers", 128, "Number of workers to write account updates")
 	Run.Flags().IntVar(&accountsdb.ProgramCacheMaxMB, "program-cache-max-mb", accountsdb.DefaultProgramCacheMaxMB, "Maximum approximate SBPF program cache size in MiB")
@@ -863,6 +873,14 @@ func initConfigAndBindFlags(cmd *cobra.Command) error {
 	}
 	validatorAdvertisedIP = getString("validator-advertised-ip", "validator.advertised_ip")
 	validatorSigverifyWorkers = getInt("tpu-sigverify-workers", "validator.tpu_sigverify_workers")
+	validatorCompletionReserveMs = getInt("leader-completion-reserve-ms", "validator.block_completion_reserve_ms")
+	validatorMaxBufferedTransactions = getInt("tpu-max-buffered-transactions", "validator.tpu_max_buffered_transactions")
+	if validatorMaxBufferedTransactions < 0 {
+		return fmt.Errorf("TPU maximum buffered transactions must be nonnegative")
+	}
+	if validatorCompletionReserveMs < 0 || validatorCompletionReserveMs >= int(blockprod.AlpenglowSlotDuration/time.Millisecond) {
+		return fmt.Errorf("leader completion reserve must be 0 (default) or between 1 and 199 milliseconds")
+	}
 
 	// [block] section
 	blockSource = getString("block-source", "block.source")
@@ -1127,10 +1145,17 @@ func initConfigAndBindFlags(cmd *cobra.Command) error {
 	// later: narya pins its backend on first use, and selecting it explicitly
 	// doubles as a startup health check, so a machine that cannot run the
 	// requested backend fails now instead of at the first block.
-	sigverify.Cfg.Backend = getString("sigverify-backend", "tuning.sigverify_backend")
+	backendKey := "sigverify.backend"
+	if !config.IsSet(backendKey) {
+		backendKey = "tuning.sigverify_backend" // older configuration files
+	}
+	sigverify.Cfg.Backend = getString("sigverify-backend", backendKey)
+	sigverify.Cfg.Workers = getInt("sigverify-workers", "sigverify.workers")
+	sigverify.Cfg.BatchTarget = getInt("sigverify-batch-target", "sigverify.batch_target")
+	sigverify.Cfg.DisableShredOverlap = getBool("sigverify-disable-shred-overlap", "sigverify.disable_shred_overlap")
 	resolved, err := sigverify.Configure(sigverify.Cfg)
 	if err != nil {
-		return fmt.Errorf("tuning.sigverify_backend: %w", err)
+		return fmt.Errorf("signature verification configuration: %w", err)
 	}
 	resolvedSigverifyBackend = resolved
 	sbpf.UsePool = getBool("use-pool", "tuning.use_pool")
@@ -2705,7 +2730,9 @@ postBootstrap:
 		defer broadcaster.Close()
 
 		controller := blockprod.NewController()
-		topicSink := scheduler.New(controller)
+		topicSink := scheduler.NewWithConfig(controller, scheduler.Config{
+			FeatureSource: replay.ChainTipFeatures, MaxBufferedTransactions: validatorMaxBufferedTransactions,
+		})
 		topicSink.Start(ctx)
 		defer topicSink.Stop()
 		tpuCfg := tpu.DefaultConfig()
@@ -2752,14 +2779,15 @@ postBootstrap:
 		leaderStop := make(chan struct{})
 		leaderDone := make(chan struct{})
 		leaderLoop := blockprod.NewLeaderLoop(blockprod.LeaderLoopConfig{
-			Controller:     controller,
-			Identity:       solana.PrivateKey(validatorIdentity),
-			AccountsDb:     accountsDb,
-			Broadcaster:    broadcaster,
-			ShredVersion:   uint16(turbineShredVersion),
-			EpochSchedule:  epochSchedule,
-			AlpenglowClock: true,
-			SlotDuration:   blockprod.AlpenglowSlotDuration,
+			Controller:        controller,
+			Identity:          solana.PrivateKey(validatorIdentity),
+			AccountsDb:        accountsDb,
+			Broadcaster:       broadcaster,
+			ShredVersion:      uint16(turbineShredVersion),
+			EpochSchedule:     epochSchedule,
+			AlpenglowClock:    true,
+			SlotDuration:      blockprod.AlpenglowSlotDuration,
+			CompletionReserve: time.Duration(validatorCompletionReserveMs) * time.Millisecond,
 			ParentContext: func(slot uint64) blockprod.ParentContext {
 				tip := replay.ChainTipParentContext()
 				// Blockprod owns the replay-readiness rule. In particular, the first
@@ -3214,6 +3242,8 @@ func printStartupInfo(commandName string) {
 		}
 		fmt.Printf("  Sigverify:    %s%s%s %s(%s)%s\n",
 			green, resolvedSigverifyBackend, reset, dim, sigverifyDesc, reset)
+		fmt.Printf("                workers=%d batch_target=%d shred_overlap=%t\n",
+			sigverify.TransactionWorkers(), sigverify.TransactionBatchTarget(), !sigverify.Cfg.DisableShredOverlap)
 	}
 
 	// Load state file for detailed info (only show for modes that use existing AccountsDB)

cmd/mithril/node/sigverify_reporter.go

@@ -42,7 +42,8 @@ func startSigverifyReporter(ctx context.Context) {
 		// against, and so the resolved backend is recorded even on a node that
 		// exits before the first tick.
 		previous := sigverify.Stats()
-		mlog.NamedFilef("sigverify", "startup: %s", previous)
+		mlog.NamedFilef("sigverify", "startup: %s workers=%d batch_target=%d shred_overlap=%t", previous,
+			sigverify.TransactionWorkers(), sigverify.TransactionBatchTarget(), !sigverify.Cfg.DisableShredOverlap)
 
 		for {
 			select {

cmd/repair-sim/main.go

@@ -0,0 +1,137 @@
+// repair-sim runs deterministic, single-node Turbine repair scenarios.
+package main
+
+import (
+	"encoding/json"
+	"flag"
+	"fmt"
+	"os"
+	"os/exec"
+	"runtime"
+	"strings"
+	"time"
+
+	"github.com/Overclock-Validator/mithril/pkg/turbine/repairsim"
+)
+
+type environment struct {
+	GoVersion string `json:"go_version"`
+	GOOS      string `json:"goos"`
+	GOARCH    string `json:"goarch"`
+	CPU       string `json:"cpu"`
+}
+
+type report struct {
+	Environment          environment            `json:"environment"`
+	Ledger               repairsim.LedgerConfig `json:"ledger"`
+	Network              repairsim.Config       `json:"network"`
+	LedgerGenerationWall time.Duration          `json:"ledger_generation_wall_ns"`
+	Result               repairsim.Result       `json:"result"`
+}
+
+func main() {
+	var (
+		scenarioFlag = flag.String("scenario", string(repairsim.ScenarioNearTip), "near-tip or deep-catchup")
+		slots        = flag.Int("slots", 200, "number of deterministic slots")
+		fecSets      = flag.Int("fec-sets", 4, "FEC sets generated per slot")
+		entries      = flag.Int("entries", 0, "entries per slot (0 derives an exact FEC count)")
+		seed         = flag.Int64("seed", 1, "deterministic content and network seed")
+		availability = flag.String("availability", "", "complete, near-loss, sparse, or mixed")
+		repair       = flag.Bool("repair", true, "enable repair requests")
+		latency      = flag.Duration("repair-latency", 20*time.Millisecond, "synthetic one-way response latency")
+		jitter       = flag.Duration("repair-jitter", 2*time.Millisecond, "deterministic +/- response jitter")
+		loss         = flag.Float64("packet-loss", 0, "repair response loss probability [0,1]")
+		duplicates   = flag.Float64("duplicates", 0.02, "duplicate response probability [0,1]")
+		bandwidth    = flag.Int64("repair-bandwidth", 100*1024*1024, "synthetic repair bytes/sec (0 is unlimited)")
+		concurrent   = flag.Int("max-concurrent", 256, "maximum outstanding repair shreds")
+		corrupt      = flag.Int("corrupt-responses", 0, "corrupt the first N repair responses")
+		naturalLate  = flag.Bool("natural-late", true, "schedule selected late live shreds during repair")
+		spoolDir     = flag.String("spool-dir", "", "persistent shred-spool directory (empty uses a temporary directory)")
+		cpuLabel     = flag.String("cpu-label", "", "explicit CPU label when platform discovery is unavailable")
+		output       = flag.String("output", "", "write JSON to this file instead of stdout")
+		includeTrace = flag.Bool("trace", true, "include the logical event trace in JSON")
+	)
+	flag.Parse()
+
+	scenario := repairsim.Scenario(*scenarioFlag)
+	network := repairsim.DefaultConfig(scenario)
+	network.Availability = repairsim.Availability(*availability)
+	network.RepairEnabled = *repair
+	network.RepairLatency = *latency
+	network.RepairJitter = *jitter
+	network.PacketLoss = *loss
+	network.DuplicateProbability = *duplicates
+	network.BandwidthBytesPerSec = *bandwidth
+	network.MaxConcurrent = *concurrent
+	network.CorruptResponses = *corrupt
+	network.NaturalLateShreds = *naturalLate
+	network.CollectTrace = *includeTrace
+	network.Seed = *seed
+	network.SpoolDir = *spoolDir
+	if network.Availability == "" {
+		network.Availability = repairsim.DefaultConfig(scenario).Availability
+	}
+
+	ledgerCfg := repairsim.LedgerConfig{
+		StartSlot:      10_000,
+		Slots:          *slots,
+		FECsPerSlot:    *fecSets,
+		EntriesPerSlot: *entries,
+		Seed:           *seed,
+		ShredVersion:   1,
+		ReferenceTick:  63,
+	}
+	started := time.Now()
+	ledger, err := repairsim.GenerateLedger(ledgerCfg)
+	if err != nil {
+		fatalf("generate ledger: %v", err)
+	}
+	generationWall := time.Since(started)
+	result, err := repairsim.Run(ledger, network)
+	if err != nil {
+		fatalf("run simulation: %v", err)
+	}
+	cpu := *cpuLabel
+	if cpu == "" {
+		cpu = cpuModel()
+	}
+	report := report{
+		Environment: environment{GoVersion: runtime.Version(), GOOS: runtime.GOOS, GOARCH: runtime.GOARCH, CPU: cpu},
+		Ledger:      ledger.Config, Network: network, LedgerGenerationWall: generationWall, Result: result,
+	}
+	encoded, err := json.MarshalIndent(report, "", "  ")
+	if err != nil {
+		fatalf("marshal report: %v", err)
+	}
+	encoded = append(encoded, '\n')
+	if *output == "" {
+		_, _ = os.Stdout.Write(encoded)
+		return
+	}
+	if err := os.WriteFile(*output, encoded, 0o644); err != nil {
+		fatalf("write %s: %v", *output, err)
+	}
+}
+
+func cpuModel() string {
+	if runtime.GOOS == "linux" {
+		if data, err := os.ReadFile("/proc/cpuinfo"); err == nil {
+			for _, line := range strings.Split(string(data), "\n") {
+				if key, value, ok := strings.Cut(line, ":"); ok && strings.TrimSpace(key) == "model name" {
+					return strings.TrimSpace(value)
+				}
+			}
+		}
+	}
+	if runtime.GOOS == "darwin" {
+		if out, err := exec.Command("sysctl", "-n", "machdep.cpu.brand_string").Output(); err == nil {
+			return strings.TrimSpace(string(out))
+		}
+	}
+	return "unknown"
+}
+
+func fatalf(format string, args ...any) {
+	_, _ = fmt.Fprintf(os.Stderr, format+"\n", args...)
+	os.Exit(1)
+}