stakey/vspd
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Use a waitgroup to wait for shutdown tasks (#21)

Browse Source
This commit is contained in:
Jamie Holdstock 2020-05-18 09:05:10 +01:00 committed by GitHub
parent 1092df5224
commit 6ca2f620b7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 238 additions and 158 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
config.go
View File

@ -154,6 +154,7 @@ func loadConfig() (*config, error) {
// directory is updated, other variables need to be updated to
// reflect the new changes.
if preCfg.HomeDir != "" {
cfg.HomeDir = cleanAndExpandPath(cfg.HomeDir)
cfg.HomeDir, _ = filepath.Abs(preCfg.HomeDir)
if preCfg.ConfigFile == defaultConfigFile {

43
database/database.go
View File

@ -1,8 +1,10 @@
package database
import (
"context"
"encoding/binary"
"fmt"
"sync"
"time"
bolt "go.etcd.io/bbolt"
@ -25,20 +27,37 @@ var (
versionK = []byte("version")
)
// New initialises and returns a database. If no database file is found at the
// provided path, a new one will be created. Returns an open database which
// should always be closed after use.
func New(dbFile string) (*VspDatabase, error) {
// Open initialises and returns an open database. If no database file is found
// at the provided path, a new one will be created.
func Open(ctx context.Context, shutdownWg *sync.WaitGroup, dbFile string) (*VspDatabase, error) {
db, err := bolt.Open(dbFile, 0600, &bolt.Options{Timeout: 1 * time.Second})
if err != nil {
return nil, fmt.Errorf("unable to open db file: %v", err)
}
log.Debugf("Opened database file %s", dbFile)
// Add the graceful shutdown to the waitgroup.
shutdownWg.Add(1)
go func() {
// Wait until shutdown is signaled before shutting down.
<-ctx.Done()
log.Debug("Closing database...")
err := db.Close()
if err != nil {
log.Errorf("Error closing database: %v", err)
} else {
log.Debug("Database closed")
}
shutdownWg.Done()
}()
// Create all storage buckets of the VSP if they don't already exist.
var newDB bool
err = db.Update(func(tx *bolt.Tx) error {
if tx.Bucket(vspBktK) == nil {
newDB = true
log.Debug("Initialising new database")
// Create parent bucket.
vspBkt, err := tx.CreateBucket(vspBktK)
if err != nil {
@ -67,17 +86,5 @@ func New(dbFile string) (*VspDatabase, error) {
return nil, err
}
if newDB {
log.Debugf("Created new database %s", dbFile)
} else {
log.Debugf("Using existing database %s", dbFile)
}
return &VspDatabase{db: db}, nil
}
// Close releases all database resources. It will block waiting for any open
// transactions to finish before closing the database and returning.
func (vdb *VspDatabase) Close() error {
return vdb.db.Close()
}

12
database/database_test.go
View File

@ -1,7 +1,9 @@
package database
import (
"context"
"os"
"sync"
"testing"
)
@ -38,7 +40,9 @@ func TestDatabase(t *testing.T) {
for testName, test := range tests {
// Create a new blank database for each sub-test.
var err error
db, err = New(testDb)
var wg sync.WaitGroup
ctx, cancel := context.WithCancel(context.TODO())
db, err = Open(ctx, &wg, testDb)
if err != nil {
t.Fatalf("error creating test database: %v", err)
}
@ -46,8 +50,10 @@ func TestDatabase(t *testing.T) {
// Run the sub-test.
t.Run(testName, test)
// Close and remove test database after each sub-test.
db.Close()
// Request database shutdown and wait for it to complete.
cancel()
wg.Wait()
os.Remove(testDb)
}
}

8
log.go
View File

@ -9,6 +9,7 @@ import (
"github.com/jrick/logrotate/rotator"
"github.com/jholdstock/dcrvsp/database"
"github.com/jholdstock/dcrvsp/webapi"
)
// logWriter implements an io.Writer that outputs to both standard output and
@ -38,19 +39,22 @@ var (
// application shutdown.
logRotator *rotator.Rotator
log = backendLog.Logger("VSP")
dbLog = backendLog.Logger(" DB")
log = backendLog.Logger("VSP")
dbLog = backendLog.Logger(" DB")
apiLog = backendLog.Logger("API")
)
// Initialize package-global logger variables.
func init() {
database.UseLogger(dbLog)
webapi.UseLogger(apiLog)
}
// subsystemLoggers maps each subsystem identifier to its associated logger.
var subsystemLoggers = map[string]slog.Logger{
"VSP": log,
" DB": dbLog,
"API": apiLog,
}
// initLogRotator initializes the logging rotater to write logs to logFile and

91
main.go
View File

@ -4,21 +4,14 @@ import (
"context"
"errors"
"fmt"
"net"
"net/http"
"os"
"time"
"sync"
"github.com/jholdstock/dcrvsp/database"
"github.com/jholdstock/dcrvsp/webapi"
"github.com/jrick/wsrpc/v2"
)
var (
cfg *config
db *database.VspDatabase
nodeConnection *wsrpc.Client
)
func main() {
// Create a context that is cancelled when a shutdown request is received
// through an interrupt signal.
@ -36,78 +29,50 @@ func main() {
// opening the database, starting the webserver, and stopping all started
// services when the context is cancelled.
func run(ctx context.Context) error {
var err error
// Load config file and parse CLI args.
cfg, err = loadConfig()
cfg, err := loadConfig()
if err != nil {
// Don't use logger here because it may not be initialised.
fmt.Fprintf(os.Stderr, "Config error: %v", err)
return err
}
// Waitgroup for services to signal when they have shutdown cleanly.
var shutdownWg sync.WaitGroup
// Open database.
db, err = database.New(cfg.dbPath)
db, err := database.Open(ctx, &shutdownWg, cfg.dbPath)
if err != nil {
log.Errorf("Database error: %v", err)
requestShutdown()
shutdownWg.Wait()
return err
}
// Close database.
defer func() {
log.Debug("Closing database...")
err := db.Close()
if err != nil {
log.Errorf("Error closing database: %v", err)
} else {
log.Debug("Database closed")
}
}()
// Create TCP listener for webserver.
var listenConfig net.ListenConfig
listener, err := listenConfig.Listen(ctx, "tcp", cfg.Listen)
if err != nil {
log.Errorf("Failed to create tcp listener: %v", err)
return err
// TODO: Create real RPC client.
var rpc *wsrpc.Client
// Create and start webapi server.
apiCfg := webapi.Config{
SignKey: cfg.signKey,
PubKey: cfg.pubKey,
VSPFee: cfg.VSPFee,
NetParams: cfg.netParams.Params,
}
log.Infof("Listening on %s", cfg.Listen)
// Create webserver.
// TODO: Make releaseMode properly configurable.
// TODO: Make releaseMode properly configurable. Release mode enables very
// detailed webserver logging and live reloading of HTML templates.
releaseMode := true
srv := http.Server{
Handler: newRouter(releaseMode),
ReadTimeout: 5 * time.Second, // slow requests should not hold connections opened
WriteTimeout: 60 * time.Second, // hung responses must die
err = webapi.Start(ctx, shutdownRequestChannel, &shutdownWg, cfg.Listen, db, rpc, releaseMode, apiCfg)
if err != nil {
log.Errorf("Failed to initialise webapi: %v", err)
requestShutdown()
shutdownWg.Wait()
return err
}
// Start webserver.
go func() {
err = srv.Serve(listener)
// If the server dies for any reason other than ErrServerClosed (from
// graceful server.Shutdown), log the error and request dcrvsp be
// shutdown.
if err != nil && err != http.ErrServerClosed {
log.Errorf("Unexpected webserver error: %v", err)
requestShutdown()
}
}()
// Stop webserver.
defer func() {
log.Debug("Stopping webserver...")
// Give the webserver 5 seconds to finish what it is doing.
timeoutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := srv.Shutdown(timeoutCtx); err != nil {
log.Errorf("Failed to stop webserver cleanly: %v", err)
}
log.Debug("Webserver stopped")
}()
// Wait until shutdown is signaled before returning and running deferred
// shutdown tasks.
<-ctx.Done()
// Wait for shutdown tasks to complete before returning.
shutdownWg.Wait()
return ctx.Err()
}

52
router.go
View File

@ -1,52 +0,0 @@
package main
import (
"net/http"
"github.com/gin-gonic/gin"
)
func newRouter(releaseMode bool) *gin.Engine {
// With release mode enabled, gin will only read template files once and cache them.
// With release mode disabled, templates will be reloaded on the fly.
if releaseMode {
gin.SetMode(gin.ReleaseMode)
}
router := gin.New()
router.LoadHTMLGlob("templates/*")
// Recovery middleware handles any go panics generated while processing web
// requests. Ensures a 500 response is sent to the client rather than
// sending no response at all.
router.Use(gin.Recovery())
if !releaseMode {
// Logger middleware outputs very detailed logging of webserver requests
// to the terminal. Does not get logged to file.
router.Use(gin.Logger())
}
// Serve static web resources
router.Static("/public", "./public/")
router.GET("/", homepage)
api := router.Group("/api")
{
api.GET("/fee", fee)
api.POST("/feeaddress", feeAddress)
api.GET("/pubkey", pubKey)
api.POST("/payfee", payFee)
api.POST("/setvotebits", setVoteBits)
api.POST("/ticketstatus", ticketStatus)
}
return router
}
func homepage(c *gin.Context) {
c.HTML(http.StatusOK, "homepage.html", gin.H{
"Message": "Welcome to dcrvsp!",
})
}

6
run_tests.sh
View File

@ -28,7 +28,7 @@ golangci-lint run --disable-all --deadline=10m \
--enable=goimports \
--enable=misspell \
--enable=unparam \
--enable=deadcode \
--enable=unused \
--enable=asciicheck
# --enable=deadcode \
# --enable=errcheck \
# --enable=unused \
# --enable=errcheck \

2
helpers.go → webapi/helpers.go
View File

@ -1,4 +1,4 @@
package main
package webapi
import (
"github.com/decred/dcrd/chaincfg/v3"

2
helpers_test.go → webapi/helpers_test.go
View File

@ -1,4 +1,4 @@
package main
package webapi
import (
"testing"

26
webapi/log.go Normal file
View File

@ -0,0 +1,26 @@
package webapi
import (
"github.com/decred/slog"
)
// log is a logger that is initialized with no output filters. This
// means the package will not perform any logging by default until the caller
// requests it.
var log slog.Logger
// The default amount of logging is none.
func init() {
DisableLog()
}
// DisableLog disables all library log output. Logging output is disabled
// by default until UseLogger is called.
func DisableLog() {
log = slog.Disabled
}
// UseLogger uses a specified Logger to output package logging info.
func UseLogger(logger slog.Logger) {
log = logger
}

28
methods.go → webapi/methods.go
View File

@ -1,4 +1,4 @@
package main
package webapi
import (
"bytes"
@ -36,7 +36,7 @@ func sendJSONResponse(resp interface{}, c *gin.Context) {
return
}
sig := ed25519.Sign(cfg.signKey, dec)
sig := ed25519.Sign(cfg.SignKey, dec)
c.Writer.Header().Set("Content-Type", "application/json; charset=utf-8")
c.Writer.Header().Set("VSP-Signature", hex.EncodeToString(sig))
c.Writer.WriteHeader(http.StatusOK)
@ -46,7 +46,7 @@ func sendJSONResponse(resp interface{}, c *gin.Context) {
func pubKey(c *gin.Context) {
sendJSONResponse(pubKeyResponse{
Timestamp: time.Now().Unix(),
PubKey: cfg.pubKey,
PubKey: cfg.PubKey,
}, c)
}
@ -121,7 +121,7 @@ func feeAddress(c *gin.Context) {
c.AbortWithError(http.StatusBadRequest, errors.New("transaction does not have minimum confirmations"))
return
}
if resp.Confirmations > int64(uint32(cfg.netParams.TicketMaturity)+cfg.netParams.TicketExpiry) {
if resp.Confirmations > int64(uint32(cfg.NetParams.TicketMaturity)+cfg.NetParams.TicketExpiry) {
c.AbortWithError(http.StatusBadRequest, errors.New("transaction too old"))
return
}
@ -147,7 +147,7 @@ func feeAddress(c *gin.Context) {
}
// Get commitment address
addr, err := stake.AddrFromSStxPkScrCommitment(msgTx.TxOut[1].PkScript, cfg.netParams)
addr, err := stake.AddrFromSStxPkScrCommitment(msgTx.TxOut[1].PkScript, cfg.NetParams)
if err != nil {
c.AbortWithError(http.StatusInternalServerError, errors.New("failed to get commitment address"))
return
@ -155,7 +155,7 @@ func feeAddress(c *gin.Context) {
// verify message
message := fmt.Sprintf("vsp v3 getfeeaddress %s", msgTx.TxHash())
err = dcrutil.VerifyMessage(addr.Address(), signature, message, cfg.netParams.Params)
err = dcrutil.VerifyMessage(addr.Address(), signature, message, cfg.NetParams)
if err != nil {
c.AbortWithError(http.StatusBadRequest, errors.New("invalid signature"))
return
@ -219,7 +219,7 @@ func payFee(c *gin.Context) {
}
votingKey := payFeeRequest.VotingKey
votingWIF, err := dcrutil.DecodeWIF(votingKey, cfg.netParams.PrivateKeyID)
votingWIF, err := dcrutil.DecodeWIF(votingKey, cfg.NetParams.PrivateKeyID)
if err != nil {
c.AbortWithError(http.StatusInternalServerError, err)
return
@ -250,7 +250,7 @@ func payFee(c *gin.Context) {
findAddress:
for _, txOut := range feeTx.TxOut {
_, addresses, _, err := txscript.ExtractPkScriptAddrs(scriptVersion,
txOut.PkScript, cfg.netParams)
txOut.PkScript, cfg.NetParams)
if err != nil {
fmt.Printf("Extract: %v", err)
c.AbortWithError(http.StatusInternalServerError, err)
@ -279,13 +279,13 @@ findAddress:
c.AbortWithError(http.StatusInternalServerError, errors.New("database error"))
return
}
voteAddr, err := dcrutil.DecodeAddress(feeEntry.CommitmentAddress, cfg.netParams)
voteAddr, err := dcrutil.DecodeAddress(feeEntry.CommitmentAddress, cfg.NetParams)
if err != nil {
fmt.Printf("PayFee: DecodeAddress: %v", err)
c.AbortWithError(http.StatusInternalServerError, errors.New("database error"))
return
}
_, err = dcrutil.NewAddressPubKeyHash(dcrutil.Hash160(votingWIF.PubKey()), cfg.netParams,
_, err = dcrutil.NewAddressPubKeyHash(dcrutil.Hash160(votingWIF.PubKey()), cfg.NetParams,
dcrec.STEcdsaSecp256k1)
if err != nil {
fmt.Printf("PayFee: NewAddressPubKeyHash: %v", err)
@ -305,7 +305,7 @@ findAddress:
return
}
minFee := txrules.StakePoolTicketFee(sDiff, relayFee, int32(feeEntry.BlockHeight), cfg.VSPFee, cfg.netParams.Params)
minFee := txrules.StakePoolTicketFee(sDiff, relayFee, int32(feeEntry.BlockHeight), cfg.VSPFee, cfg.NetParams)
if feeAmount < minFee {
fmt.Printf("too cheap: %v %v", feeAmount, minFee)
c.AbortWithError(http.StatusInternalServerError, fmt.Errorf("dont get cheap on me, dodgson (sent:%v required:%v)", feeAmount, minFee))
@ -411,7 +411,7 @@ func setVoteBits(c *gin.Context) {
// votebits
voteBits := setVoteBitsRequest.VoteBits
if !isValidVoteBits(cfg.netParams.Params, currentVoteVersion(cfg.netParams.Params), voteBits) {
if !isValidVoteBits(cfg.NetParams, currentVoteVersion(cfg.NetParams), voteBits) {
c.AbortWithError(http.StatusBadRequest, errors.New("invalid votebits"))
return
}
@ -424,7 +424,7 @@ func setVoteBits(c *gin.Context) {
// verify message
message := fmt.Sprintf("vsp v3 setvotebits %d %s %d", setVoteBitsRequest.Timestamp, txHash, voteBits)
err = dcrutil.VerifyMessage(addr, signature, message, cfg.netParams.Params)
err = dcrutil.VerifyMessage(addr, signature, message, cfg.NetParams)
if err != nil {
c.AbortWithError(http.StatusBadRequest, errors.New("message did not pass verification"))
return
@ -478,7 +478,7 @@ func ticketStatus(c *gin.Context) {
// verify message
message := fmt.Sprintf("vsp v3 ticketstatus %d %s", ticketStatusRequest.Timestamp, ticketHashStr)
err = dcrutil.VerifyMessage(addr, signature, message, cfg.netParams.Params)
err = dcrutil.VerifyMessage(addr, signature, message, cfg.NetParams)
if err != nil {
c.AbortWithError(http.StatusBadRequest, errors.New("invalid signature"))
return

2
responses.go → webapi/responses.go
View File

@ -1,4 +1,4 @@
package main
package webapi
type pubKeyResponse struct {
Timestamp int64 `json:"timestamp"`

123
webapi/server.go Normal file
View File

@ -0,0 +1,123 @@
package webapi
import (
"context"
"crypto/ed25519"
"net"
"net/http"
"sync"
"time"
"github.com/decred/dcrd/chaincfg/v3"
"github.com/gin-gonic/gin"
"github.com/jholdstock/dcrvsp/database"
"github.com/jrick/wsrpc/v2"
)
type Config struct {
SignKey ed25519.PrivateKey
PubKey ed25519.PublicKey
VSPFee float64
NetParams *chaincfg.Params
}
var cfg Config
var db *database.VspDatabase
var nodeConnection *wsrpc.Client
func Start(ctx context.Context, requestShutdownChan chan struct{}, shutdownWg *sync.WaitGroup,
listen string, db *database.VspDatabase, nodeConnection *wsrpc.Client, releaseMode bool, config Config) error {
// Create TCP listener.
var listenConfig net.ListenConfig
listener, err := listenConfig.Listen(ctx, "tcp", listen)
if err != nil {
return err
}
log.Infof("Listening on %s", listen)
srv := http.Server{
Handler: router(releaseMode),
ReadTimeout: 5 * time.Second, // slow requests should not hold connections opened
WriteTimeout: 60 * time.Second, // hung responses must die
}
// Add the graceful shutdown to the waitgroup.
shutdownWg.Add(1)
go func() {
// Wait until shutdown is signaled before shutting down.
<-ctx.Done()
log.Debug("Stopping webserver...")
// Give the webserver 5 seconds to finish what it is doing.
timeoutCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
if err := srv.Shutdown(timeoutCtx); err != nil {
log.Errorf("Failed to stop webserver cleanly: %v", err)
} else {
log.Debug("Webserver stopped")
}
shutdownWg.Done()
}()
// Start webserver.
go func() {
err = srv.Serve(listener)
// If the server dies for any reason other than ErrServerClosed (from
// graceful server.Shutdown), log the error and request dcrvsp be
// shutdown.
if err != nil && err != http.ErrServerClosed {
log.Errorf("Unexpected webserver error: %v", err)
requestShutdownChan <- struct{}{}
}
}()
cfg = config
return nil
}
func router(releaseMode bool) *gin.Engine {
// With release mode enabled, gin will only read template files once and cache them.
// With release mode disabled, templates will be reloaded on the fly.
if releaseMode {
gin.SetMode(gin.ReleaseMode)
}
router := gin.New()
router.LoadHTMLGlob("templates/*")
// Recovery middleware handles any go panics generated while processing web
// requests. Ensures a 500 response is sent to the client rather than
// sending no response at all.
router.Use(gin.Recovery())
if !releaseMode {
// Logger middleware outputs very detailed logging of webserver requests
// to the terminal. Does not get logged to file.
router.Use(gin.Logger())
}
// Serve static web resources
router.Static("/public", "./public/")
router.GET("/", homepage)
api := router.Group("/api")
{
api.GET("/fee", fee)
api.POST("/feeaddress", feeAddress)
api.GET("/pubkey", pubKey)
api.POST("/payfee", payFee)
api.POST("/setvotebits", setVoteBits)
api.POST("/ticketstatus", ticketStatus)
}
return router
}
func homepage(c *gin.Context) {
c.HTML(http.StatusOK, "homepage.html", gin.H{
"Message": "Welcome to dcrvsp!",
})
}