diff --git a/cmd/mithril/main.go b/cmd/mithril/main.go index 176c810d..d6c1e3a3 100644 --- a/cmd/mithril/main.go +++ b/cmd/mithril/main.go @@ -2,7 +2,9 @@ package main import ( "context" + "errors" "flag" + "fmt" "os" "os/signal" "syscall" @@ -65,5 +67,10 @@ func init() { func main() { ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer cancel() - cobra.CheckErr(cmd.ExecuteContext(ctx)) + err := cmd.ExecuteContext(ctx) + if errors.Is(err, node.ErrRetryableBlockFetchStall) && ctx.Err() == nil { + fmt.Fprintln(os.Stderr, err) + os.Exit(75) // EX_TEMPFAIL: cleanup completed; a bounded supervisor may retry. + } + cobra.CheckErr(err) } diff --git a/cmd/mithril/node/first_fold_integrity_test.go b/cmd/mithril/node/first_fold_integrity_test.go new file mode 100644 index 00000000..d11367ea --- /dev/null +++ b/cmd/mithril/node/first_fold_integrity_test.go @@ -0,0 +1,223 @@ +package node + +import ( + "encoding/binary" + "encoding/json" + "os" + "path/filepath" + "strings" + "sync/atomic" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/accountsdb" + "github.com/Overclock-Validator/mithril/pkg/state" + "github.com/cockroachdb/pebble" + "github.com/cockroachdb/pebble/vfs" + "github.com/cockroachdb/pebble/vfs/errorfs" + "github.com/mr-tron/base58" + "github.com/stretchr/testify/require" +) + +func firstFoldIntegrityFixture(t *testing.T) (string, *accountsdb.AccountsDb, *state.MithrilState) { + t.Helper() + dir := t.TempDir() + require.NoError(t, os.Mkdir(filepath.Join(dir, "accounts"), 0o700)) + for _, name := range []string{"largest_file_id", "bootstrap_high_file_id"} { + require.NoError(t, os.WriteFile(filepath.Join(dir, name), make([]byte, 8), 0o600)) + } + db, err := accountsdb.OpenDb(dir) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, db.CloseDb()) }) + s := state.NewReadyState(7, 0, "", "", 0, 0) + hash := [32]byte{8} + require.NoError(t, s.UpdateOnShutdown(dir, 8, hash[:], nil)) + s, err = state.LoadState(dir) + require.NoError(t, err) + return dir, db, s +} + +// The fixture saves an executed tip through the real state writer; no durable +// fold has committed that tip. +func TestValidateRecoveredStateBeforeIntegrityFirstFold(t *testing.T) { + for _, fresh := range []bool{false, true} { + name := "saved_unfolded_tip" + if fresh { + name = "fresh_snapshot" + } + t.Run(name, func(t *testing.T) { + dir, db, s := firstFoldIntegrityFixture(t) + if fresh { + s = state.NewReadyState(7, 0, "", "", 0, 0) + require.NoError(t, s.Save(dir)) + } + before, err := os.ReadFile(filepath.Join(dir, "mithril_state.json")) + require.NoError(t, err) + original := *s + rec, err := db.RecoverFoldState() + require.NoError(t, err) + require.Zero(t, rec.DurableThrough) + require.Zero(t, rec.BatchSeq) + require.NoError(t, adoptRecoveredManifestContextBeforeIntegrity(dir, s, rec)) + require.NoError(t, validateRecoveredStateBeforeIntegrity(s, db, true, rec)) + require.Equal(t, original, *s, "validation must preserve executed-tip diagnostics") + after, err := os.ReadFile(filepath.Join(dir, "mithril_state.json")) + require.NoError(t, err) + require.Equal(t, before, after) + }) + } +} + +func TestValidateRecoveredStateBeforeIntegrityClassic(t *testing.T) { + for _, outcome := range []string{"matching", "missing", "mismatch", "beyond_tip"} { + t.Run(outcome, func(t *testing.T) { + _, db, s := firstFoldIntegrityFixture(t) + s.Cluster = "alpenglow" // Configured mode, not the JSON label, is authoritative. + hash := [32]byte{8} + if outcome != "missing" { + if outcome == "mismatch" { + hash[0]++ + } + require.NoError(t, db.StoreBankHashForSlot(8, hash[:])) + } + if outcome == "beyond_tip" { + require.NoError(t, db.StoreBankHashForSlot(9, hash[:])) + } + err := validateRecoveredStateBeforeIntegrity(s, db, false, accountsdb.RecoveryResult{}) + if outcome == "matching" { + require.NoError(t, err) + } else { + require.EqualError(t, err, s.ValidateAgainstBankhashDB(db).Error()) + } + }) + } +} + +func TestValidateRecoveredStateBeforeIntegrityRejectsContradictions(t *testing.T) { + cases := []struct { + name string + alter func(*state.MithrilState, *accountsdb.RecoveryResult) + }{ + {"saved_root", func(s *state.MithrilState, _ *accountsdb.RecoveryResult) { s.LastRootedSlot = 8 }}, + {"saved_root_hash", func(s *state.MithrilState, _ *accountsdb.RecoveryResult) { s.LastRootedBankhash = "unexpected" }}, + {"saved_root_context", func(s *state.MithrilState, _ *accountsdb.RecoveryResult) { + s.LastRootedContext = &state.ResumeContext{} + }}, + {"batch_without_frontier", func(_ *state.MithrilState, r *accountsdb.RecoveryResult) { r.BatchSeq = 1 }}, + {"recovered_context", func(_ *state.MithrilState, r *accountsdb.RecoveryResult) { r.ResumeCtx = []byte("{}") }}, + {"recovered_root_hash", func(_ *state.MithrilState, r *accountsdb.RecoveryResult) { r.RootedBankhash[0] = 8 }}, + {"interrupted_rewind", func(_ *state.MithrilState, r *accountsdb.RecoveryResult) { r.RewindInProgress = true }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + _, db, s := firstFoldIntegrityFixture(t) + // Fresh JSON would pass the legacy check: only the contradictory + // recovery/root metadata should make these controls fail. + s.LastSlot = 0 + s.LastBankhash = "" + rec, err := db.RecoverFoldState() + require.NoError(t, err) + tc.alter(s, &rec) + require.Error(t, validateRecoveredStateBeforeIntegrity(s, db, true, rec)) + }) + } +} + +func TestValidateRecoveredStateBeforeIntegrityRejectsNonemptyStore(t *testing.T) { + for _, name := range []string{"unrelated_slot", "malformed_key"} { + key := []byte("malformed") + if name == "unrelated_slot" { + key = make([]byte, 8) + } + if len(key) == 8 { + binary.LittleEndian.PutUint64(key, 50) // Neither the tip nor its successor. + } + t.Run(name, func(t *testing.T) { + _, db, s := firstFoldIntegrityFixture(t) + require.NoError(t, db.BankHashStore.Set(key, []byte{1}, nil)) + rec, err := db.RecoverFoldState() + require.NoError(t, err) + require.Error(t, validateRecoveredStateBeforeIntegrity(s, db, true, rec)) + }) + } +} + +func TestValidateRecoveredStateBeforeIntegrityLegacyAlpenglow(t *testing.T) { + _, db, s := firstFoldIntegrityFixture(t) + hash := [32]byte{8} + require.NoError(t, db.StoreBankHashForSlot(8, hash[:])) + rec, err := db.RecoverFoldState() + require.NoError(t, err) + require.NoError(t, validateRecoveredStateBeforeIntegrity(s, db, true, rec)) +} + +func TestValidateRecoveredStateBeforeIntegrityIteratorError(t *testing.T) { + var inject atomic.Bool + fs := errorfs.Wrap(vfs.NewMem(), errorfs.InjectorFunc(func(op errorfs.Op, path string) error { + if inject.Load() && strings.HasSuffix(path, ".sst") && op.OpKind() == errorfs.OpKindRead { + return errorfs.ErrInjected + } + return nil + })) + opts := &pebble.Options{FS: fs} + store, err := pebble.Open("bankhash", opts) + require.NoError(t, err) + require.NoError(t, store.Set([]byte("unrelated"), []byte{1}, nil)) + require.NoError(t, store.Flush()) + require.NoError(t, store.Close()) + store, err = pebble.Open("bankhash", opts) + require.NoError(t, err) + t.Cleanup(func() { + inject.Store(false) + require.NoError(t, store.Close()) + }) + inject.Store(true) + err = validateRecoveredStateBeforeIntegrity(state.NewReadyState(7, 0, "", "", 0, 0), + &accountsdb.AccountsDb{BankHashStore: store}, true, accountsdb.RecoveryResult{}) + require.ErrorIs(t, err, errorfs.ErrInjected) +} + +func TestValidateRecoveredStateBeforeIntegrityFolded(t *testing.T) { + for _, outcome := range []string{"matching", "missing", "mismatch", "lost_frontier_with_row", "invalid_context"} { + t.Run(outcome, func(t *testing.T) { + dir, db, s := firstFoldIntegrityFixture(t) + hash := [32]byte{8} + ctx := testDurableStatusContext(t, dir, 8) + ctx.Bankhash = base58.Encode(hash[:]) + encoded, err := json.Marshal(ctx) + require.NoError(t, err) + if outcome == "invalid_context" { + encoded = []byte("{}") + } + _, err = db.CommitBatch(nil, 8, map[uint64][32]byte{8: hash}, encoded) + require.NoError(t, err) + rec, err := db.RecoverFoldState() + require.NoError(t, err) + require.Equal(t, uint64(8), rec.DurableThrough) + require.Equal(t, uint64(1), rec.BatchSeq) + switch outcome { + case "missing": + var key [8]byte + binary.LittleEndian.PutUint64(key[:], 8) + require.NoError(t, db.BankHashStore.Delete(key[:], nil)) + case "mismatch": + hash[0]++ + require.NoError(t, db.StoreBankHashForSlot(8, hash[:])) + case "lost_frontier_with_row": + s.LastRootedSlot = 9 + s.LastRootedBankhash = base58.Encode(hash[:]) + require.NoError(t, db.StoreBankHashForSlot(9, hash[:])) + require.NoError(t, s.ValidateAgainstBankhashDB(db), "advisory row alone must not establish durability") + } + err = adoptRecoveredManifestContextBeforeIntegrity(dir, s, rec) + if err == nil { + err = validateRecoveredStateBeforeIntegrity(s, db, true, rec) + } + if outcome == "matching" { + require.NoError(t, err) + require.Equal(t, uint64(8), s.LastRootedSlot) + } else { + require.Error(t, err) + } + }) + } +} diff --git a/cmd/mithril/node/node.go b/cmd/mithril/node/node.go index c9c96406..466e9eab 100644 --- a/cmd/mithril/node/node.go +++ b/cmd/mithril/node/node.go @@ -69,11 +69,15 @@ var ( PreRunE: func(cmd *cobra.Command, args []string) error { return initConfigAndBindFlags(cmd) }, - Run: func(cmd *cobra.Command, args []string) { - runLive(cmd, args) + RunE: func(cmd *cobra.Command, args []string) error { + return runLive(cmd, args) }, } + // ErrRetryableBlockFetchStall marks a fetch timeout only after node cleanup + // succeeds and the run is qualified for a bounded supervisor restart. + ErrRetryableBlockFetchStall = errors.New("retryable block fetch stall") + bootstrapMode string // "auto", "snapshot", "new-snapshot", "new-incremental", or "accountsdb" snapshotArchivePath string incrementalSnapshotFilename string @@ -1283,13 +1287,17 @@ func buildSnapshotConfig(rpcEndpoints []string) snapshotdl.SnapshotConfig { return cfg } -func runLive(c *cobra.Command, args []string) { +func runLive(c *cobra.Command, args []string) (runErr error) { alpenglowMode := cluster == "alpenglow" if pprofPort != -1 { startPprofHandlers(int(pprofPort)) } ctx, cancelRun := context.WithCancel(c.Context()) defer cancelRun() + var cleanupErr error + defer func() { + runErr = liveRunExitError(runErr, cleanupErr, ctx.Err() != nil, rpcPort != 0) + }() // Print the Mithril banner first, before any other output progress.PrintBanner() @@ -1662,6 +1670,7 @@ func runLive(c *cobra.Command, args []string) { } else { defer func() { if err := lbManager.Stop(10 * time.Second); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("stop Lightbringer: %w", err)) mlog.Log.Warnf("lightbringer: shutdown error: %v", err) } }() @@ -1829,7 +1838,7 @@ func runLive(c *cobra.Command, args []string) { } // Run integrity check if we have a state file (warn only, don't fail - user chose force mode) if hasValidState { - if err := mithrilState.ValidateAgainstBankhashDB(accountsDb); err != nil { + if err := validateRecoveredStateBeforeIntegrity(mithrilState, accountsDb, alpenglowMode, foldRecovery); err != nil { mlog.Log.Errorf("WARNING: integrity check failed: %v", err) mlog.Log.Errorf("WARNING: AccountsDB may be corrupted. Consider using --bootstrap snapshot to rebuild.") } @@ -2088,7 +2097,7 @@ func runLive(c *cobra.Command, args []string) { } // Validate state file matches AccountsDB (detect Ctrl+Z / kill -9 corruption) - if err := mithrilState.ValidateAgainstBankhashDB(accountsDb); err != nil { + if err := validateRecoveredStateBeforeIntegrity(mithrilState, accountsDb, alpenglowMode, foldRecovery); err != nil { mlog.Log.Errorf("INTEGRITY CHECK FAILED: %v", err) mlog.Log.Infof("AccountsDB appears to have been modified beyond what state file records") mlog.Log.Infof("This usually happens when the process was killed with Ctrl+Z or kill -9") @@ -2561,6 +2570,14 @@ postBootstrap: } } + // Register before services with shutdown hooks so they stop before DB close. + defer func() { + mlog.Log.Infof("Done replaying, closing DB") + if err := accountsDb.CloseDb(); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("close accounts database: %w", err)) + } + }() + var rpcServer *rpcserver.RpcServer if rpcPort < 0 || rpcPort > 65535 { klog.Fatalf("invalid port: %d", rpcPort) @@ -2623,8 +2640,10 @@ postBootstrap: klog.Fatalf("unable to start consensus engine %q: %v", consensusEngine.Name(), err) } defer func() { - if err := consensusEngine.Close(); err != nil { + // Close drains the vote writer; a fault may arrive during that drain. + if err := errors.Join(consensusEngine.Close(), consensusEngine.AlpenglowSafetyError()); err != nil { mlog.Log.Warnf("consensus engine close failed: %v", err) + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("close consensus engine: %w", err)) } }() } @@ -2708,7 +2727,11 @@ postBootstrap: if err != nil { klog.Fatalf("validator turbine broadcaster: %v", err) } - defer broadcaster.Close() + defer func() { + if err := broadcaster.Close(); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("close turbine broadcaster: %w", err)) + } + }() controller := blockprod.NewController() topicSink := scheduler.NewWithConfig(controller, scheduler.Config{ @@ -2742,6 +2765,7 @@ postBootstrap: defer cancel() if err := tpuService.Stop(shutdownCtx); err != nil { mlog.Log.Warnf("validator TPU shutdown: %v", err) + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("stop validator TPU: %w", err)) } }() tpuAdvertise, err := tpuService.AdvertisedQUICAddr() @@ -2862,6 +2886,7 @@ postBootstrap: select { case localBlocks <- produced: case <-ctx.Done(): + case <-leaderStop: } }, }) @@ -2941,12 +2966,15 @@ postBootstrap: // (move + scan budgets) and serialized against folds via the store's // internal lock; the rewind horizon pins are what it must never touch. if compactEnabled { + compactCtx, stopCompaction := context.WithCancel(ctx) + compactDone := make(chan struct{}) go func() { + defer close(compactDone) ticker := time.NewTicker(10 * time.Minute) defer ticker.Stop() for { select { - case <-ctx.Done(): + case <-compactCtx.Done(): return case <-ticker.C: if _, cerr := accountsDb.CompactOnce(compactCfg); cerr != nil { @@ -2955,6 +2983,10 @@ postBootstrap: } } }() + defer func() { + stopCompaction() + <-compactDone + }() } startSigverifyReporter(ctx) @@ -3032,12 +3064,14 @@ postBootstrap: } // Record shutdown in history (must be inside this block where shutdownReason is defined) if err := mithrilState.UpdateOnShutdown(accountsPath, result.LastPersistedSlot, result.LastPersistedBankhash, shutdownCtx); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("update shutdown state: %w", err)) mlog.Log.Errorf("failed to update state file: %v", err) } state.RecordShutdown(accountsPath, result.LastPersistedSlot, base58.Encode(result.LastPersistedBankhash), replay.CurrentRunID, getVersion(), getCommit(), getBranch(), shutdownReason) } else { // No shutdown context - just update slot if err := mithrilState.UpdateOnShutdown(accountsPath, result.LastPersistedSlot, result.LastPersistedBankhash, shutdownCtx); err != nil { + cleanupErr = errors.Join(cleanupErr, fmt.Errorf("update shutdown state: %w", err)) mlog.Log.Errorf("failed to update state file: %v", err) } } @@ -3061,8 +3095,21 @@ postBootstrap: }) } - mlog.Log.Infof("Done replaying, closing DB") - accountsDb.CloseDb() + return result.Error +} + +// liveRunExitError preserves terminal failures and qualifies a fetch timeout for +// retry only after cleanup succeeds. Cancellation never hides an existing fault. +func liveRunExitError(runErr, cleanupErr error, cancelled, rpcActive bool) error { + if cleanupErr != nil { + return errors.Join(runErr, cleanupErr) + } + // ponytail: RPC has no close/join path; qualify retries only with RPC off + // until its database readers can be stopped before AccountsDB closes. + if errors.Is(runErr, replay.ErrBlockFetchStalled) && !cancelled && !rpcActive { + return fmt.Errorf("%w: %w", ErrRetryableBlockFetchStall, runErr) + } + return runErr } type turbineEntrypointQuery func(*net.UDPAddr, time.Duration) (gossip.EchoResponse, error) @@ -3999,6 +4046,38 @@ func mustRecoverFoldState(accountsDb *accountsdb.AccountsDb) accountsdb.Recovery return rec } +// validateRecoveredStateBeforeIntegrity allows an executed RAM-only LastSlot +// before the first Alpenglow fold, when recovery and the entire bankhash store +// confirm no fold. The saved tip remains diagnostic; replay starts at snapshot+1. +func validateRecoveredStateBeforeIntegrity(s *state.MithrilState, db *accountsdb.AccountsDb, alpenglowMode bool, rec accountsdb.RecoveryResult) error { + if !alpenglowMode { + return s.ValidateAgainstBankhashDB(db) + } + if rec.DurableThrough < s.LastRootedSlot { + return fmt.Errorf("state file shows last_rooted_slot=%d but the store is only durably folded through %d", s.LastRootedSlot, rec.DurableThrough) + } + if rec.DurableThrough == 0 { + if rec.BatchSeq != 0 || len(rec.ResumeCtx) != 0 || rec.RootedBankhash != ([32]byte{}) || rec.RewindInProgress || + s.LastRootedContext != nil || s.LastRootedBankhash != "" { + return fmt.Errorf("zero durable frontier has conflicting fold or rooted-state metadata") + } + iter, err := db.BankHashStore.NewIter(nil) + if err != nil { + return fmt.Errorf("open bankhash iterator before first fold: %w", err) + } + nonempty := iter.First() + if err := errors.Join(iter.Error(), iter.Close()); err != nil { + return fmt.Errorf("check bankhash store before first fold: %w", err) + } + if !nonempty { + return nil + } + // A legacy store can have per-block rows without fold metadata. + // Keep its existing validation instead of granting the empty-store exception. + } + return s.ValidateAgainstBankhashDB(db) +} + // adoptRecoveredManifestContextBeforeIntegrity runs before the legacy // state-vs-bankhash check. A committed fold manifest is the authority when its // root is equal to or ahead of the stale state file; otherwise a first-fold diff --git a/cmd/mithril/node/recovery_exit_test.go b/cmd/mithril/node/recovery_exit_test.go new file mode 100644 index 00000000..172ad734 --- /dev/null +++ b/cmd/mithril/node/recovery_exit_test.go @@ -0,0 +1,166 @@ +package node + +import ( + "errors" + "fmt" + "go/ast" + "go/parser" + "go/token" + "testing" + + "github.com/Overclock-Validator/mithril/pkg/replay" +) + +func TestLiveRunExitError(t *testing.T) { + stall := fmt.Errorf("fetch: %w", replay.ErrBlockFetchStalled) + fault := errors.New("vote history could not be sealed") + stateWrite := errors.New("shutdown state write failed") + for _, test := range []struct { + name string + run error + cleanup error + cancelled bool + rpcActive bool + want error + retry bool + }{ + {name: "clean stall", run: stall, want: stall, retry: true}, + {name: "close failure", run: stall, cleanup: fault, want: fault}, + {name: "late safety fault", run: stall, cleanup: errors.Join(nil, fault), want: fault}, + {name: "state save failure", run: stall, cleanup: stateWrite, want: stateWrite}, + {name: "state save failure after clean cancellation", cleanup: stateWrite, cancelled: true, want: stateWrite}, + {name: "cancelled stall stays a failure", run: stall, cancelled: true, want: stall}, + {name: "cancel does not hide cleanup failure", run: stall, cleanup: fault, cancelled: true, want: fault}, + {name: "RPC readers not joined stays a failure", run: stall, rpcActive: true, want: stall}, + {name: "other error", run: fault, want: fault}, + {name: "cancellation does not hide other error", run: fault, cancelled: true, want: fault}, + {name: "clean cancellation", cancelled: true}, + {name: "successful run"}, + } { + t.Run(test.name, func(t *testing.T) { + got := liveRunExitError(test.run, test.cleanup, test.cancelled, test.rpcActive) + if !errors.Is(got, test.want) { + t.Fatalf("exit error = %v, want %v", got, test.want) + } + if test.run != nil && !errors.Is(got, test.run) { + t.Errorf("exit error = %v, lost original cause %v", got, test.run) + } + if retry := errors.Is(got, ErrRetryableBlockFetchStall); retry != test.retry { + t.Errorf("retry eligible = %t, want %t", retry, test.retry) + } + }) + } +} + +// TestRunLiveForwardsTerminalErrors checks the wiring that a classifier unit test +// cannot exercise: the real runLive must forward replay and persistence errors. +// This structural check is not a substitute for the whole-runtime exit test. +func TestRunLiveForwardsTerminalErrors(t *testing.T) { + if Run.RunE == nil || Run.Run != nil { + t.Error("run command must use Cobra RunE to forward terminal errors") + } + file, err := parser.ParseFile(token.NewFileSet(), "node.go", nil, 0) + if err != nil { + t.Fatal(err) + } + var run *ast.FuncDecl + for _, decl := range file.Decls { + if fn, ok := decl.(*ast.FuncDecl); ok && fn.Name.Name == "runLive" { + run = fn + } + } + if run == nil { + t.Fatal("runLive not found") + } + last, ok := run.Body.List[len(run.Body.List)-1].(*ast.ReturnStmt) + if !ok || len(last.Results) != 1 { + t.Fatal("runLive must return its replay result after persistence") + } + field, ok := last.Results[0].(*ast.SelectorExpr) + if !ok || field.Sel.Name != "Error" { + t.Errorf("runLive terminal return drops ReplayResult.Error") + } else if name, ok := field.X.(*ast.Ident); !ok || name.Name != "result" { + t.Errorf("runLive terminal return does not forward result.Error") + } + stateWrites := 0 + ast.Inspect(run.Body, func(n ast.Node) bool { + branch, ok := n.(*ast.IfStmt) + if !ok { + return true + } + init, ok := branch.Init.(*ast.AssignStmt) + if !ok || len(init.Rhs) != 1 { + return true + } + call, ok := init.Rhs[0].(*ast.CallExpr) + if !ok { + return true + } + method, ok := call.Fun.(*ast.SelectorExpr) + if !ok || method.Sel.Name != "UpdateOnShutdown" { + return true + } + stateWrites++ + forwards := false + ast.Inspect(branch.Body, func(n ast.Node) bool { + assign, ok := n.(*ast.AssignStmt) + if ok && len(assign.Lhs) == 1 { + name, ok := assign.Lhs[0].(*ast.Ident) + if ok && name.Name == "cleanupErr" { + forwards = true + } + } + return true + }) + if !forwards { + t.Error("UpdateOnShutdown failure is not forwarded to cleanupErr") + } + return true + }) + if stateWrites != 2 { + t.Fatalf("checked %d shutdown state writes, want 2", stateWrites) + } +} + +// TestReplayShutdownResaveAllowsFinalRetry checks the second cancellation-save +// call site. A failed re-save must not tell runLive that persistence succeeded. +func TestReplayShutdownResaveAllowsFinalRetry(t *testing.T) { + file, err := parser.ParseFile(token.NewFileSet(), "../../../pkg/replay/block.go", nil, 0) + if err != nil { + t.Fatal(err) + } + var saves []*ast.IfStmt + ast.Inspect(file, func(n ast.Node) bool { + branch, ok := n.(*ast.IfStmt) + if !ok { + return true + } + init, ok := branch.Init.(*ast.AssignStmt) + if !ok || len(init.Rhs) != 1 { + return true + } + call, ok := init.Rhs[0].(*ast.CallExpr) + if !ok { + return true + } + if name, ok := call.Fun.(*ast.Ident); ok && name.Name == "onCancelWriteState" { + saves = append(saves, branch) + } + return true + }) + if len(saves) != 2 { + t.Fatalf("found %d cancellation-save calls, want 2", len(saves)) + } + for _, stmt := range saves[1].Body.List { + assign, ok := stmt.(*ast.AssignStmt) + if !ok || len(assign.Lhs) != 1 || len(assign.Rhs) != 1 { + continue + } + field, fieldOK := assign.Lhs[0].(*ast.SelectorExpr) + value, valueOK := assign.Rhs[0].(*ast.Ident) + if fieldOK && field.Sel.Name == "StateWrittenOnCancel" && valueOK && value.Name == "false" { + return + } + } + t.Fatal("failed shutdown re-save must clear StateWrittenOnCancel for runLive's final save") +} diff --git a/cmd/mithril/restart_exit_test.go b/cmd/mithril/restart_exit_test.go new file mode 100644 index 00000000..a757a6d9 --- /dev/null +++ b/cmd/mithril/restart_exit_test.go @@ -0,0 +1,85 @@ +package main + +import ( + "errors" + "fmt" + "os" + "os/exec" + "path/filepath" + "syscall" + "testing" + + "github.com/Overclock-Validator/mithril/cmd/mithril/node" + "github.com/Overclock-Validator/mithril/pkg/replay" + "github.com/spf13/cobra" +) + +// TestRestartExit runs the real main/Cobra boundary in a disposable subprocess. +// The synthetic RunE checks deferred cleanup ordering, not live node cleanup. +func TestRestartExit(t *testing.T) { + if scenario := os.Getenv("MITHRIL_TEST_RESTART_EXIT"); scenario != "" { + cmd = cobra.Command{Use: "test", SilenceUsage: true, SilenceErrors: true, + RunE: func(c *cobra.Command, args []string) error { + defer func() { + if err := os.WriteFile(os.Getenv("MITHRIL_TEST_CLEANUP"), []byte("closed"), 0600); err != nil { + panic(err) + } + }() + switch scenario { + case "clean": + return nil + case "fetch": + return node.ErrRetryableBlockFetchStall + case "wrapped": + return fmt.Errorf("after cleanup: %w: %w", node.ErrRetryableBlockFetchStall, replay.ErrBlockFetchStalled) + case "raw fetch": + return replay.ErrBlockFetchStalled + case "state failure": + return errors.Join(replay.ErrBlockFetchStalled, errors.New("shutdown state write failed")) + case "cancel", "cancel with fault", "cancel after qualification": + if err := syscall.Kill(os.Getpid(), syscall.SIGTERM); err != nil { + panic(err) + } + <-c.Context().Done() + if scenario == "cancel with fault" { + return errors.New("vote history seal failed") + } + if scenario == "cancel after qualification" { + return node.ErrRetryableBlockFetchStall + } + return nil + default: + return errors.New("block fetch stalled") + } + }} + cmd.SetArgs(nil) + main() + os.Exit(0) + } + for _, tc := range []struct { + name string + want int + }{{"clean", 0}, {"fetch", 75}, {"wrapped", 75}, {"raw fetch", 1}, {"state failure", 1}, + {"cancel", 0}, {"cancel with fault", 1}, {"cancel after qualification", 1}, {"same text, other error", 1}} { + t.Run(tc.name, func(t *testing.T) { + marker := filepath.Join(t.TempDir(), "cleanup") + child := exec.Command(os.Args[0], "-test.run=^TestRestartExit$") + child.Env = append(os.Environ(), "MITHRIL_TEST_RESTART_EXIT="+tc.name, "MITHRIL_TEST_CLEANUP="+marker) + out, err := child.CombinedOutput() + got := 0 + if err != nil { + var status *exec.ExitError + if !errors.As(err, &status) { + t.Fatal(err) + } + got = status.ExitCode() + } + if got != tc.want { + t.Errorf("exit = %d, want %d; output: %s", got, tc.want, out) + } + if data, err := os.ReadFile(marker); err != nil || string(data) != "closed" { + t.Fatalf("cleanup did not complete before exit: %q, %v", data, err) + } + }) + } +} diff --git a/pkg/accountsdb/accountsdb.go b/pkg/accountsdb/accountsdb.go index acc1666e..6d9b4153 100644 --- a/pkg/accountsdb/accountsdb.go +++ b/pkg/accountsdb/accountsdb.go @@ -203,17 +203,21 @@ func (accountsDb *AccountsDb) WaitForStoreWorker() { accountsDb.storeWorkerDone = nil } -func (accountsDb *AccountsDb) CloseDb() { +// CloseDb waits for the store worker and closes both databases, returning any close errors. +func (accountsDb *AccountsDb) CloseDb() error { accountsDb.WaitForStoreWorker() mlog.Log.Infof("CloseDb: syncing and closing Index...") - if err := accountsDb.Index.Close(); err != nil { - mlog.Log.Errorf("CloseDb: Index.Close() error: %v", err) + indexErr := accountsDb.Index.Close() + if indexErr != nil { + mlog.Log.Errorf("CloseDb: Index.Close() error: %v", indexErr) } mlog.Log.Infof("CloseDb: syncing and closing BankHashStore...") - if err := accountsDb.BankHashStore.Close(); err != nil { - mlog.Log.Errorf("CloseDb: BankHashStore.Close() error: %v", err) + bankHashErr := accountsDb.BankHashStore.Close() + if bankHashErr != nil { + mlog.Log.Errorf("CloseDb: BankHashStore.Close() error: %v", bankHashErr) } mlog.Log.Infof("CloseDb: done\n") // extra newline for spacing after close + return errors.Join(indexErr, bankHashErr) } func (accountsDb *AccountsDb) InitCaches() { diff --git a/pkg/accountsdb/close_db_test.go b/pkg/accountsdb/close_db_test.go new file mode 100644 index 00000000..ace045a3 --- /dev/null +++ b/pkg/accountsdb/close_db_test.go @@ -0,0 +1,123 @@ +package accountsdb + +import ( + "errors" + "strings" + "sync/atomic" + "testing" + + "github.com/cockroachdb/pebble" + "github.com/cockroachdb/pebble/vfs" + "github.com/cockroachdb/pebble/vfs/errorfs" +) + +// TestCloseDbReturnsStoreCloseErrors arms empty WALs only after both stores open, +// so the observed writes are Close's EOF trailers. It checks error propagation +// and Pebble's closed marker, not fsync failures or complete resource release. +func TestCloseDbReturnsStoreCloseErrors(t *testing.T) { + indexErr := errors.New("index close-time wal write failure") + bankErr := errors.New("bank close-time wal write failure") + for _, tc := range []struct { + name string + failIndex bool + failBank bool + }{ + {name: "neither"}, + {name: "index", failIndex: true}, + {name: "bank", failBank: true}, + {name: "both", failIndex: true, failBank: true}, + } { + t.Run(tc.name, func(t *testing.T) { + var armed atomic.Bool + var indexWrites, bankWrites atomic.Int32 + openStore := func(name string, fail bool, injected error, writes *atomic.Int32) (*pebble.DB, error) { + fs := errorfs.Wrap(vfs.NewMem(), errorfs.InjectorFunc(func(op errorfs.Op, path string) error { + if !armed.Load() || op != errorfs.OpFileWrite || !strings.HasSuffix(path, ".log") { + return nil + } + writes.Add(1) + if fail { + return injected + } + return nil + })) + return pebble.Open(name, &pebble.Options{FS: fs, Logger: silentLogger{}}) + } + index, err := openStore("index", tc.failIndex, indexErr, &indexWrites) + if err != nil { + t.Fatalf("open index: %v", err) + } + bank, err := openStore("bank", tc.failBank, bankErr, &bankWrites) + if err != nil { + if closeErr := index.Close(); closeErr != nil { + t.Errorf("close index after bank open failure: %v", closeErr) + } + t.Fatalf("open bank: %v", err) + } + + armed.Store(true) + got := closeDbResultForTest(&AccountsDb{Index: index, BankHashStore: bank}) + armed.Store(false) + if !tc.failIndex && !tc.failBank && got != nil { + t.Errorf("CloseDb() = %v, want nil", got) + } + for _, store := range []struct { + name string + db *pebble.DB + writes int32 + err error + wantErr bool + }{ + {name: "index", db: index, writes: indexWrites.Load(), err: indexErr, wantErr: tc.failIndex}, + {name: "bank", db: bank, writes: bankWrites.Load(), err: bankErr, wantErr: tc.failBank}, + } { + if store.writes != 1 { + t.Errorf("%s close-time WAL write callbacks = %d, want 1", store.name, store.writes) + } + if errors.Is(got, store.err) != store.wantErr { + t.Errorf("CloseDb() = %v; errors.Is(%s failure) = %t, want %t", got, store.name, errors.Is(got, store.err), store.wantErr) + } + if !closeDbStoreClosedForTest(t, store.db) { + t.Errorf("%s still accepts Get after AccountsDb.CloseDb", store.name) + // Close only a store proven open, never a second Close on a closed DB. + if err := store.db.Close(); err != nil { + t.Errorf("cleanup unclosed %s: %v", store.name, err) + } + } + } + }) + } +} + +// closeDbResultForTest keeps the void baseline runnable, so lost errors fail +// behaviorally rather than through an incompatible method signature. +func closeDbResultForTest(db *AccountsDb) error { + if closer, ok := any(db).(interface{ CloseDb() error }); ok { + return closer.CloseDb() + } + db.CloseDb() + return nil +} + +func closeDbStoreClosedForTest(t *testing.T, db *pebble.DB) (closed bool) { + t.Helper() + defer func() { + if value := recover(); value != nil { + err, ok := value.(error) + if !ok || !errors.Is(err, pebble.ErrClosed) { + panic(value) + } + closed = true + } + }() + _, closer, err := db.Get([]byte("close-state-probe")) + if closer != nil { + if closeErr := closer.Close(); closeErr != nil { + t.Errorf("close probe value: %v", closeErr) + } + } + if !errors.Is(err, pebble.ErrNotFound) { + t.Errorf("open-store probe Get() = %v, want %v", err, pebble.ErrNotFound) + } + return false +} diff --git a/pkg/alpenglow/receiver.go b/pkg/alpenglow/receiver.go index 03e1937b..98a7b012 100644 --- a/pkg/alpenglow/receiver.go +++ b/pkg/alpenglow/receiver.go @@ -95,6 +95,7 @@ type Receiver struct { connsByIP map[string]int connsByPeer map[solana.PublicKey]int stats ReceiverStats + handlers sync.WaitGroup } type receiverConnection struct { @@ -188,6 +189,7 @@ func (r *Receiver) Close() error { r.mu.Lock() if r.closed { r.mu.Unlock() + r.handlers.Wait() return nil } r.closed = true @@ -205,6 +207,7 @@ func (r *Receiver) Close() error { for _, conn := range conns { _ = conn.CloseWithError(0, "alpenglow receiver closed") } + r.handlers.Wait() if errors.Is(err, quic.ErrServerClosed) { return nil } @@ -218,6 +221,7 @@ func (r *Receiver) Stats() ReceiverStats { } func (r *Receiver) handleConn(ctx context.Context, conn *quic.Conn, peer solana.PublicKey) { + defer r.handlers.Done() defer r.removeConn(conn) defer conn.CloseWithError(0, "alpenglow receiver done") @@ -407,6 +411,9 @@ func (r *Receiver) addConn(conn *quic.Conn, peer solana.PublicKey) bool { r.conns[conn] = receiverConnection{remoteIP: remoteIP, peer: peer} r.connsByIP[remoteIP]++ r.connsByPeer[peer]++ + // Close sets closed under this lock before waiting, so no handler can be + // added after shutdown begins, even if Run has just accepted a connection. + r.handlers.Add(1) r.mu.Unlock() return true } diff --git a/pkg/alpenglow/receiver_close_test.go b/pkg/alpenglow/receiver_close_test.go new file mode 100644 index 00000000..a5cf3621 --- /dev/null +++ b/pkg/alpenglow/receiver_close_test.go @@ -0,0 +1,64 @@ +package alpenglow + +import ( + "crypto/ed25519" + "sync" + "testing" + "time" + + "github.com/gagliardetto/solana-go" + "github.com/stretchr/testify/require" +) + +func TestReceiverCloseWaitsForMessageAdmission(t *testing.T) { + entered, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + unblock := func() { releaseOnce.Do(func() { close(release) }) } + receiver, err := NewReceiver(ReceiverConfig{ + BindAddr: "127.0.0.1:0", + LogInterval: -1, + AdmitMessage: func(_ solana.PublicKey, msg Message) (Message, bool) { + close(entered) + <-release + return msg, true + }, + OnMessage: func(Message) { close(finished) }, + }, NewObserver()) + require.NoError(t, err) + runVotorReceiver(t, receiver) + t.Cleanup(unblock) // Release admission before the helper's receiver cleanup. + payload, err := EncodeMessage(NewVoteMessage(NewSkipVote(42), testSignatureSeq(0x22), 1)) + require.NoError(t, err) + identity := ed25519.NewKeyFromSeed(bytesOf(21, ed25519.SeedSize)) + conn := sendVotorQUICDatagram(t, receiver.Addr().String(), identity, payload) + t.Cleanup(func() { require.NoError(t, conn.CloseWithError(0, "test done")) }) + select { + case <-entered: + case <-time.After(2 * time.Second): + t.Fatal("message admission was not reached") + } + + closed := make(chan error, 2) + go func() { closed <- receiver.Close() }() + require.Eventually(t, receiver.isClosed, time.Second, time.Millisecond) + go func() { closed <- receiver.Close() }() // Already-closed callers must also join. + select { + case err := <-closed: + t.Fatalf("Close returned while message admission was blocked: %v", err) + case <-time.After(50 * time.Millisecond): + } + unblock() + for range 2 { + select { + case err := <-closed: + require.NoError(t, err) + case <-time.After(2 * time.Second): + t.Fatal("Close did not finish after releasing message admission") + } + } + select { + case <-finished: + default: + t.Fatal("Close returned before the admitted message callback completed") + } +} diff --git a/pkg/blockstream/block_source.go b/pkg/blockstream/block_source.go index 32a7fec1..25947489 100644 --- a/pkg/blockstream/block_source.go +++ b/pkg/blockstream/block_source.go @@ -2583,14 +2583,25 @@ func (bs *BlockSource) getMaxTipFromAllRpcs(timeout time.Duration) uint64 { func (bs *BlockSource) worker(wg *sync.WaitGroup, id int) { defer wg.Done() - for slot := range bs.workQueue { + for { + var slot uint64 + select { + case slot = <-bs.workQueue: + case <-bs.stopChan: + return + } // Check for shutdown if bs.stopped.Load() { return } - // Wait for rate limiter - bs.rateLimiter.Wait(context.Background()) + // The limiter always has a positive burst. Stop must interrupt queued + // reservations rather than wait for them and start another RPC fetch. + reservation := bs.rateLimiter.Reserve() + if bs.waitForStopOrTimeout(reservation.Delay()) || bs.stopped.Load() { + reservation.Cancel() + return + } // Lightbringer handoff may have become active after this slot was queued. // If so, skip the RPC fetch and let the stream source satisfy the slot. @@ -3677,7 +3688,8 @@ func (bs *BlockSource) Start() { // a rooted fork recovery. Keeping the close idempotent lets that external // teardown race safely with an ordinary scheduler completion. bs.Stop() - close(bs.workQueue) + // Replay may still schedule while consuming buffered blocks after Stop. + // Workers exit via stopChan; keeping this queue open avoids racing senders. wg.Wait() bs.liveStreamWg.Wait() bs.backgroundWg.Wait() @@ -3850,6 +3862,11 @@ func (bs *BlockSource) Stalled() bool { return bs.stallError.Load() } +// FetchStalled reports only the fetch-progress timeout, not unavailable history. +func (bs *BlockSource) FetchStalled() bool { + return bs.stopReasonEnum() == blockSourceStopReasonStalled +} + // StallTimeout returns the configured stall timeout duration. func (bs *BlockSource) StallTimeout() time.Duration { return bs.stallTimeout diff --git a/pkg/blockstream/fetch_stall_test.go b/pkg/blockstream/fetch_stall_test.go new file mode 100644 index 00000000..78e4928a --- /dev/null +++ b/pkg/blockstream/fetch_stall_test.go @@ -0,0 +1,81 @@ +package blockstream + +import ( + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/rpcclient" +) + +// fetchStalledForCheck uses the real baseline callers' shared predicate until +// the exact-reason query exists. This keeps the baseline control behavioral, +// rather than failing compilation because a new symbol is absent. +func fetchStalledForCheck(bs *BlockSource) bool { + if status, ok := any(bs).(interface{ FetchStalled() bool }); ok { + return status.FetchStalled() + } + return bs.Stalled() +} + +func TestFetchStalledDistinguishesHistoryAndSafetyStops(t *testing.T) { + tests := []struct { + name string + reason blockSourceStopReason + legacy bool + want bool + }{ + {name: "fetch timeout", reason: blockSourceStopReasonStalled, legacy: true, want: true}, + {name: "pruned history", reason: blockSourceStopReasonHistoryUnavailable, legacy: true}, + {name: "conflict", reason: blockSourceStopReasonAlpenglowConflict}, + {name: "invalid certificate", reason: blockSourceStopReasonInvalidAlpenglowCertificate}, + {name: "unexpected end", reason: blockSourceStopReasonUnexpectedLiveEnd}, + {name: "finite completion", reason: blockSourceStopReasonCompleted}, + {name: "operator stop"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + var bs BlockSource + bs.setStopReason(tt.reason, 100) + bs.stallError.Store(tt.legacy) + if got := fetchStalledForCheck(&bs); got != tt.want { + t.Fatalf("exact fetch stall = %t, want %t for reason %d", got, tt.want, tt.reason) + } + if got := bs.Stalled(); got != tt.legacy { + t.Fatalf("legacy stall flag changed: got %t, want %t", got, tt.legacy) + } + }) + } +} + +func TestFetchStalledPreservesFirstStopReason(t *testing.T) { + var bs BlockSource + bs.setStopReason(blockSourceStopReasonHistoryUnavailable, 100) + bs.setStopReason(blockSourceStopReasonStalled, 101) + bs.stallError.Store(true) + if fetchStalledForCheck(&bs) { + t.Fatal("first history-unavailable stop was classified as a fetch timeout") + } + if bs.stopReasonEnum() != blockSourceStopReasonHistoryUnavailable { + t.Fatal("first stop reason changed") + } +} + +func TestSchedulerFetchStallPublishesExactReason(t *testing.T) { + bs := NewBlockSource(&BlockSourceOpts{ + RpcClient: rpcclient.NewRpcClient("http://127.0.0.1:1"), + SourceType: BlockSourceRpc, + StartSlot: 100, + EndSlot: 200, + }) + // The scheduler takes its existing timeout branch before any work or RPC. + // No source Start, client call, socket or live validator is run. + bs.stallTimeout = time.Second + bs.lastProgress.Store(time.Now().Add(-time.Minute).Unix()) + bs.scheduler() + if !fetchStalledForCheck(bs) || !bs.Stalled() { + t.Fatal("scheduler timeout did not publish the fetch-stall status") + } + if bs.stopReasonEnum() != blockSourceStopReasonStalled { + t.Fatalf("scheduler reason = %d, want fetch stall", bs.stopReasonEnum()) + } +} diff --git a/pkg/blockstream/finite_queue_test.go b/pkg/blockstream/finite_queue_test.go new file mode 100644 index 00000000..ddbb5c78 --- /dev/null +++ b/pkg/blockstream/finite_queue_test.go @@ -0,0 +1,180 @@ +package blockstream + +import ( + "encoding/json" + "math" + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/rpcclient" + "github.com/gagliardetto/solana-go" +) + +// These tests run the real RPC worker, scheduler, ordered emitter and Start +// teardown. Empty synthetic blocks test queue ownership, not bank execution. +func finiteQueueRPC(t *testing.T, beforeBlock func(uint64)) *httptest.Server { + t.Helper() + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var req struct { + ID any `json:"id"` + Method string `json:"method"` + Params []json.RawMessage `json:"params"` + } + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + t.Error(err) + return + } + var value any + switch req.Method { + case "getSlot": + value = 20 + case "getBlock": + var slot uint64 + if len(req.Params) != 2 || json.Unmarshal(req.Params[0], &slot) != nil || slot < 8 { + t.Errorf("unexpected block request: %s", req.Params) + return + } + if beforeBlock != nil { + beforeBlock(slot) + } + if slot > 10 { + http.Error(w, "end of fixture", http.StatusServiceUnavailable) + return + } + value = map[string]any{"parentSlot": slot - 1, "blockhash": (solana.Hash{byte(slot)}).String(), "previousBlockhash": (solana.Hash{byte(slot - 1)}).String(), "blockHeight": slot, "blockTime": 223, "transactions": []any{}, "rewards": []any{map[string]any{"pubkey": (solana.PublicKey{1}).String(), "lamports": 1, "postBalance": 1, "rewardType": "Fee"}}} + default: + t.Errorf("unexpected RPC method %q", req.Method) + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(map[string]any{"jsonrpc": "2.0", "id": req.ID, "result": value}); err != nil { + t.Error(err) + } + })) + t.Cleanup(server.Close) + return server +} + +func startFiniteQueueSource(t *testing.T, endpoint string, end uint64) (*BlockSource, <-chan struct{}) { + t.Helper() + bs := NewBlockSource(&BlockSourceOpts{ + SourceType: BlockSourceRpc, RpcClient: rpcclient.NewRpcClient(endpoint), + StartSlot: 8, EndSlot: end, MaxRPS: 1000, MaxInflight: 1, + }) + done := make(chan struct{}) + go func() { + bs.Start() + close(done) + }() + t.Cleanup(func() { + bs.Stop() + waitFiniteQueueDone(t, done) + }) + return bs, done +} + +func waitFiniteQueueDone(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Fatal("real BlockSource.Start failed to join workers and close output") + } +} + +func TestFiniteQueueLateReplayCallback(t *testing.T) { + server := finiteQueueRPC(t, nil) + // A closed stop case and closed send case race in select. Eight independent + // finite lifetimes make the old failure reliable without changing scheduling + // state or manually closing a production-owned channel in the test. + for attempt := 0; attempt < 8; attempt++ { + bs, done := startFiniteQueueSource(t, server.URL, 11) + waitFiniteQueueDone(t, done) + if bs.stopReasonEnum() != blockSourceStopReasonCompleted { + t.Fatalf("stop reason = %s, want finite completion", bs.StopReason()) + } + for slot := uint64(8); slot <= 10; slot++ { + blk := bs.NextBlock() + if blk == nil || blk.Slot != slot || blk.LastBlockhash != [32]byte{byte(slot - 1)} || blk.Blockhash != [32]byte{byte(slot)} { + t.Fatalf("finite output at slot %d = %#v, want ordered fixture block", slot, blk) + } + func() { + defer func() { + if recovered := recover(); recovered != nil { + t.Errorf("real finite source completed before replay callback at slot %d: %v", slot, recovered) + } + }() + // Exactly the callback made by ReplayBlocks after consuming a bank. + bs.SetLastExecutedSlot(blk.Slot) + }() + } + if blk := bs.NextBlock(); blk != nil { + t.Fatalf("extra finite output: slot %d", blk.Slot) + } + if !bs.isNearTip.Load() { + t.Fatal("fixture did not enter the affected near-tip callback path") + } + } +} + +func TestFiniteQueueIdleWorkerShutdown(t *testing.T) { + var requests atomic.Int64 + server := finiteQueueRPC(t, func(uint64) { requests.Add(1) }) + bs, done := startFiniteQueueSource(t, server.URL, 8) + waitFiniteQueueDone(t, done) + if bs.NextBlock() != nil || requests.Load() != 0 { + t.Fatal("empty finite range fetched or emitted a block") + } +} + +func TestFiniteQueueBufferedWorkerShutdown(t *testing.T) { + entered := make(chan struct{}) + release := make(chan struct{}) + var enterOnce, releaseOnce sync.Once + var requests atomic.Int64 + server := finiteQueueRPC(t, func(uint64) { + requests.Add(1) + enterOnce.Do(func() { close(entered) }) + <-release + }) + bs, done := startFiniteQueueSource(t, server.URL, math.MaxUint64) + // Register after source cleanup, so failure releases the real HTTP request + // before cleanup tries to join its worker. + t.Cleanup(func() { releaseOnce.Do(func() { close(release) }) }) + select { + case <-entered: + case <-time.After(3 * time.Second): + t.Fatal("worker never reached the real HTTP request") + } + waitForBlockSourceCondition(t, func() bool { return len(bs.workQueue) > 0 }) + bs.Stop() + releaseOnce.Do(func() { close(release) }) + waitFiniteQueueDone(t, done) + if got := requests.Load(); got != 1 { + t.Fatalf("HTTP requests = %d, want only the one already in flight at Stop", got) + } +} + +func TestFiniteQueueUnboundedCancellation(t *testing.T) { + server := finiteQueueRPC(t, nil) + bs, done := startFiniteQueueSource(t, server.URL, math.MaxUint64) + for slot := uint64(8); slot <= 10; slot++ { + select { + case blk := <-bs.streamChan: + if blk == nil || blk.Slot != slot || blk.Blockhash != [32]byte{byte(slot)} { + t.Fatalf("live output at slot %d = %#v, want ordered fixture block", slot, blk) + } + bs.SetLastExecutedSlot(blk.Slot) + case <-time.After(3 * time.Second): + t.Fatalf("no live output for slot %d", slot) + } + } + bs.Stop() + waitFiniteQueueDone(t, done) + if bs.stopReasonEnum() == blockSourceStopReasonCompleted || bs.Stalled() { + t.Fatalf("clean unbounded stop misclassified: %s", bs.StopReason()) + } +} diff --git a/pkg/blockstream/rate_stop_test.go b/pkg/blockstream/rate_stop_test.go new file mode 100644 index 00000000..0db8ed65 --- /dev/null +++ b/pkg/blockstream/rate_stop_test.go @@ -0,0 +1,52 @@ +package blockstream + +import ( + "math" + "sync/atomic" + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/rpcclient" +) + +func TestRateLimitedSourceStop(t *testing.T) { + var requests atomic.Int64 + server := finiteQueueRPC(t, func(uint64) { requests.Add(1) }) + bs := NewBlockSource(&BlockSourceOpts{ + SourceType: BlockSourceRpc, RpcClient: rpcclient.NewRpcClient(server.URL), + StartSlot: 8, EndSlot: math.MaxUint64, MaxRPS: 1, MaxInflight: 2, + }) + done := make(chan struct{}) + go func() { bs.Start(); close(done) }() + t.Cleanup(func() { bs.Stop(); waitFiniteQueueDone(t, done) }) + select { + case blk := <-bs.streamChan: + if blk == nil || blk.Slot != 8 { + t.Fatalf("first block = %#v, want slot 8", blk) + } + case <-time.After(3 * time.Second): + t.Fatal("source did not fetch its first block") + } + // Two workers have reserved future tokens. Neither has an HTTP request in + // flight, and the limiter's public configuration has not been changed. + deadline := time.Now().Add(500 * time.Millisecond) + for bs.rateLimiter.Tokens() > -1.5 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if tokens := bs.rateLimiter.Tokens(); tokens > -1.5 { + t.Fatalf("queued tokens = %.2f, want two pending reservations", tokens) + } + before := requests.Load() + start := time.Now() + bs.Stop() + select { + case <-done: + case <-time.After(300 * time.Millisecond): + t.Error("Stop left Start waiting for rate-limited workers") + waitFiniteQueueDone(t, done) + } + if after := requests.Load(); after != before { + t.Errorf("getBlock requests after Stop = %d, want 0", after-before) + } + t.Logf("shutdown=%s requests_before=%d requests_after=%d", time.Since(start), before, requests.Load()) +} diff --git a/pkg/blockstream/turbine_stream.go b/pkg/blockstream/turbine_stream.go index 5e53b32a..2d343373 100644 --- a/pkg/blockstream/turbine_stream.go +++ b/pkg/blockstream/turbine_stream.go @@ -518,9 +518,9 @@ func (bs *BlockSource) runTurbineStream() { if parentDiagnostic == "" { parentDiagnostic = "none" } - mlog.Log.FileOnlyf("native turbine receiver stats: packets=%d data=%d coding=%d recovered=%d blocks=%d active_slots=%d evicted_slots=%d ignored_old_shreds=%d priority_repair_slots=%d noncanonical_block_ids=%d last_noncanonical=%s repair_requests=%d repair_responses=%d repair_timeouts=%d repair_outstanding=%d repair_peers=%d repair_pings=%d/%d repair_errors=%d repair_socket_packets=%d repair_socket_unmatched=%d retransmit_shreds=%d retransmit_packets=%d retransmit_target_packets=%d retransmit_packet_attempts=%d retransmit_unsent_packets=%d retransmit_duplicates=%d retransmit_queue_drops=%d retransmit_no_peers=%d retransmit_parent_sigs=%d/%d retransmit_sig_errors=%d retransmit_sig_direct_leader=%d retransmit_sig_repair_socket=%d retransmit_sig_unexplained=%d retransmit_parent_sig_samples=%d retransmit_actual_signer_found=%d retransmit_actual_signer_unknown=%d retransmit_send_errors=%d retransmit_peer_selection_errors=%d retransmit_send_buffer_bytes=%d retransmit_send_syscall_errors=%d retransmit_short_batches=%d retransmit_retry_batches=%d retransmit_retry_packets=%d retransmit_retry_sent=%d retransmit_exhausted_batches=%d retransmit_send_samples=%d retransmit_last_send=%q retransmit_last_parent_sig=%q shred_version_mismatches=%d parse_errors=%d sig_errors=%d missing_leaders=%d assembly_errors=%d last_packet=%s last_data_slot=%d last_block_slot=%d", + mlog.Log.FileOnlyf("native turbine receiver stats: packets=%d data=%d coding=%d recovered=%d blocks=%d active_slots=%d evicted_slots=%d ignored_old_shreds=%d priority_repair_slots=%d noncanonical_block_ids=%d last_noncanonical=%s repair_requests=%d repair_responses=%d repair_timeouts=%d repair_outstanding=%d repair_peer_inflight=%d repair_per_peer_cap=%d repair_peers=%d repair_pings=%d/%d repair_errors=%d repair_socket_packets=%d repair_socket_unmatched=%d retransmit_shreds=%d retransmit_packets=%d retransmit_target_packets=%d retransmit_packet_attempts=%d retransmit_unsent_packets=%d retransmit_duplicates=%d retransmit_queue_drops=%d retransmit_no_peers=%d retransmit_parent_sigs=%d/%d retransmit_sig_errors=%d retransmit_sig_direct_leader=%d retransmit_sig_repair_socket=%d retransmit_sig_unexplained=%d retransmit_parent_sig_samples=%d retransmit_actual_signer_found=%d retransmit_actual_signer_unknown=%d retransmit_send_errors=%d retransmit_peer_selection_errors=%d retransmit_send_buffer_bytes=%d retransmit_send_syscall_errors=%d retransmit_short_batches=%d retransmit_retry_batches=%d retransmit_retry_packets=%d retransmit_retry_sent=%d retransmit_exhausted_batches=%d retransmit_send_samples=%d retransmit_last_send=%q retransmit_last_parent_sig=%q shred_version_mismatches=%d parse_errors=%d sig_errors=%d missing_leaders=%d assembly_errors=%d last_packet=%s last_data_slot=%d last_block_slot=%d", stats.Packets, stats.DataShreds, stats.CodingShreds, stats.RecoveredData, stats.BlocksEmitted, stats.ActiveSlots, - stats.EvictedSlots, stats.IgnoredOldShreds, stats.PriorityRepairSlots, stats.NonCanonicalBlockIDs, nonCanonicalDesc, stats.Repair.Requests, stats.Repair.Responses, stats.Repair.Timeouts, stats.Repair.Outstanding, stats.Repair.Peers, + stats.EvictedSlots, stats.IgnoredOldShreds, stats.PriorityRepairSlots, stats.NonCanonicalBlockIDs, nonCanonicalDesc, stats.Repair.Requests, stats.Repair.Responses, stats.Repair.Timeouts, stats.Repair.Outstanding, stats.Repair.PeerInflight, stats.Repair.PerPeerCap, stats.Repair.Peers, stats.Repair.Pings, stats.Repair.Pongs, stats.Repair.Errors, stats.RepairSocketPackets, stats.RepairSocketUnmatched, stats.Retransmit.SentShreds, stats.Retransmit.SentPackets, stats.Retransmit.TargetPackets, stats.Retransmit.PacketAttempts, stats.Retransmit.UnsentPackets, stats.Retransmit.DuplicateShreds, stats.Retransmit.QueueDrops, stats.Retransmit.NoPeers, stats.Retransmit.ParentSignaturesVerified, stats.Retransmit.ParentSignaturesSkipped, stats.Retransmit.InvalidParentSignatures, stats.Retransmit.ParentSourceDirectLeader, stats.Retransmit.ParentSourceRepairSocket, stats.Retransmit.ParentSourceUnexplained, stats.Retransmit.ParentDiagnosticSamples, stats.Retransmit.ParentSignerFound, stats.Retransmit.ParentSignerNotFound, stats.Retransmit.SendErrors, stats.Retransmit.PeerSelectionErrors, stats.Retransmit.SendBufferBytes, diff --git a/pkg/consensus/engine.go b/pkg/consensus/engine.go index 5cf2cb20..8d27e836 100644 --- a/pkg/consensus/engine.go +++ b/pkg/consensus/engine.go @@ -1698,19 +1698,19 @@ func (e *AlpenglowObserverEngine) AlpenglowSafetyError() error { } func (e *AlpenglowObserverEngine) Close() error { + // Stop and join message admission before sealing the final voting history. + var err error + if e.receiver != nil { + err = e.receiver.Close() + } e.voterMu.Lock() voter := e.voter e.voter = nil e.voterMu.Unlock() if voter != nil { - if err := voter.close(); err != nil { - return err - } + err = errors.Join(err, voter.close()) } - if e.receiver != nil { - return e.receiver.Close() - } - return nil + return err } func (e *AlpenglowObserverEngine) ensureObserver() *alpenglow.Observer { diff --git a/pkg/consensus/engine_close_test.go b/pkg/consensus/engine_close_test.go new file mode 100644 index 00000000..c0fa5e23 --- /dev/null +++ b/pkg/consensus/engine_close_test.go @@ -0,0 +1,50 @@ +package consensus + +import ( + "errors" + "net" + "os" + "path/filepath" + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/alpenglow" + "github.com/stretchr/testify/require" +) + +func TestEngineCloseReleasesReceiverAfterVoterCloseError(t *testing.T) { + receiver, err := alpenglow.NewReceiver(alpenglow.ReceiverConfig{BindAddr: "127.0.0.1:0"}, nil) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := receiver.Close(); err != nil { + t.Errorf("close receiver: %v", err) + } + }) + addr := receiver.Addr().String() + historyLock, err := os.Create(filepath.Join(t.TempDir(), "history.lock")) + if err != nil { + t.Fatal(err) + } + if err := historyLock.Close(); err != nil { + t.Fatal(err) + } + // The real voter close returns os.ErrClosed from the already-closed lock. + engine := &AlpenglowObserverEngine{ + receiver: receiver, + voter: &alpenglowVoter{done: make(chan struct{}), historyLock: historyLock}, + } + if err := engine.Close(); !errors.Is(err, os.ErrClosed) { + t.Fatalf("engine close error = %v, want %v", err, os.ErrClosed) + } + // QUIC releases its owned UDP transport asynchronously after listener close. + var rebound net.PacketConn + require.Eventually(t, func() bool { + rebound, err = net.ListenPacket("udp4", addr) + return err == nil + }, time.Second, time.Millisecond, "receiver port still bound after failed voter close") + if err := rebound.Close(); err != nil { + t.Errorf("close rebound socket: %v", err) + } +} diff --git a/pkg/replay/alpenglow_live_feature_test.go b/pkg/replay/alpenglow_live_feature_test.go index b5ba212f..7e86e9db 100644 --- a/pkg/replay/alpenglow_live_feature_test.go +++ b/pkg/replay/alpenglow_live_feature_test.go @@ -28,7 +28,11 @@ func TestLiveAlpenglowFeatureAccountEnablesVoteRewardProcessing(t *testing.T) { db, err := accountsdb.OpenDb(dir) require.NoError(t, err) db.InitCaches() - t.Cleanup(db.CloseDb) + t.Cleanup(func() { + if err := db.CloseDb(); err != nil { + t.Errorf("close database: %v", err) + } + }) featureActivationSlot := activationSlot featureData, err := features.MarshalFeatureAcct(&features.FeatureAcct{ActivatedAt: &featureActivationSlot}) diff --git a/pkg/replay/alpenglow_startup_test.go b/pkg/replay/alpenglow_startup_test.go index 464e0f00..e344233b 100644 --- a/pkg/replay/alpenglow_startup_test.go +++ b/pkg/replay/alpenglow_startup_test.go @@ -88,7 +88,11 @@ func TestValidateAlpenglowStartupStateWithAccountsDB(t *testing.T) { db, err := accountsdb.OpenDb(dir) require.NoError(t, err) db.InitCaches() - t.Cleanup(db.CloseDb) + t.Cleanup(func() { + if err := db.CloseDb(); err != nil { + t.Errorf("close database: %v", err) + } + }) stored := make(chan struct{}) startupAccounts := validAlpenglowStartupAccounts(t, 486_000) diff --git a/pkg/replay/block.go b/pkg/replay/block.go index 5dc2e431..abbc6f13 100644 --- a/pkg/replay/block.go +++ b/pkg/replay/block.go @@ -144,6 +144,9 @@ func formatBlockSourceStatus(fetchStats blockstream.FetchStatsSnapshot) string { return fetchStats.SourceStatus } +// ErrBlockFetchStalled identifies a fetch timeout after replay cleanup. +var ErrBlockFetchStalled = errors.New("block fetch stalled") + // ReplayResult contains the result of a replay operation, including shutdown state type ReplayResult struct { // LastPersistedSlot is the last slot whose state was successfully persisted to AccountsDB @@ -1677,6 +1680,20 @@ func ReplayBlocks( onCancelWriteState OnCancelWriteState, // callback to write state immediately on cancellation (can be nil) ) *ReplayResult { result := &ReplayResult{} + var blockStream *blockstream.BlockSource + var foldFailed bool + // Register first so safety checks and every existing cleanup defer run + // before the fetch timeout can become a retry-eligible error. + defer func() { + if result.Error == nil && !result.WasCancelled && ctx.Err() == nil && + blockStream != nil && blockStream.FetchStalled() { + if foldFailed { + result.Error = fmt.Errorf("block fetch stalled - no progress for %v", blockStream.StallTimeout()) + } else { + result.Error = fmt.Errorf("%w - no progress for %v", ErrBlockFetchStalled, blockStream.StallTimeout()) + } + } + }() alpenglowMode := consensusOpts != nil && consensusOpts.Alpenglow replayFrontier := uint64(0) if startSlot > 0 { @@ -2041,12 +2058,29 @@ func ReplayBlocks( if TrailingVerifierCfg.Enabled && unrootedTailState != nil { trailingVerifier = newTrailingVerifier(&rpcVerificationSource{rpcc: rpcclient.NewRpcClient(rpcEndpoints[0])}, TrailingVerifierCfg) - go trailingVerifier.Run(ctx) + verifyCtx, stopVerifier := context.WithCancel(ctx) + verifierDone := make(chan struct{}) + go func() { + defer close(verifierDone) + trailingVerifier.Run(verifyCtx) + }() // Publish a run-local tx-capture registry for the verifier's lifetime and // unpublish it on return, so no other run (a later re-replay, a test, a // sim) shares or inherits this run's capture state. stopCapture := beginTxCapture() - defer stopCapture() + defer func() { + stopVerifier() + <-verifierDone + stopCapture() + // An in-flight RPC check can discover divergence during shutdown. + if div := trailingVerifier.Failure(); div != nil { + recordReplayDivergenceEvidence(mithrilState, div) + // A cancellation save may predate this evidence. Persist it again, + // and let divergence override any earlier recoverable fork switch. + result.StateWrittenOnCancel = false + result.Error = fmt.Errorf("replay verification failed during shutdown: %w", div) + } + }() if !TrailingVerifierCfg.Required { mlog.Log.Warnf("trailing verifier running in ADVISORY mode (verifier.required=false): folds are NOT gated on execution verification") } @@ -2110,11 +2144,13 @@ func ReplayBlocks( return } if res.err != nil { + foldFailed = true mlog.Log.Errorf("rooted-durable: async fold failed: %v", res.err) return } rootedCtx := unrootedTailState.applyFoldJob(res.job) if rootedCtx == nil { + foldFailed = true mlog.Log.Errorf("rooted-durable: fold through slot %d returned no resume context; watermark held back", res.job.through) return } @@ -2227,10 +2263,12 @@ func ReplayBlocks( if rewardsHoldBelowSlot > 0 && partitionedRewardsInfo != nil && promoteThrough >= rewardsHoldBelowSlot { job, jerr := unrootedTailState.buildRewardsCompletionFoldJob(rewardsCompletion.slot) if jerr != nil { + foldFailed = true mlog.Log.Errorf("rooted-durable: rewards completion fold: %v", jerr) return false } if err := runFoldJob(unrootedTailState.committer, job); err != nil { + foldFailed = true mlog.Log.Errorf("rooted-durable: rewards completion fold: %v", err) return false } @@ -2238,6 +2276,7 @@ func ReplayBlocks( } promotedThrough, rootedCtx, perr := unrootedTailState.flush(promoteThrough) if perr != nil { + foldFailed = true mlog.Log.Errorf("rooted-durable: forced fold stopped at slot %d: %v", promotedThrough, perr) } if promotedThrough > mithrilState.LastRootedSlot && rootedCtx != nil { @@ -2257,6 +2296,7 @@ func ReplayBlocks( job, jerr = unrootedTailState.buildFoldJob(promoteThrough, false) } if jerr != nil { + foldFailed = true mlog.Log.Errorf("rooted-durable: %v; watermark held back", jerr) return false } @@ -2395,7 +2435,7 @@ func ReplayBlocks( } } - blockStream := blockstream.NewBlockSource(opts) + blockStream = blockstream.NewBlockSource(opts) if !isLive { blockStream.DownloadInitialBlocks() @@ -2724,6 +2764,8 @@ func ReplayBlocks( if block == nil { if result.Error == nil { switch { + case blockStream.FetchStalled(): + // Leave shutdown-only safety findings room to set Error. case blockStream.Stalled(): result.Error = fmt.Errorf("block fetch stalled - no progress for %v", blockStream.StallTimeout()) case isLive && !blockStream.Completed(): @@ -3623,8 +3665,8 @@ func ReplayBlocks( } - // Check if block source stalled (this provides explicit error info) - if blockStream.Stalled() && result.Error == nil { + // Keep unavailable-history handling unchanged; exact timeouts classify last. + if blockStream.Stalled() && !blockStream.FetchStalled() && result.Error == nil { result.Error = fmt.Errorf("block fetch stalled - no progress for %v", blockStream.StallTimeout()) } @@ -3644,6 +3686,7 @@ func ReplayBlocks( stateChanged := mithrilState.LastRootedSlot > preFlushRooted || len(mithrilState.AlpenglowEvidence) > preFlushEvidence if result.StateWrittenOnCancel && onCancelWriteState != nil && stateChanged { if err := onCancelWriteState(result); err != nil { + result.StateWrittenOnCancel = false mlog.Log.Errorf("failed to re-write state after shutdown flush (recovery reconcile will cover it): %v", err) } } diff --git a/pkg/replay/epoch_boundary_capture_test.go b/pkg/replay/epoch_boundary_capture_test.go index 2be67670..d107766a 100644 --- a/pkg/replay/epoch_boundary_capture_test.go +++ b/pkg/replay/epoch_boundary_capture_test.go @@ -105,7 +105,11 @@ func capturedEpochBoundaryDB(t *testing.T, slot uint64, delegations []capturedBo db, err := accountsdb.OpenDb(dir) require.NoError(t, err) db.InitCaches() - t.Cleanup(db.CloseDb) + t.Cleanup(func() { + if err := db.CloseDb(); err != nil { + t.Errorf("close database: %v", err) + } + }) global.ClearPendingStakePubkeys() accts := make([]*accounts.Account, 0, len(delegations)) for _, captured := range delegations { diff --git a/pkg/replay/epoch_boundary_stakes_test.go b/pkg/replay/epoch_boundary_stakes_test.go index fc448622..91916e24 100644 --- a/pkg/replay/epoch_boundary_stakes_test.go +++ b/pkg/replay/epoch_boundary_stakes_test.go @@ -110,7 +110,11 @@ func epochBoundaryStakeDB(t *testing.T, slot uint64, delegations []sealevel.Dele db, err := accountsdb.OpenDb(dir) require.NoError(t, err) db.InitCaches() - t.Cleanup(db.CloseDb) + t.Cleanup(func() { + if err := db.CloseDb(); err != nil { + t.Errorf("close database: %v", err) + } + }) global.ClearPendingStakePubkeys() var accts []*accounts.Account diff --git a/pkg/replay/epoch_inflation_test.go b/pkg/replay/epoch_inflation_test.go index fba7408e..6d067b24 100644 --- a/pkg/replay/epoch_inflation_test.go +++ b/pkg/replay/epoch_inflation_test.go @@ -166,7 +166,11 @@ func TestStageEpochInflationAccountRollsStateAndCapitalization(t *testing.T) { db, err := accountsdb.OpenDb(dir) require.NoError(t, err) db.InitCaches() - t.Cleanup(db.CloseDb) + t.Cleanup(func() { + if err := db.CloseDb(); err != nil { + t.Errorf("close database: %v", err) + } + }) previousRent := sealevel.SysvarCache.Rent.Sysvar rent := sealevel.NewDefaultRentSysvar() diff --git a/pkg/replay/rewards_empty_partition_test.go b/pkg/replay/rewards_empty_partition_test.go index 89a7c74f..6cb978b9 100644 --- a/pkg/replay/rewards_empty_partition_test.go +++ b/pkg/replay/rewards_empty_partition_test.go @@ -37,7 +37,11 @@ func TestSameBankRewardPartitionWaitsForStartingBlockHeight(t *testing.T) { db, err := accountsdb.OpenDb(dir) require.NoError(t, err) db.InitCaches() - t.Cleanup(db.CloseDb) + t.Cleanup(func() { + if err := db.CloseDb(); err != nil { + t.Errorf("close database: %v", err) + } + }) previousCacheAcct := sealevel.SysvarCache.EpochRewards.Acct previousCacheSysvar := sealevel.SysvarCache.EpochRewards.Sysvar diff --git a/pkg/rewards/distribution_test.go b/pkg/rewards/distribution_test.go index 68d825ab..16feb352 100644 --- a/pkg/rewards/distribution_test.go +++ b/pkg/rewards/distribution_test.go @@ -21,7 +21,11 @@ func TestDistributeVotingRewardsBuildsOnStagedAccount(t *testing.T) { db, err := accountsdb.OpenDb(dir) require.NoError(t, err) db.InitCaches() - t.Cleanup(db.CloseDb) + t.Cleanup(func() { + if err := db.CloseDb(); err != nil { + t.Errorf("close database: %v", err) + } + }) votePubkey := solana.NewWallet().PublicKey() parent := &accounts.Account{ diff --git a/pkg/rewards/inactive_stakes_test.go b/pkg/rewards/inactive_stakes_test.go index 278e8c3d..60f0a110 100644 --- a/pkg/rewards/inactive_stakes_test.go +++ b/pkg/rewards/inactive_stakes_test.go @@ -48,7 +48,11 @@ func TestEpoch116InactiveStakeBankHash(t *testing.T) { db, err := accountsdb.OpenDb(dir) require.NoError(t, err) db.InitCaches() - t.Cleanup(db.CloseDb) + t.Cleanup(func() { + if err := db.CloseDb(); err != nil { + t.Errorf("close database: %v", err) + } + }) global.ClearPendingStakePubkeys() t.Cleanup(global.ClearPendingStakePubkeys) diff --git a/pkg/turbine/entry_pipeline_trace.go b/pkg/turbine/entry_pipeline_trace.go index 38ea1055..8770b9dc 100644 --- a/pkg/turbine/entry_pipeline_trace.go +++ b/pkg/turbine/entry_pipeline_trace.go @@ -329,5 +329,5 @@ func traceRepairResponse(rec outstandingRepairRequest, sh *Shred, peer string, l if !entryTraceSelected(sh.Slot) { return } - emitRepairTrace(entryRepairTrace{Event: "repair_response", Origin: entryTraceOrigin.UnixNano(), Slot: sh.Slot, Index: rec.key.index, Highest: rec.key.kind == repairRequestHighestWindowIndex, Attempt: rec.key.attempt, Nonce: rec.nonce, Peer: peer, RequestedAt: entryTraceTime(rec.sentAt), ResponseAt: entryTraceNow(), ReturnedIndex: sh.Index, FEC: sh.FECSetIndex, Late: late}) + emitRepairTrace(entryRepairTrace{Event: "repair_response", Origin: entryTraceOrigin.UnixNano(), Slot: sh.Slot, Index: rec.key.index, Highest: rec.key.kind == repairRequestHighestWindowIndex, Attempt: uint8(rec.key.attempt), Nonce: rec.nonce, Peer: peer, RequestedAt: entryTraceTime(rec.sentAt), ResponseAt: entryTraceNow(), ReturnedIndex: sh.Index, FEC: sh.FECSetIndex, Late: late}) } diff --git a/pkg/turbine/repair.go b/pkg/turbine/repair.go index 9a39ab8f..1af2fa91 100644 --- a/pkg/turbine/repair.go +++ b/pkg/turbine/repair.go @@ -198,6 +198,11 @@ type RepairStats struct { Errors uint64 Outstanding int Peers int + // PeerInflight includes sends still signing, before request registration. + // A transient excess over Outstanding is normal; a persistent excess can + // explain admission stalls even when no registered requests remain. + PeerInflight int + PerPeerCap int // Peers that answered a request within repairResponderWindow — the // number that separates "cluster serves this range" from "only N nodes // still retain it". @@ -227,7 +232,7 @@ type repairRequestKey struct { // different peers — head-slot fanout AND fast-retry (a stale attempt // spawns the next one). Monotonic per shred; keyed so each attempt has // its own outstanding entry and nonce. - attempt uint8 + attempt uint64 } // shredKey identifies a missing shred independent of which attempt is asking @@ -248,7 +253,7 @@ func (k repairRequestKey) shred() shredKey { // decision is O(1) instead of an O(outstanding) scan per missing shred. type shredInflight struct { concurrent uint8 - nextAttempt uint8 + nextAttempt uint64 lastSentAt time.Time } @@ -408,7 +413,7 @@ func (c *repairClient) run(ctx context.Context, conn *net.UDPConn, assembler *Sl // Keep periodic scans for retries/freshness. Coalesce urgent hints and bound // scan frequency; all sends still use the existing token bucket, admission, -// retry, fanout and peer budgets. The loop remains the sole sender. +// retry, fanout and peer budgets, shared with response-driven followup sends. func runRepairSchedule(ctx context.Context, wake <-chan struct{}, interval, minSpacing time.Duration, scan func()) { ticker := time.NewTicker(interval) defer ticker.Stop() @@ -811,19 +816,20 @@ func (c *repairClient) cancelSlotRequests(slot uint64) { } // cancelShredRequestsLocked retires the handful of concurrent attempts for a -// single shred in O(attempts), not O(all outstanding). It must be called with -// c.mu held. +// single shred. It must be called with c.mu held. +// ponytail: scans all issued attempts in this generation; track active IDs +// separately only if long-lived generations make cancellation costly. func (c *repairClient) cancelShredRequestsLocked(sk shredKey) { inf := c.inflight[sk] if inf == nil { return } - for attempt := 0; attempt < int(inf.nextAttempt); attempt++ { + for attempt := uint64(0); attempt < inf.nextAttempt; attempt++ { key := repairRequestKey{ kind: sk.kind, slot: sk.slot, index: sk.index, - attempt: uint8(attempt), + attempt: attempt, } req, ok := c.outstanding[key] if !ok { @@ -890,7 +896,7 @@ func (c *repairClient) sendShredAttempt(conn *net.UDPConn, peers []gossip.Repair c.mu.Unlock() return false } - attempt := uint8(0) + attempt := uint64(0) if inf := c.inflight[sk]; inf != nil { attempt = inf.nextAttempt } @@ -900,16 +906,19 @@ func (c *repairClient) sendShredAttempt(conn *net.UDPConn, peers []gossip.Repair // microseconds; at tens of thousands of req/s, holding the lock across it // serialized every send against the response-processing path // (matchShredResponse needs the same lock) and could stall the UDP - // receive loop into kernel drops. The reserve is enough for coherence: a - // response for this attempt cannot arrive until after we WriteToUDP below, - // which is strictly after we register outstanding/byResponse. + // receive loop into kernel drops. Revalidate the reservation after signing: + // broadcast admission or reset may cancel it meanwhile. Register the request + // before the UDP write so this attempt's response can find it. c.addInflightLocked(sk, now) + reservation := c.inflight[sk] c.notePeerSentLocked(addrKey) c.mu.Unlock() undoReserve := func() { c.mu.Lock() - c.decInflightLocked(sk) + if c.inflight[sk] == reservation { + c.decInflightLocked(sk) + } c.notePeerSendAbortedLocked(addrKey) c.mu.Unlock() } @@ -933,6 +942,14 @@ func (c *repairClient) sendShredAttempt(conn *net.UDPConn, peers []gossip.Repair responseKey := repairResponseKey{addr: addrKey, nonce: nonce} c.mu.Lock() + // Broadcast/FEC admission or a slot reset may cancel this generation while + // signing. Do not resurrect its attempt key and overwrite a newer request. + // Cancellation only releases registered peer charges. + if c.inflight[sk] != reservation { + c.notePeerSendAbortedLocked(addrKey) + c.mu.Unlock() + return false + } c.outstanding[key] = outstandingRepairRequest{ key: key, nonce: nonce, @@ -950,18 +967,21 @@ func (c *repairClient) sendShredAttempt(conn *net.UDPConn, peers []gossip.Repair traceBinding = entryRepairTrace{Peer: peer.Addr.String(), Nonce: nonce} } if _, err := conn.WriteToUDP(packet, peer.Addr); err != nil { - traceRepairSend(slot, index, kind, attempt, traceStart, false, traceBinding) + traceRepairSend(slot, index, kind, uint8(attempt), traceStart, false, traceBinding) c.mu.Lock() - delete(c.outstanding, key) - delete(c.byResponse, responseKey) - c.decInflightLocked(sk) - c.notePeerSendAbortedLocked(addrKey) + // Cancellation/expiry may already have released the registered charge. + if req, ok := c.outstanding[key]; ok && c.inflight[sk] == reservation && req.nonce == nonce && req.addr == addrKey { + delete(c.outstanding, key) + delete(c.byResponse, responseKey) + c.decInflightLocked(sk) + c.notePeerSendAbortedLocked(addrKey) + } c.mu.Unlock() c.errors.Add(1) return false } - traceRepairSend(slot, index, kind, attempt, traceStart, true, traceBinding) + traceRepairSend(slot, index, kind, uint8(attempt), traceStart, true, traceBinding) c.requests.Add(1) return true } @@ -1510,12 +1530,15 @@ func (c *repairClient) stats() RepairStats { c.mu.Lock() outstanding := len(c.outstanding) responding := 0 + peerInflight := 0 for _, rec := range c.perPeer { + peerInflight += rec.inflight if rec.lastMatched.After(cutoff) { responding++ } } avgResponseMillis := int64(c.respLatencyEWMA * 1000) + perPeerCap := c.adaptivePerPeerCapLocked(peers) c.mu.Unlock() return RepairStats{ Requests: c.requests.Load(), @@ -1526,6 +1549,8 @@ func (c *repairClient) stats() RepairStats { Errors: c.errors.Load(), Outstanding: outstanding, Peers: peers, + PeerInflight: peerInflight, + PerPeerCap: perPeerCap, RespondingPeers: responding, LateResponses: c.lateResponses.Load(), TimeoutMillis: int64(time.Duration(c.timeoutNanos.Load()) / time.Millisecond), diff --git a/pkg/turbine/repair_attempt_test.go b/pkg/turbine/repair_attempt_test.go new file mode 100644 index 00000000..4d1211d2 --- /dev/null +++ b/pkg/turbine/repair_attempt_test.go @@ -0,0 +1,162 @@ +package turbine + +import ( + "crypto/ed25519" + "crypto/rand" + "io" + "net" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/Overclock-Validator/mithril/pkg/gossip" +) + +// Production retry/accounting durations run on synctest's virtual clock. +// Requests still traverse the real sender and loopback UDP; no request or +// capacity maps are fabricated or edited by the fixture. +func sendLongLivedRepairForTest(t *testing.T, target int) (*repairClient, *net.UDPConn, []gossip.RepairPeer) { + t.Helper() + conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + peers := []gossip.RepairPeer{{Addr: conn.LocalAddr().(*net.UDPAddr)}} + c, err := newRepairClient(nil, func() []gossip.RepairPeer { return peers }) + if err != nil { + t.Fatal(err) + } + copy(peers[0].Pubkey[:], c.identity.Public().(ed25519.PublicKey)) + pol := headPolicy(1) + start := time.Now() + for c.requests.Load() < uint64(target) && time.Since(start) < 5*time.Minute { + c.expireOutstanding(time.Now()) + if !c.sendShredAttempt(conn, peers, repairRequestWindowIndex, 50, 7, pol, c.accountingTimeout()) { + time.Sleep(repairScanInterval) + } + if len(c.byResponse) != len(c.outstanding) { + t.Fatal("nonce collision invalidates the wrap fixture") + } + } + sk := shredKey{kind: repairRequestWindowIndex, slot: 50, index: 7} + inf := c.inflight[sk] + if c.requests.Load() != uint64(target) || inf == nil || uint8(inf.nextAttempt) != uint8(target) { + t.Fatalf("long-lived generation premise: requests=%d inflight=%+v target=%d", c.requests.Load(), inf, target) + } + t.Logf("real sends=%d virtual elapsed=%s outstanding=%d nextAttempt=%d", target, time.Since(start), len(c.outstanding), inf.nextAttempt) + return c, conn, peers +} + +type repairNoncePause struct { + reader io.Reader + used atomic.Bool + started, release chan struct{} +} + +func (g *repairNoncePause) Read(p []byte) (int, error) { + if g.used.CompareAndSwap(false, true) { + close(g.started) + <-g.release + } + return g.reader.Read(p) +} + +func TestRepairAttemptReuseDuringSigning(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + peers := []gossip.RepairPeer{{Addr: conn.LocalAddr().(*net.UDPAddr)}} + c, err := newRepairClient(nil, func() []gossip.RepairPeer { return peers }) + if err != nil { + t.Fatal(err) + } + copy(peers[0].Pubkey[:], c.identity.Public().(ed25519.PublicKey)) + previous := rand.Reader + gate := &repairNoncePause{reader: previous, started: make(chan struct{}), release: make(chan struct{})} + rand.Reader = gate + defer func() { rand.Reader = previous }() + done := make(chan bool, 1) + // Response-driven followup uses bulk policy, concurrently with the + // periodic head sender. Hold that first reservation before registration. + go func() { + done <- c.sendShredAttempt(conn, peers, repairRequestWindowIndex, 50, 7, bulkPolicy(), c.accountingTimeout()) + }() + <-gate.started + start := time.Now() + for c.requests.Load() < 256 && time.Since(start) < 5*time.Minute { + c.expireOutstanding(time.Now()) + if !c.sendShredAttempt(conn, peers, repairRequestWindowIndex, 50, 7, headPolicy(1), c.accountingTimeout()) { + time.Sleep(repairScanInterval) + } + } + // Release and join before any assertion can terminate the bubble. + close(gate.release) + sent := <-done + if !sent || c.requests.Load() != 257 { + t.Fatalf("pending-reservation premise: sent=%v requests=%d", sent, c.requests.Load()) + } + t.Logf("pending signer released after %s; registered=%d peer_charges=%d", time.Since(start), len(c.outstanding), repairPeerCharges(c)) + time.Sleep(2 * repairMaxRequestTimeout) + c.expireOutstanding(time.Now()) + if len(c.outstanding) != 0 || repairPeerCharges(c) != 0 || len(c.byResponse) != 0 { + t.Errorf("byte attempt key reused while old reservation remained: outstanding=%d peer_inflight=%d response_keys=%d", + len(c.outstanding), repairPeerCharges(c), len(c.byResponse)) + } + }) +} + +func repairPeerCharges(c *repairClient) int { + total := 0 + for _, rec := range c.perPeer { + total += rec.inflight + } + return total +} + +func TestRepairAttemptWrapCancellation(t *testing.T) { + for _, target := range []int{255, 256, 257, 512, 513} { + synctest.Test(t, func(t *testing.T) { + c, _, _ := sendLongLivedRepairForTest(t, target) + c.satisfyDataShred(&Shred{Type: ShredTypeData, Slot: 50, Index: 7}) + if len(c.outstanding) != 0 || repairPeerCharges(c) != 0 || len(c.byResponse) != 0 { + t.Errorf("after admission-cancellation function: target=%d outstanding=%d peer_inflight=%d response_keys=%d; want all zero", + target, len(c.outstanding), repairPeerCharges(c), len(c.byResponse)) + } + }) + } +} + +func TestRepairAttemptWrapPoisonsReplacement(t *testing.T) { + for _, target := range []int{255, 256} { + synctest.Test(t, func(t *testing.T) { + c, conn, peers := sendLongLivedRepairForTest(t, target) + c.satisfyDataShred(&Shred{Type: ShredTypeData, Slot: 50, Index: 7}) + // Exercise same-shred sends after admission cancellation. A complete + // receiver ResetSlot also cancels all slot requests and is NOT this + // interleaving. Observe deadlines without changing bookkeeping. + deadline := time.Now().Add(repairMaxRequestTimeout) + for _, req := range c.outstanding { + if req.accountAt.Before(deadline) { + deadline = req.accountAt + } + } + if !c.sendShredAttempt(conn, peers, repairRequestWindowIndex, 50, 7, headPolicy(1), c.accountingTimeout()) { + t.Fatal("replacement did not send") + } + time.Sleep(deadline.Sub(time.Now())) + c.expireOutstanding(time.Now()) + c.sendShredAttempt(conn, peers, repairRequestWindowIndex, 50, 7, headPolicy(1), c.accountingTimeout()) + time.Sleep(2 * repairMaxRequestTimeout) + c.expireOutstanding(time.Now()) + if len(c.outstanding) != 0 || repairPeerCharges(c) != 0 || len(c.byResponse) != 0 { + t.Errorf("all accounting deadlines elapsed: target=%d outstanding=%d peer_inflight=%d response_keys=%d; want all zero", + target, len(c.outstanding), repairPeerCharges(c), len(c.byResponse)) + } + }) + } +} diff --git a/pkg/turbine/repair_capacity_stats_test.go b/pkg/turbine/repair_capacity_stats_test.go new file mode 100644 index 00000000..9d980042 --- /dev/null +++ b/pkg/turbine/repair_capacity_stats_test.go @@ -0,0 +1,50 @@ +package turbine + +import ( + "net" + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/gossip" +) + +func TestRepairCapacityStats(t *testing.T) { + conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + peers := []gossip.RepairPeer{{Addr: conn.LocalAddr().(*net.UDPAddr)}} + c, err := newRepairClient(nil, func() []gossip.RepairPeer { return peers }) + if err != nil { + t.Fatal(err) + } + c.setRateLimit(40) + check := func(outstanding, charged, perPeerCap int) { + t.Helper() + st := c.stats() + if st.Outstanding != outstanding || st.PeerInflight != charged || st.PerPeerCap != perPeerCap || st.Peers != 1 { + t.Errorf("stats=%+v want outstanding=%d peer_inflight=%d per_peer_cap=%d peers=1", st, outstanding, charged, perPeerCap) + } + t.Logf("repair_outstanding=%d repair_peer_inflight=%d repair_per_peer_cap=%d", st.Outstanding, st.PeerInflight, st.PerPeerCap) + } + send := func() int { + return c.sendTier(conn, peers, []SlotRepairRequest{{Slot: 50, MissingDataShreds: []uint32{7}}}, 1, nil, repairMinRequestTimeout) + } + check(0, 0, 31) + if sent := cancelDuringRepairSend(t, send, func() { + check(0, 1, 31) // Signing reserves capacity before registration. + c.satisfyDataShred(&Shred{Slot: 50, Index: 7, Type: ShredTypeData}) + }); sent != 0 { + t.Fatal("cancelled reservation sent a request") + } + check(0, 0, 31) + if send() != 1 { + t.Fatal("fresh request did not send") + } + check(1, 1, 31) + c.expireOutstanding(time.Now().Add(2 * repairMaxRequestTimeout)) + check(0, 0, 31) + c.setRateLimit(20) + check(0, 0, 16) +} diff --git a/pkg/turbine/repair_incoming_test.go b/pkg/turbine/repair_incoming_test.go new file mode 100644 index 00000000..fce7ad61 --- /dev/null +++ b/pkg/turbine/repair_incoming_test.go @@ -0,0 +1,151 @@ +package turbine + +import ( + "context" + "crypto/ed25519" + "encoding/binary" + "errors" + "net" + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/gossip" + repairproto "github.com/Overclock-Validator/mithril/pkg/repair" + "github.com/gagliardetto/solana-go" +) + +// This uses the real UDP read/parse/version/leader-signature/assembly path. +// The nonce gate selects the interleaving; no cancellation or accounting helper +// is called by the test. Synthetic test-leader keys never leave loopback. +func TestRepairIncomingAdmissionReservation(t *testing.T) { + for _, mode := range []string{"broadcast", "late-repair", "bad-signature", "wrong-version", "missing-leader"} { + t.Run(mode, func(t *testing.T) { + listen := func() *net.UDPConn { + t.Helper() + conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + if err := conn.Close(); err != nil && !errors.Is(err, net.ErrClosed) { + t.Error(err) + } + }) + return conn + } + conn, peerSocket := listen(), listen() + leader := testShredLeader(t) + gen := ShredGenerator{Slot: 93, ParentSlot: 92, Version: 7} + packets, _, _, _, err := gen.MakeShredsFromData(leader, []byte("signed incoming repair-race fixture"), false, solana.Hash{4}, 0, 0) + if err != nil { + t.Fatal(err) + } + packet := append([]byte(nil), packets[0]...) + peer := gossip.RepairPeer{Pubkey: gossip.Pubkey(leader.PublicKey()), Addr: peerSocket.LocalAddr().(*net.UDPAddr)} + peers := []gossip.RepairPeer{peer} + r := NewUDPReceiver(conn.LocalAddr().String()) + r.SetShredVersion(7) + r.SetLeaderForSlot(func(slot uint64) (solana.PublicKey, bool) { + return leader.PublicKey(), slot == 93 && mode != "missing-leader" + }) + if err := r.SetRepairPeerSource(nil, func() []gossip.RepairPeer { return peers }); err != nil { + t.Fatal(err) + } + c := r.repairClient + send := func(budget int) int { + return c.sendTier(conn, peers, []SlotRepairRequest{{Slot: 93, MissingDataShreds: []uint32{0}}}, budget, nil, repairMinRequestTimeout) + } + if mode == "late-repair" { + // Capture a genuinely sent request, expire it through the normal + // API, then return its nonce with the signed data over UDP while a + // replacement request is paused before registration. + if send(1) != 1 { + t.Fatal("initial repair request did not send") + } + if err := peerSocket.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + wire := make([]byte, 256) + n, _, err := peerSocket.ReadFromUDP(wire) + if err != nil { + t.Fatal(err) + } + sender := gossip.Pubkey(c.identity.Public().(ed25519.PublicKey)) + if n != 160 || !repairproto.VerifySignedRequest(wire[:n], sender) { + t.Fatal("initial request was not an authenticated repair packet") + } + // Signature(64), sender(32), recipient(32), timestamp(8) follow + // the four-byte request variant; the nonce starts at byte 140. + packet = binary.LittleEndian.AppendUint32(packet, binary.LittleEndian.Uint32(wire[140:144])) + c.expireOutstanding(time.Now().Add(2 * repairMaxRequestTimeout)) + } + if mode == "bad-signature" { + packet[0] ^= 1 + } + if mode == "wrong-version" { + r.SetShredVersion(8) + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- r.readPackets(ctx, conn, mode == "late-repair") }() + t.Cleanup(func() { + cancel() + if err := conn.Close(); err != nil { + t.Error(err) + } + select { + case err := <-done: + if err != nil { + t.Error(err) + } + case <-time.After(time.Second): + t.Error("receiver reader did not join") + } + }) + accepted := mode == "broadcast" || mode == "late-repair" + sent := cancelDuringRepairSend(t, func() int { return send(3) }, func() { + if _, err := peerSocket.WriteToUDP(packet, conn.LocalAddr().(*net.UDPAddr)); err != nil { + t.Error(err) + return + } + deadline := time.Now().Add(time.Second) + for time.Now().Before(deadline) { + detail, held := r.HeadShredDetail(93) + stats := r.Stats() + if accepted && held && detail.DataShreds == 1 { + return // Assembly follows the actual admission cancellation. + } + if !accepted && stats.SignatureErrors+stats.ShredVersionMismatch+stats.MissingLeaders == 1 { + return + } + time.Sleep(time.Millisecond) + } + t.Error("incoming packet did not finish the expected admission/rejection path") + }) + c.expireOutstanding(time.Now().Add(2 * repairMaxRequestTimeout)) + stats := r.Stats() + detail, held := r.HeadShredDetail(93) + if accepted { + if !held || detail.DataShreds != 1 || stats.SigVerifies != 1 || stats.SignatureErrors != 0 || stats.AssemblyErrors != 0 { + t.Fatalf("valid shred was not verified and assembled: held=%v detail=%+v stats=%+v", held, detail, stats) + } + if mode == "late-repair" && (stats.Repair.LateResponses != 1 || stats.UsefulRepairShreds != 1 || stats.RepairSocketUnmatched != 0) { + t.Fatalf("late response was not matched and admitted: %+v", stats) + } + } else if held || sent != 1 { + t.Fatalf("rejected incoming shred admitted or retired valid repair: held=%v sent=%d", held, sent) + } + c.mu.Lock() + charges := 0 + for _, record := range c.perPeer { + charges += record.inflight + } + outstanding, inflight := len(c.outstanding), len(c.inflight) + c.mu.Unlock() + t.Logf("mode=%s accepted=%v signature_verifies=%d sent_during_gate=%d outstanding=%d inflight=%d peer_charges=%d", mode, accepted, stats.SigVerifies, sent, outstanding, inflight, charges) + if charges != 0 || outstanding != 0 || inflight != 0 || (accepted && sent != 0) { + t.Fatalf("incoming admission resurrected cancelled repair: sent=%d outstanding=%d inflight=%d peer_charges=%d", sent, outstanding, inflight, charges) + } + }) + } +} diff --git a/pkg/turbine/repair_reservation_test.go b/pkg/turbine/repair_reservation_test.go new file mode 100644 index 00000000..60acac6a --- /dev/null +++ b/pkg/turbine/repair_reservation_test.go @@ -0,0 +1,150 @@ +package turbine + +import ( + "crypto/rand" + "io" + "net" + "testing" + "time" + + "github.com/Overclock-Validator/mithril/pkg/gossip" +) + +type repairNonceGate struct { + reader io.Reader + started chan struct{} + release chan struct{} +} + +func (g *repairNonceGate) Read(p []byte) (int, error) { + if g.started != nil { + started := g.started + g.started = nil + close(started) + <-g.release + } + return g.reader.Read(p) +} + +// The nonce read is after capacity reservation but before registration. Pause +// there to deterministically reproduce a broadcast/reset arriving while signing. +// These tests must not run in parallel because crypto/rand.Reader is global. +func cancelDuringRepairSend(t *testing.T, send func() int, cancel func()) int { + t.Helper() + started, release := make(chan struct{}), make(chan struct{}) + previous := rand.Reader + rand.Reader = &repairNonceGate{reader: previous, started: started, release: release} + defer func() { rand.Reader = previous }() + done := make(chan int, 1) + go func() { done <- send() }() + select { + case <-started: + case <-time.After(time.Second): + close(release) + <-done + t.Fatal("sender did not reach nonce generation") + } + cancel() + close(release) + return <-done +} + +func TestRepairReservationCancellation(t *testing.T) { + for _, when := range []string{"before", "during-shred", "during-slot", "during-replaced", "during-highest", "after", "timeout", "send-error"} { + t.Run(when, func(t *testing.T) { + conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + c := newPacingTestClient(t) + peers := []gossip.RepairPeer{{Addr: conn.LocalAddr().(*net.UDPAddr)}} + cancel := func() { c.satisfyDataShred(&Shred{Slot: 50, Index: 7, Type: ShredTypeData}) } + send := func() int { + if when == "during-highest" { + return c.sendTier(conn, peers, []SlotRepairRequest{{Slot: 50, NeedHighestDataShred: true}}, 3, nil, repairMinRequestTimeout) + } + return c.sendTier(conn, peers, []SlotRepairRequest{{Slot: 50, MissingDataShreds: []uint32{7}}}, 3, nil, repairMinRequestTimeout) + } + var sent int + switch when { + case "before": + cancel() + sent = send() + case "during-shred": + sent = cancelDuringRepairSend(t, send, cancel) + case "during-slot", "during-highest": + sent = cancelDuringRepairSend(t, send, func() { c.cancelSlotRequests(50) }) + case "during-replaced": + sent = cancelDuringRepairSend(t, send, func() { + cancel() + if send() != 1 { + t.Error("replacement request did not send") + } + }) + case "after": + sent = send() + cancel() + case "timeout": + sent = send() + case "send-error": + if err := conn.Close(); err != nil { + t.Fatal(err) + } + sent = send() + } + c.expireOutstanding(time.Now().Add(2 * repairMaxRequestTimeout)) + if len(c.outstanding) != 0 || len(c.inflight) != 0 { + t.Fatal("requests remained after cancellation/expiry") + } + for _, peer := range c.perPeer { + if peer.inflight != 0 { + t.Fatalf("sent=%d outstanding=0 but leaked peer capacity=%d", sent, peer.inflight) + } + } + if (when == "during-shred" || when == "during-slot" || when == "during-replaced" || when == "during-highest") && sent != 0 { + t.Fatalf("cancelled reservation still sent %d requests", sent) + } + if when == "send-error" && (sent != 0 || c.errors.Load() != 1) { + t.Fatal("send error was not accounted exactly once") + } + }) + } +} + +func TestRepairCancellationDoesNotStallAfterRateBackoff(t *testing.T) { + conn, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + peers := []gossip.RepairPeer{{Addr: conn.LocalAddr().(*net.UDPAddr)}} + c, err := newRepairClient(nil, func() []gossip.RepairPeer { return peers }) + if err != nil { + t.Fatal(err) + } + // Small rates make the same adaptive-cap transition cheap to reproduce: + // one peer's cap falls from 31 to 16 after the 40 -> 20 requests/s backoff. + c.setRateLimit(40) + for i := 0; i < 16; i++ { + cancelDuringRepairSend(t, func() int { + return c.sendTier(conn, peers, []SlotRepairRequest{{Slot: 50, MissingDataShreds: []uint32{7}}}, 3, nil, repairMinRequestTimeout) + }, func() { c.satisfyDataShred(&Shred{Slot: 50, Index: 7, Type: ShredTypeData}) }) + c.expireOutstanding(time.Now().Add(2 * repairMaxRequestTimeout)) + } + c.setRateLimit(20) + a := NewSlotAssembler() + a.PrioritizeRepairSlot(51) + previous := c.requests.Load() + c.repairOnce(conn, a) + if c.requests.Load() == previous { + t.Fatalf("repair stopped after rate backoff: requests=%d outstanding=%d", previous, len(c.outstanding)) + } + packet := make([]byte, 256) + if err := conn.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } + if _, _, err := conn.ReadFromUDP(packet); err != nil { + t.Fatalf("no repair UDP packet after backoff: %v", err) + } +}