diff --git a/api/api.go b/api/api.go index 2e078b2..afe97e2 100644 --- a/api/api.go +++ b/api/api.go @@ -123,3 +123,9 @@ type ConsensusUpdatesResponse struct { Applied []ApplyUpdate `json:"applied"` Reverted []RevertUpdate `json:"reverted"` } + +// DebugMineRequest is the request type for /debug/mine. +type DebugMineRequest struct { + Blocks int `json:"blocks"` + Address types.Address `json:"address"` +} diff --git a/api/api_test.go b/api/api_test.go index c2c2160..48e20a8 100644 --- a/api/api_test.go +++ b/api/api_test.go @@ -41,17 +41,24 @@ func testNetwork() (*consensus.Network, types.Block) { return n, genesisBlock } -func runServer(cm api.ChainManager, s api.Syncer, wm api.WalletManager) (*api.Client, func()) { +func runServer(t *testing.T, cm api.ChainManager, s api.Syncer, wm api.WalletManager) *api.Client { + t.Helper() + l, err := net.Listen("tcp", ":0") if err != nil { - panic(err) - } - go func() { - srv := api.NewServer(cm, s, wm) - http.Serve(l, jape.BasicAuth("password")(srv)) - }() - c := api.NewClient("http://"+l.Addr().String(), "password") - return c, func() { l.Close() } + t.Fatal("failed to listen:", err) + } + t.Cleanup(func() { l.Close() }) + + server := &http.Server{ + Handler: jape.BasicAuth("password")(api.NewServer(cm, s, wm, api.WithDebug(), api.WithLogger(zaptest.NewLogger(t)))), + ReadTimeout: 15 * time.Second, + WriteTimeout: 15 * time.Second, + } + t.Cleanup(func() { server.Close() }) + + go server.Serve(l) + return api.NewClient("http://"+l.Addr().String(), "password") } func waitForBlock(tb testing.TB, cm *chain.Manager, ws wallet.Store) { @@ -95,8 +102,7 @@ func TestWalletAdd(t *testing.T) { } defer wm.Close() - c, shutdown := runServer(cm, nil, wm) - defer shutdown() + c := runServer(t, cm, nil, wm) checkWalletResponse := func(wr api.WalletUpdateRequest, w wallet.Wallet, isUpdate bool) error { // check wallet @@ -287,8 +293,7 @@ func TestWallet(t *testing.T) { sav := wallet.NewSeedAddressVault(wallet.NewSeed(), 0, 20) // run server - c, shutdown := runServer(cm, s, wm) - defer shutdown() + c := runServer(t, cm, s, wm) w, err := c.AddWallet(api.WalletUpdateRequest{Name: "primary"}) if err != nil { t.Fatal(err) @@ -506,8 +511,7 @@ func TestAddresses(t *testing.T) { defer wm.Close() sav := wallet.NewSeedAddressVault(wallet.NewSeed(), 0, 20) - c, shutdown := runServer(cm, nil, wm) - defer shutdown() + c := runServer(t, cm, nil, wm) w, err := c.AddWallet(api.WalletUpdateRequest{Name: "primary"}) if err != nil { t.Fatal(err) @@ -702,8 +706,7 @@ func TestV2(t *testing.T) { } defer wm.Close() - c, shutdown := runServer(cm, nil, wm) - defer shutdown() + c := runServer(t, cm, nil, wm) primaryWallet, err := c.AddWallet(api.WalletUpdateRequest{Name: "primary"}) if err != nil { t.Fatal(err) @@ -942,8 +945,7 @@ func TestP2P(t *testing.T) { }) go s1.Run(context.Background()) defer s1.Close() - c1, shutdown := runServer(cm1, s1, wm1) - defer shutdown() + c1 := runServer(t, cm1, s1, wm1) w1, err := c1.AddWallet(api.WalletUpdateRequest{Name: "primary"}) if err != nil { t.Fatal(err) @@ -986,8 +988,7 @@ func TestP2P(t *testing.T) { }, syncer.WithLogger(zaptest.NewLogger(t))) go s2.Run(context.Background()) defer s2.Close() - c2, shutdown2 := runServer(cm2, s2, wm2) - defer shutdown2() + c2 := runServer(t, cm2, s2, wm2) w2, err := c2.AddWallet(api.WalletUpdateRequest{Name: "secondary"}) if err != nil { @@ -1248,8 +1249,7 @@ func TestConsensusUpdates(t *testing.T) { } defer wm.Close() - c, shutdown := runServer(cm, nil, wm) - defer shutdown() + c := runServer(t, cm, nil, wm) for i := 0; i < 10; i++ { b, ok := coreutils.MineBlock(cm, types.VoidAddress, time.Second) @@ -1283,3 +1283,65 @@ func TestConsensusUpdates(t *testing.T) { } } } + +func TestDebugMine(t *testing.T) { + log := zaptest.NewLogger(t) + n, genesisBlock := testNetwork() + + // create wallets + dbstore, tipState, err := chain.NewDBStore(chain.NewMemDB(), n, genesisBlock) + if err != nil { + t.Fatal(err) + } + cm := chain.NewManager(dbstore, tipState) + + l, err := net.Listen("tcp", ":0") + if err != nil { + t.Fatal(err) + } + defer l.Close() + + ws, err := sqlite.OpenDatabase(filepath.Join(t.TempDir(), "wallets.db"), log.Named("sqlite3")) + if err != nil { + t.Fatal(err) + } + defer ws.Close() + + ps, err := sqlite.NewPeerStore(ws) + if err != nil { + t.Fatal(err) + } + + s := syncer.New(l, cm, ps, gateway.Header{ + GenesisID: genesisBlock.ID(), + UniqueID: gateway.GenerateUniqueID(), + NetAddress: l.Addr().String(), + }) + defer s.Close() + go s.Run(context.Background()) + + wm, err := wallet.NewManager(cm, ws, wallet.WithLogger(log.Named("wallet"))) + if err != nil { + t.Fatal(err) + } + defer wm.Close() + + c := runServer(t, cm, s, wm) + + jc := jape.Client{ + BaseURL: c.BaseURL(), + Password: "password", + } + + err = jc.POST("/debug/mine", api.DebugMineRequest{ + Blocks: 5, + Address: types.VoidAddress, + }, nil) + if err != nil { + t.Fatal(err) + } + + if cm.Tip().Height != 5 { + t.Fatalf("expected tip height to be 5, got %v", cm.Tip().Height) + } +} diff --git a/api/client.go b/api/client.go index 3cd1685..776617b 100644 --- a/api/client.go +++ b/api/client.go @@ -34,6 +34,11 @@ func (c *Client) getNetwork() (*consensus.Network, error) { return c.n, nil } +// BaseURL returns the URL of the walletd server. +func (c *Client) BaseURL() string { + return c.c.BaseURL +} + // State returns information about the current state of the walletd daemon. func (c *Client) State() (resp StateResponse, err error) { err = c.c.GET("/state", &resp) diff --git a/api/mine.go b/api/mine.go new file mode 100644 index 0000000..9666d0c --- /dev/null +++ b/api/mine.go @@ -0,0 +1,81 @@ +package api + +import ( + "context" + "encoding/binary" + "errors" + + "go.sia.tech/core/types" +) + +// mineBlock constructs a block from the provided address and the transactions +// in the txpool, and attempts to find a nonce for it that meets the PoW target. +func mineBlock(ctx context.Context, cm ChainManager, addr types.Address) (types.Block, error) { + cs := cm.TipState() + txns := cm.PoolTransactions() + v2Txns := cm.V2PoolTransactions() + + b := types.Block{ + ParentID: cs.Index.ID, + Timestamp: types.CurrentTimestamp(), + MinerPayouts: []types.SiacoinOutput{{ + Value: cs.BlockReward(), + Address: addr, + }}, + } + + if cs.Index.Height >= cs.Network.HardforkV2.AllowHeight { + b.V2 = &types.V2BlockData{ + Height: cs.Index.Height + 1, + } + } + + var weight uint64 + for _, txn := range txns { + if weight += cs.TransactionWeight(txn); weight > cs.MaxBlockWeight() { + break + } + b.Transactions = append(b.Transactions, txn) + b.MinerPayouts[0].Value = b.MinerPayouts[0].Value.Add(txn.TotalFees()) + } + for _, txn := range v2Txns { + if weight += cs.V2TransactionWeight(txn); weight > cs.MaxBlockWeight() { + break + } + b.V2.Transactions = append(b.V2.Transactions, txn) + b.MinerPayouts[0].Value = b.MinerPayouts[0].Value.Add(txn.MinerFee) + } + if b.V2 != nil { + b.V2.Commitment = cs.Commitment(cs.TransactionsCommitment(b.Transactions, b.V2Transactions()), addr) + } + + b.Nonce = 0 + buf := make([]byte, 32+8+8+32) + binary.LittleEndian.PutUint64(buf[32:], b.Nonce) + binary.LittleEndian.PutUint64(buf[40:], uint64(b.Timestamp.Unix())) + if b.V2 != nil { + copy(buf[:32], "sia/id/block|") + copy(buf[48:], b.V2.Commitment[:]) + } else { + root := b.MerkleRoot() + copy(buf[:32], b.ParentID[:]) + copy(buf[48:], root[:]) + } + factor := cs.NonceFactor() + for types.BlockID(types.HashBytes(buf)).CmpWork(cs.ChildTarget) < 0 { + select { + case <-ctx.Done(): + return types.Block{}, ctx.Err() + default: + } + + // tip changed, abort mining + if cm.Tip() != cs.Index { + return types.Block{}, errors.New("tip changed") + } + + b.Nonce += factor + binary.LittleEndian.PutUint64(buf[32:], b.Nonce) + } + return b, nil +} diff --git a/api/server.go b/api/server.go index e5f6677..f8300ae 100644 --- a/api/server.go +++ b/api/server.go @@ -4,12 +4,14 @@ import ( "context" "errors" "net/http" + "net/http/pprof" "reflect" "runtime" "sync" "time" "go.sia.tech/jape" + "go.uber.org/zap" "lukechampine.com/frand" "go.sia.tech/core/consensus" @@ -21,11 +23,29 @@ import ( "go.sia.tech/walletd/wallet" ) +// A ServerOption sets an optional parameter for the server. +type ServerOption func(*server) + +// WithLogger sets the logger used by the server. +func WithLogger(log *zap.Logger) ServerOption { + return func(s *server) { + s.log = log + } +} + +// WithDebug enables debug endpoints. +func WithDebug() ServerOption { + return func(s *server) { + s.debugEnabled = true + } +} + type ( // A ChainManager manages blockchain and txpool state. ChainManager interface { UpdatesSince(types.ChainIndex, int) ([]chain.RevertUpdate, []chain.ApplyUpdate, error) + Tip() types.ChainIndex BestIndex(height uint64) (types.ChainIndex, bool) TipState() consensus.State AddBlocks([]types.Block) error @@ -85,11 +105,13 @@ type ( ) type server struct { - startTime time.Time + startTime time.Time + debugEnabled bool - cm ChainManager - s Syncer - wm WalletManager + log *zap.Logger + cm ChainManager + s Syncer + wm WalletManager // for walletsReserveHandler mu sync.Mutex @@ -814,17 +836,78 @@ func (s *server) outputsSiafundHandlerGET(jc jape.Context) { jc.Encode(output) } +func (s *server) debugMineHandler(jc jape.Context) { + var req DebugMineRequest + if jc.Decode(&req) != nil { + return + } + + log := s.log.Named("miner") + ctx := jc.Request.Context() + + for n := req.Blocks; n > 0; { + b, err := mineBlock(ctx, s.cm, req.Address) + if errors.Is(err, context.Canceled) { + return + } else if err != nil { + log.Warn("failed to mine block", zap.Error(err)) + } else if err := s.cm.AddBlocks([]types.Block{b}); err != nil { + log.Warn("failed to add block", zap.Error(err)) + } + + if b.V2 == nil { + s.s.BroadcastHeader(gateway.BlockHeader{ + ParentID: b.ParentID, + Nonce: b.Nonce, + Timestamp: b.Timestamp, + MerkleRoot: b.MerkleRoot(), + }) + } else { + s.s.BroadcastV2BlockOutline(gateway.OutlineBlock(b, s.cm.PoolTransactions(), s.cm.V2PoolTransactions())) + } + + log.Debug("mined block", zap.Stringer("blockID", b.ID())) + n-- + } +} + +func (s *server) pprofHandler(jc jape.Context) { + var handler string + if err := jc.DecodeParam("handler", &handler); err != nil { + return + } + + switch handler { + case "cmdline": + pprof.Cmdline(jc.ResponseWriter, jc.Request) + case "profile": + pprof.Profile(jc.ResponseWriter, jc.Request) + case "symbol": + pprof.Symbol(jc.ResponseWriter, jc.Request) + case "trace": + pprof.Trace(jc.ResponseWriter, jc.Request) + default: + pprof.Index(jc.ResponseWriter, jc.Request) + } + pprof.Index(jc.ResponseWriter, jc.Request) +} + // NewServer returns an HTTP handler that serves the walletd API. -func NewServer(cm ChainManager, s Syncer, wm WalletManager) http.Handler { +func NewServer(cm ChainManager, s Syncer, wm WalletManager, opts ...ServerOption) http.Handler { srv := server{ - startTime: time.Now(), + log: zap.NewNop(), + debugEnabled: false, + startTime: time.Now(), cm: cm, s: s, wm: wm, used: make(map[types.Hash256]bool), } - return jape.Mux(map[string]jape.Handler{ + for _, opt := range opts { + opt(&srv) + } + handlers := map[string]jape.Handler{ "GET /state": srv.stateHandler, "GET /consensus/network": srv.consensusNetworkHandler, @@ -872,5 +955,12 @@ func NewServer(cm ChainManager, s Syncer, wm WalletManager) http.Handler { "GET /outputs/siafund/:id": srv.outputsSiafundHandlerGET, "GET /events/:id": srv.eventsHandlerGET, - }) + } + + if srv.debugEnabled { + handlers["POST /debug/mine"] = srv.debugMineHandler + handlers["GET /debug/pprof/:handler"] = srv.pprofHandler + } + + return jape.Mux(handlers) } diff --git a/cmd/walletd/main.go b/cmd/walletd/main.go index d5de169..0915146 100644 --- a/cmd/walletd/main.go +++ b/cmd/walletd/main.go @@ -199,9 +199,11 @@ func main() { var minerAddrStr string var minerBlocks int + var enableDebug bool rootCmd := flagg.Root rootCmd.Usage = flagg.SimpleUsage(rootCmd, rootUsage) + rootCmd.BoolVar(&enableDebug, "debug", false, "enable debug mode with additional profiling and mining endpoints") rootCmd.StringVar(&cfg.Directory, "dir", cfg.Directory, "directory to store node state in") rootCmd.StringVar(&cfg.HTTP.Address, "http", cfg.HTTP.Address, "address to serve API on") @@ -314,7 +316,7 @@ func main() { log.Fatal("failed to parse index mode", zap.Error(err)) } - if err := runNode(ctx, cfg, log); err != nil { + if err := runNode(ctx, cfg, log, enableDebug); err != nil { log.Fatal("failed to run node", zap.Error(err)) } case versionCmd: diff --git a/cmd/walletd/node.go b/cmd/walletd/node.go index dedde8f..1004415 100644 --- a/cmd/walletd/node.go +++ b/cmd/walletd/node.go @@ -44,7 +44,7 @@ func setupUPNP(ctx context.Context, port uint16, log *zap.Logger) (string, error return d.ExternalIP() } -func runNode(ctx context.Context, cfg config.Config, log *zap.Logger) error { +func runNode(ctx context.Context, cfg config.Config, log *zap.Logger, enableDebug bool) error { var network *consensus.Network var genesisBlock types.Block var bootstrapPeers []string @@ -145,7 +145,13 @@ func runNode(ctx context.Context, cfg config.Config, log *zap.Logger) error { } defer wm.Close() - api := jape.BasicAuth(cfg.HTTP.Password)(api.NewServer(cm, s, wm)) + apiOpts := []api.ServerOption{ + api.WithLogger(log.Named("api")), + } + if enableDebug { + apiOpts = append(apiOpts, api.WithDebug()) + } + api := jape.BasicAuth(cfg.HTTP.Password)(api.NewServer(cm, s, wm, apiOpts...)) web := walletd.Handler() server := &http.Server{ Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/persist/sqlite/consensus.go b/persist/sqlite/consensus.go index 048d03b..4518933 100644 --- a/persist/sqlite/consensus.go +++ b/persist/sqlite/consensus.go @@ -331,7 +331,6 @@ func scanAddress(s scanner) (ab addressRef, err error) { func applyMatureSiacoinBalance(tx *txn, index types.ChainIndex, log *zap.Logger) error { log = log.With(zap.Uint64("maturityHeight", index.Height)) - log.Debug("applying mature siacoin balance") const query = `SELECT id, address_id, siacoin_value FROM siacoin_elements WHERE maturity_height=$1 AND matured=false AND spent_index_id IS NULL`