diff --git a/client/asset/btc/btc.go b/client/asset/btc/btc.go index 75da49f62e..db5a6510c7 100644 --- a/client/asset/btc/btc.go +++ b/client/asset/btc/btc.go @@ -17,6 +17,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "decred.org/dcrdex/client/asset" @@ -39,6 +40,7 @@ const ( // rpcclient.Client's GetBlockVerboseTx appears to be busted. methodGetBlockVerboseTx = "getblock" methodGetNetworkInfo = "getnetworkinfo" + methodGetBlockchainInfo = "getblockchaininfo" // BipID is the BIP-0044 asset ID. BipID = 0 @@ -348,6 +350,7 @@ type ExchangeWallet struct { log dex.Logger symbol string tipChange func(error) + tipAtConnect int64 minNetworkVersion uint64 fallbackFeeRate uint64 redeemConfTarget uint64 @@ -519,6 +522,7 @@ func (btc *ExchangeWallet) Connect(ctx context.Context) (*sync.WaitGroup, error) if err != nil { return nil, fmt.Errorf("error initializing best block for %s: %v", btc.symbol, err) } + atomic.StoreInt64(&btc.tipAtConnect, btc.currentTip.height) var wg sync.WaitGroup wg.Add(1) go func() { @@ -541,6 +545,41 @@ func (btc *ExchangeWallet) shutdown() { btc.findRedemptionMtx.Unlock() } +// getBlockchainInfoResult models the data returned from the getblockchaininfo +// command. +type getBlockchainInfoResult struct { + Blocks int64 `json:"blocks"` + Headers int64 `json:"headers"` + BestBlockHash string `json:"bestblockhash"` + InitialBlockDownload bool `json:"initialblockdownload"` +} + +// getBlockchainInfo sends the getblockchaininfo request and returns the result. +func (btc *ExchangeWallet) getBlockchainInfo() (*getBlockchainInfoResult, error) { + chainInfo := new(getBlockchainInfoResult) + err := btc.wallet.call(methodGetBlockchainInfo, nil, chainInfo) + if err != nil { + return nil, err + } + return chainInfo, nil +} + +// SyncStatus is information about the blockchain sync status. +func (btc *ExchangeWallet) SyncStatus() (bool, float32, error) { + chainInfo, err := btc.getBlockchainInfo() + if err != nil { + return false, 0, fmt.Errorf("getblockchaininfo error: %w", err) + } + toGo := chainInfo.Headers - chainInfo.Blocks + if chainInfo.InitialBlockDownload || toGo > 1 { + ogTip := atomic.LoadInt64(&btc.tipAtConnect) + totalToSync := chainInfo.Headers - ogTip + progress := 1 - (float32(toGo) / float32(totalToSync)) + return false, progress, nil + } + return true, 1, nil +} + // Balance returns the total available funds in the wallet. Part of the // asset.Wallet interface. func (btc *ExchangeWallet) Balance() (*asset.Balance, error) { diff --git a/client/asset/btc/btc_test.go b/client/asset/btc/btc_test.go index fe9484b09e..8753221153 100644 --- a/client/asset/btc/btc_test.go +++ b/client/asset/btc/btc_test.go @@ -2077,3 +2077,46 @@ func testSendEdges(t *testing.T, segwit bool) { } } } + +func TestSyncStatus(t *testing.T) { + wallet, node, shutdown := tNewWallet(false) + defer shutdown() + node.rawRes[methodGetBlockchainInfo] = mustMarshal(t, &getBlockchainInfoResult{ + Headers: 100, + Blocks: 99, + }) + + synced, progress, err := wallet.SyncStatus() + if err != nil { + t.Fatalf("SyncStatus error (synced expected): %v", err) + } + if !synced { + t.Fatalf("synced = false for 1 block to go") + } + if progress < 1 { + t.Fatalf("progress not complete when loading last block") + } + + node.rawErr[methodGetBlockchainInfo] = tErr + _, _, err = wallet.SyncStatus() + if err == nil { + t.Fatalf("SyncStatus error not propagated") + } + node.rawErr[methodGetBlockchainInfo] = nil + + wallet.tipAtConnect = 100 + node.rawRes[methodGetBlockchainInfo] = mustMarshal(t, &getBlockchainInfoResult{ + Headers: 200, + Blocks: 150, + }) + synced, progress, err = wallet.SyncStatus() + if err != nil { + t.Fatalf("SyncStatus error (half-synced): %v", err) + } + if synced { + t.Fatalf("synced = true for 50 blocks to go") + } + if progress > 0.500001 || progress < 0.4999999 { + t.Fatalf("progress out of range. Expected 0.5, got %.2f", progress) + } +} diff --git a/client/asset/dcr/dcr.go b/client/asset/dcr/dcr.go index a4c87c1b27..52b52f2d53 100644 --- a/client/asset/dcr/dcr.go +++ b/client/asset/dcr/dcr.go @@ -19,6 +19,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "decred.org/dcrdex/client/asset" @@ -131,6 +132,7 @@ var ( // rpcClient is an rpcclient.Client, or a stub for testing. type rpcClient interface { EstimateSmartFee(confirmations int64, mode chainjson.EstimateSmartFeeMode) (float64, error) + GetBlockChainInfo() (*chainjson.GetBlockChainInfoResult, error) SendRawTransaction(tx *wire.MsgTx, allowHighFees bool) (*chainhash.Hash, error) GetTxOut(txHash *chainhash.Hash, index uint32, mempool bool) (*chainjson.GetTxOutResult, error) GetBalanceMinConf(account string, minConfirms int) (*walletjson.GetBalanceResult, error) @@ -352,6 +354,7 @@ type ExchangeWallet struct { log dex.Logger acct string tipChange func(error) + tipAtConnect int64 fallbackFeeRate uint64 redeemConfTarget uint64 useSplitTx bool @@ -521,6 +524,7 @@ func (dcr *ExchangeWallet) Connect(ctx context.Context) (*sync.WaitGroup, error) if err != nil { return nil, fmt.Errorf("error initializing best block for DCR: %v", err) } + atomic.StoreInt64(&dcr.tipAtConnect, dcr.currentTip.height) dcr.log.Infof("Connected to dcrwallet (JSON-RPC API v%s) proxying dcrd (JSON-RPC API v%s) on %v", walletSemver, nodeSemver, curnet) @@ -1847,6 +1851,22 @@ func (dcr *ExchangeWallet) shutdown() { } } +// SyncStatus is information about the blockchain sync status. +func (dcr *ExchangeWallet) SyncStatus() (bool, float32, error) { + chainInfo, err := dcr.node.GetBlockChainInfo() + if err != nil { + return false, 0, fmt.Errorf("getblockchaininfo error: %w", err) + } + toGo := chainInfo.Headers - chainInfo.Blocks + if chainInfo.InitialBlockDownload || toGo > 1 { + ogTip := atomic.LoadInt64(&dcr.tipAtConnect) + totalToSync := chainInfo.Headers - ogTip + progress := 1 - (float32(toGo) / float32(totalToSync)) + return false, progress, nil + } + return true, 1, nil +} + // Combines the RPC type with the spending input information. type compositeUTXO struct { rpc walletjson.ListUnspentResult diff --git a/client/asset/dcr/dcr_test.go b/client/asset/dcr/dcr_test.go index e48c01eae3..9f2d657f52 100644 --- a/client/asset/dcr/dcr_test.go +++ b/client/asset/dcr/dcr_test.go @@ -173,41 +173,43 @@ func signFunc(msgTx *wire.MsgTx, scriptSize int) (*wire.MsgTx, bool, error) { } type tRPCClient struct { - sendRawHash *chainhash.Hash - sendRawErr error - sentRawTx *wire.MsgTx - txOutRes map[outPoint]*chainjson.GetTxOutResult - txOutErr error - bestBlockErr error - mempool []*chainhash.Hash - mempoolErr error - rawTx *chainjson.TxRawResult - rawTxErr error - unspent []walletjson.ListUnspentResult - unspentErr error - balanceResult *walletjson.GetBalanceResult - balanceErr error - lockUnspentErr error - changeAddr dcrutil.Address - changeAddrErr error - newAddr dcrutil.Address - newAddrErr error - signFunc func(tx *wire.MsgTx) (*wire.MsgTx, bool, error) - privWIF *dcrutil.WIF - privWIFErr error - walletTx *walletjson.GetTransactionResult - walletTxErr error - lockErr error - passErr error - disconnected bool - rawRes map[string]json.RawMessage - rawErr map[string]error - blockchainMtx sync.RWMutex - verboseBlocks map[string]*chainjson.GetBlockVerboseResult - mainchain map[int64]*chainhash.Hash - lluCoins []walletjson.ListUnspentResult // Returned from ListLockUnspent - lockedCoins []*wire.OutPoint // Last submitted to LockUnspent - listLockedErr error + sendRawHash *chainhash.Hash + sendRawErr error + sentRawTx *wire.MsgTx + txOutRes map[outPoint]*chainjson.GetTxOutResult + txOutErr error + bestBlockErr error + mempool []*chainhash.Hash + mempoolErr error + rawTx *chainjson.TxRawResult + rawTxErr error + unspent []walletjson.ListUnspentResult + unspentErr error + balanceResult *walletjson.GetBalanceResult + balanceErr error + lockUnspentErr error + changeAddr dcrutil.Address + changeAddrErr error + newAddr dcrutil.Address + newAddrErr error + signFunc func(tx *wire.MsgTx) (*wire.MsgTx, bool, error) + privWIF *dcrutil.WIF + privWIFErr error + walletTx *walletjson.GetTransactionResult + walletTxErr error + lockErr error + passErr error + disconnected bool + rawRes map[string]json.RawMessage + rawErr map[string]error + blockchainMtx sync.RWMutex + verboseBlocks map[string]*chainjson.GetBlockVerboseResult + mainchain map[int64]*chainhash.Hash + lluCoins []walletjson.ListUnspentResult // Returned from ListLockUnspent + lockedCoins []*wire.OutPoint // Last submitted to LockUnspent + listLockedErr error + blockchainInfo *chainjson.GetBlockChainInfoResult + blockchainInfoErr error } func defaultSignFunc(tx *wire.MsgTx) (*wire.MsgTx, bool, error) { return tx, true, nil } @@ -236,6 +238,10 @@ func (c *tRPCClient) EstimateSmartFee(confirmations int64, mode chainjson.Estima return optimalRate, nil // optimalFeeRate: 22 atoms/byte = 0.00022 DCR/KB * 1e8 atoms/DCR * 1e-3 KB/Byte } +func (c *tRPCClient) GetBlockChainInfo() (*chainjson.GetBlockChainInfoResult, error) { + return c.blockchainInfo, c.blockchainInfoErr +} + func (c *tRPCClient) SendRawTransaction(tx *wire.MsgTx, allowHighFees bool) (*chainhash.Hash, error) { c.sentRawTx = tx if c.sendRawErr == nil && c.sendRawHash == nil { @@ -1828,3 +1834,46 @@ func TestSendEdges(t *testing.T) { } } } + +func TestSyncStatus(t *testing.T) { + wallet, node, shutdown := tNewWallet() + defer shutdown() + node.blockchainInfo = &chainjson.GetBlockChainInfoResult{ + Headers: 100, + Blocks: 99, + } + + synced, progress, err := wallet.SyncStatus() + if err != nil { + t.Fatalf("SyncStatus error (synced expected): %v", err) + } + if !synced { + t.Fatalf("synced = false for 1 block to go") + } + if progress < 1 { + t.Fatalf("progress not complete when loading last block") + } + + node.blockchainInfoErr = tErr + _, _, err = wallet.SyncStatus() + if err == nil { + t.Fatalf("SyncStatus error not propagated") + } + node.blockchainInfoErr = nil + + wallet.tipAtConnect = 100 + node.blockchainInfo = &chainjson.GetBlockChainInfoResult{ + Headers: 200, + Blocks: 150, + } + synced, progress, err = wallet.SyncStatus() + if err != nil { + t.Fatalf("SyncStatus error (half-synced): %v", err) + } + if synced { + t.Fatalf("synced = true for 50 blocks to go") + } + if progress > 0.500001 || progress < 0.4999999 { + t.Fatalf("progress out of range. Expected 0.5, got %.2f", progress) + } +} diff --git a/client/asset/interface.go b/client/asset/interface.go index a67d899aee..49d77deaa0 100644 --- a/client/asset/interface.go +++ b/client/asset/interface.go @@ -150,6 +150,8 @@ type Wallet interface { Withdraw(address string, value uint64) (Coin, error) // ValidateSecret checks that the secret hashes to the secret hash. ValidateSecret(secret, secretHash []byte) bool + // SyncStatus is information about the blockchain sync status. + SyncStatus() (synced bool, progress float32, err error) } // Balance is categorized information about a wallet's balance. diff --git a/client/core/core.go b/client/core/core.go index 298f070c0e..b4eeaf92f1 100644 --- a/client/core/core.go +++ b/client/core/core.go @@ -52,6 +52,9 @@ var ( aYear = time.Hour * 24 * 365 // The coin waiters will query for transaction data every recheckInterval. recheckInterval = time.Second * 5 + // When waiting for a wallet to sync, a SyncStatus check will be performed + // ever syncTickerPeriod. var instead of const for testing purposes. + syncTickerPeriod = 10 * time.Second ) // dexConnection is the websocket connection and the DEX configuration. @@ -1031,13 +1034,58 @@ func (c *Core) connectedWallet(assetID uint32) (*xcWallet, error) { return wallet, nil } +func (c *Core) connectWallet(w *xcWallet) error { + err := w.Connect(c.ctx) + if err != nil { + return newError(walletErr, "error connecting %s wallet: %v", unbip(w.AssetID), err) + } + // If the wallet is not synced, start a loop to check the sync status until + // it is. + if !w.synced { + // If the wallet is shut down before sync is complete, exit the syncer + // loop. + innerCtx, cancel := context.WithCancel(c.ctx) + go func() { + w.connector.Wait() + cancel() + }() + // The syncer loop. + go func() { + ticker := time.NewTicker(syncTickerPeriod) + defer ticker.Stop() + for { + select { + case <-ticker.C: + synced, progress, err := w.SyncStatus() + if err != nil { + c.log.Errorf("error monitoring sync status for %s", unbip(w.AssetID)) + return + } + w.mtx.Lock() + w.synced = synced + w.syncProgress = progress + w.mtx.Unlock() + c.notify(newWalletStateNote(w.state())) + if synced { + return + } + + case <-innerCtx.Done(): + return + } + } + }() + } + return nil +} + // Connect to the wallet if not already connected. Unlock the wallet if not // already unlocked. func (c *Core) connectAndUnlock(crypter encrypt.Crypter, wallet *xcWallet) error { if !wallet.connected() { - err := wallet.Connect(c.ctx) + err := c.connectWallet(wallet) if err != nil { - return newError(walletErr, "error connecting %s wallet: %v", unbip(wallet.AssetID), err) + return err } } if !wallet.unlocked() { @@ -2423,15 +2471,28 @@ func (c *Core) prepareTrackedTrade(dc *dexConnection, form *TradeForm, crypter e if err != nil { return nil, 0, err } - fromWallet, toWallet := wallets.fromWallet, wallets.toWallet - err = c.connectAndUnlock(crypter, fromWallet) + + prepareWallet := func(w *xcWallet) error { + err := c.connectAndUnlock(crypter, w) + if err != nil { + return fmt.Errorf("%s connectAndUnlock error: %w", wallets.fromAsset.Symbol, err) + } + w.mtx.RLock() + defer w.mtx.RUnlock() + if !w.synced { + return fmt.Errorf("%s still syncing. progress = %.2f", unbip(w.AssetID), w.syncProgress) + } + return nil + } + + err = prepareWallet(fromWallet) if err != nil { - return nil, 0, fmt.Errorf("%s connectAndUnlock error: %w", wallets.fromAsset.Symbol, err) + return nil, 0, err } - err = c.connectAndUnlock(crypter, toWallet) + err = prepareWallet(toWallet) if err != nil { - return nil, 0, fmt.Errorf("%s connectAndUnlock error: %w", wallets.toAsset.Symbol, err) + return nil, 0, err } // Get an address for the swap contract. diff --git a/client/core/core_test.go b/client/core/core_test.go index 88d1f0fdaa..14cbc301b7 100644 --- a/client/core/core_test.go +++ b/client/core/core_test.go @@ -536,20 +536,26 @@ type TXCWallet struct { fundingCoinErr error lockErr error changeCoin *tCoin + syncStatus func() (bool, float32, error) + connectWG *sync.WaitGroup } func newTWallet(assetID uint32) (*xcWallet, *TXCWallet) { w := &TXCWallet{ changeCoin: &tCoin{id: encode.RandomBytes(36)}, + syncStatus: func() (synced bool, progress float32, err error) { return true, 1, nil }, + connectWG: &sync.WaitGroup{}, } return &xcWallet{ - Wallet: w, - connector: dex.NewConnectionMaster(w), - AssetID: assetID, - lockTime: time.Now().Add(time.Hour), - hookedUp: true, - dbID: encode.Uint32Bytes(assetID), - encPW: []byte{0x01}, + Wallet: w, + connector: dex.NewConnectionMaster(w), + AssetID: assetID, + lockTime: time.Now().Add(time.Hour), + hookedUp: true, + dbID: encode.Uint32Bytes(assetID), + encPW: []byte{0x01}, + synced: true, + syncProgress: 1, }, w } @@ -558,7 +564,7 @@ func (w *TXCWallet) Info() *asset.WalletInfo { } func (w *TXCWallet) Connect(ctx context.Context) (*sync.WaitGroup, error) { - return &sync.WaitGroup{}, w.connectErr + return w.connectWG, w.connectErr } func (w *TXCWallet) Run(ctx context.Context) { <-ctx.Done() } @@ -687,6 +693,10 @@ func (w *TXCWallet) ValidateSecret(secret, secretHash []byte) bool { return !w.badSecret } +func (w *TXCWallet) SyncStatus() (synced bool, progress float32, err error) { + return w.syncStatus() +} + func (w *TXCWallet) setConfs(confs uint32) { w.mtx.Lock() w.payFeeCoin.confs = confs @@ -2129,6 +2139,15 @@ func TestTrade(t *testing.T) { ensureErr("signature error") tDcrWallet.signCoinErr = nil + // Sync-in-progress error + dcrWallet.synced = false + ensureErr("base not synced") + dcrWallet.synced = true + + btcWallet.synced = false + ensureErr("quote not synced") + btcWallet.synced = true + // LimitRoute error rig.ws.reqErr = tErr ensureErr("Request error") @@ -5553,3 +5572,57 @@ func TestSuspectTrades(t *testing.T) { t.Fatalf("suspect redeem matches not run or not run separately. expected 2 new calls to Redeem, got %d", tBtcWallet.redeemCounter-1) } } + +func TestWalletSyncing(t *testing.T) { + rig := newTestRig() + tCore := rig.core + + noteFeed := tCore.NotificationFeed() + dcrWallet, tDcrWallet := newTWallet(tDCR.ID) + tDcrWallet.connectWG.Add(1) + defer tDcrWallet.connectWG.Done() + dcrWallet.synced = false + dcrWallet.syncProgress = 0 + dcrWallet.Connect(tCtx) + + tStart := time.Now() + testDuration := 100 * time.Millisecond + syncTickerPeriod = 10 * time.Millisecond + + tDcrWallet.syncStatus = func() (bool, float32, error) { + progress := float32(float64(time.Since(tStart)) / float64(testDuration)) + if progress >= 1 { + return true, 1, nil + } + return false, progress, nil + } + + err := tCore.connectWallet(dcrWallet) + if err != nil { + t.Fatalf("connectWallet error: %v", err) + } + + timeout := time.NewTimer(time.Second).C + var progressNotes int +out: + for { + select { + case note := <-noteFeed: + walletNote, ok := note.(*WalletStateNote) + if !ok { + continue + } + if walletNote.Wallet.Synced { + break out + } + progressNotes++ + case <-timeout: + t.Fatalf("timed out waiting for synced wallet note. Received %d progress notes", progressNotes) + } + } + // Should get 9 notes, but just make sure we get at least half of them to + // avoid github vm false positives. + if progressNotes < 5 { + t.Fatalf("expected 23 progress notes, got %d", progressNotes) + } +} diff --git a/client/core/types.go b/client/core/types.go index e4b4b5b9d9..cda49385c6 100644 --- a/client/core/types.go +++ b/client/core/types.go @@ -83,14 +83,16 @@ type WalletBalance struct { // WalletState is the current status of an exchange wallet. type WalletState struct { - Symbol string `json:"symbol"` - AssetID uint32 `json:"assetID"` - Open bool `json:"open"` - Running bool `json:"running"` - Balance *WalletBalance `json:"balance"` - Address string `json:"address"` - Units string `json:"units"` - Encrypted bool `json:"encrypted"` + Symbol string `json:"symbol"` + AssetID uint32 `json:"assetID"` + Open bool `json:"open"` + Running bool `json:"running"` + Balance *WalletBalance `json:"balance"` + Address string `json:"address"` + Units string `json:"units"` + Encrypted bool `json:"encrypted"` + Synced bool `json:"synced"` + SyncProgress float32 `json:"syncProgress"` } // User is information about the user's wallets and DEX accounts. diff --git a/client/core/wallet.go b/client/core/wallet.go index 460e35ed17..0a37463d63 100644 --- a/client/core/wallet.go +++ b/client/core/wallet.go @@ -17,15 +17,17 @@ import ( // xcWallet is a wallet. type xcWallet struct { asset.Wallet - connector *dex.ConnectionMaster - AssetID uint32 - mtx sync.RWMutex - lockTime time.Time - hookedUp bool - balance *WalletBalance - encPW []byte - address string - dbID []byte + connector *dex.ConnectionMaster + AssetID uint32 + mtx sync.RWMutex + lockTime time.Time + hookedUp bool + balance *WalletBalance + encPW []byte + address string + dbID []byte + synced bool + syncProgress float32 } // Unlock unlocks the wallet. @@ -74,14 +76,16 @@ func (w *xcWallet) state() *WalletState { defer w.mtx.RUnlock() winfo := w.Info() return &WalletState{ - Symbol: unbip(w.AssetID), - AssetID: w.AssetID, - Open: w.unlocked(), - Running: w.connector.On(), - Balance: w.balance, - Address: w.address, - Units: winfo.Units, - Encrypted: len(w.encPW) > 0, + Symbol: unbip(w.AssetID), + AssetID: w.AssetID, + Open: w.unlocked(), + Running: w.connector.On(), + Balance: w.balance, + Address: w.address, + Units: winfo.Units, + Encrypted: len(w.encPW) > 0, + Synced: w.synced, + SyncProgress: w.syncProgress, } } @@ -113,8 +117,14 @@ func (w *xcWallet) Connect(ctx context.Context) error { if err != nil { return err } + synced, progress, err := w.SyncStatus() + if err != nil { + return err + } w.mtx.Lock() w.hookedUp = true + w.synced = synced + w.syncProgress = progress w.mtx.Unlock() return nil } diff --git a/client/webserver/live_test.go b/client/webserver/live_test.go index 9343d8e63b..eecec10060 100644 --- a/client/webserver/live_test.go +++ b/client/webserver/live_test.go @@ -144,14 +144,16 @@ func mkSupportedAsset(symbol string, state *tWalletState, bal *core.WalletBalanc var wallet *core.WalletState if state != nil { wallet = &core.WalletState{ - Symbol: unbip(assetID), - AssetID: assetID, - Open: state.open, - Running: state.running, - Address: ordertest.RandomAddress(), - Balance: bal, - Units: winfo.Units, - Encrypted: true, + Symbol: unbip(assetID), + AssetID: assetID, + Open: state.open, + Running: state.running, + Address: ordertest.RandomAddress(), + Balance: bal, + Units: winfo.Units, + Encrypted: true, + Synced: false, + SyncProgress: 0.5, } } return &core.SupportedAsset{ diff --git a/client/webserver/site/src/css/icons.scss b/client/webserver/site/src/css/icons.scss index a83fb8d6b4..fc7f8c0120 100644 --- a/client/webserver/site/src/css/icons.scss +++ b/client/webserver/site/src/css/icons.scss @@ -83,3 +83,7 @@ .ico-open::before { content: "\e909"; } + +.ico-sync::before { + content: "\e90a"; +} diff --git a/client/webserver/site/src/font/icomoon.svg b/client/webserver/site/src/font/icomoon.svg index 4dd148ab11..deede388a0 100644 --- a/client/webserver/site/src/font/icomoon.svg +++ b/client/webserver/site/src/font/icomoon.svg @@ -24,4 +24,5 @@ + diff --git a/client/webserver/site/src/font/icomoon.ttf b/client/webserver/site/src/font/icomoon.ttf index 81e538d425..482e60cc99 100644 Binary files a/client/webserver/site/src/font/icomoon.ttf and b/client/webserver/site/src/font/icomoon.ttf differ diff --git a/client/webserver/site/src/font/icomoon.woff b/client/webserver/site/src/font/icomoon.woff index 523ce29d5b..2156407398 100644 Binary files a/client/webserver/site/src/font/icomoon.woff and b/client/webserver/site/src/font/icomoon.woff differ diff --git a/client/webserver/site/src/html/markets.tmpl b/client/webserver/site/src/html/markets.tmpl index c83e562a06..07e162a3c8 100644 --- a/client/webserver/site/src/html/markets.tmpl +++ b/client/webserver/site/src/html/markets.tmpl @@ -2,6 +2,7 @@ + {{/* not showing ico-cross */}} {{end}} diff --git a/client/webserver/site/src/html/wallets.tmpl b/client/webserver/site/src/html/wallets.tmpl index 22c75bfd7e..6f087d3af3 100644 --- a/client/webserver/site/src/html/wallets.tmpl +++ b/client/webserver/site/src/html/wallets.tmpl @@ -6,12 +6,16 @@ + {{walletStatusString $w}} {{else}} + no wallet {{end}} {{end}} diff --git a/client/webserver/site/src/js/doc.js b/client/webserver/site/src/js/doc.js index a8de94b878..139e56d048 100644 --- a/client/webserver/site/src/js/doc.js +++ b/client/webserver/site/src/js/doc.js @@ -232,19 +232,20 @@ var Easing = { /* WalletIcons are used for controlling wallets in various places. */ export class WalletIcons { constructor (box) { - const stateElement = (row, name) => row.querySelector(`[data-state=${name}]`) + const stateElement = (name) => box.querySelector(`[data-state=${name}]`) this.icons = {} - this.icons.sleeping = stateElement(box, 'sleeping') - this.icons.locked = stateElement(box, 'locked') - this.icons.unlocked = stateElement(box, 'unlocked') - this.icons.nowallet = stateElement(box, 'nowallet') - this.status = stateElement(box, 'status') + this.icons.sleeping = stateElement('sleeping') + this.icons.locked = stateElement('locked') + this.icons.unlocked = stateElement('unlocked') + this.icons.nowallet = stateElement('nowallet') + this.icons.syncing = stateElement('syncing') + this.status = stateElement('status') } /* sleeping sets the icons to indicate that the wallet is not connected. */ sleeping () { const i = this.icons - Doc.hide(i.locked, i.unlocked, i.nowallet) + Doc.hide(i.locked, i.unlocked, i.nowallet, i.syncing) Doc.show(i.sleeping) if (this.status) this.status.textContent = 'off' } @@ -273,13 +274,26 @@ export class WalletIcons { /* sleeping sets the icons to indicate that no wallet exists. */ nowallet () { const i = this.icons - Doc.hide(i.locked, i.unlocked, i.sleeping) + Doc.hide(i.locked, i.unlocked, i.sleeping, i.syncing) Doc.show(i.nowallet) if (this.status) this.status.textContent = 'no wallet' } + setSyncing (wallet) { + const icon = this.icons.syncing + if (!wallet || !wallet.running) { + Doc.hide(icon) + return + } + if (!wallet.synced) { + Doc.show(icon) + icon.dataset.tooltip = `wallet is ${(wallet.syncProgress * 100).toFixed(1)}% synced` + } else Doc.hide(icon) + } + /* reads the core.Wallet state and sets the icon visibility. */ readWallet (wallet) { + this.setSyncing(wallet) switch (true) { case (!wallet): this.nowallet() diff --git a/client/webserver/template.go b/client/webserver/template.go index 182a7385c9..29256e1469 100644 --- a/client/webserver/template.go +++ b/client/webserver/template.go @@ -155,4 +155,7 @@ var templateFuncs = template.FuncMap{ return "off" } }, + "x100": func(v float32) float32 { + return v * 100 + }, } diff --git a/server/asset/btc/btc.go b/server/asset/btc/btc.go index 12fee7a2a0..c48c20cdf7 100644 --- a/server/asset/btc/btc.go +++ b/server/asset/btc/btc.go @@ -9,6 +9,7 @@ import ( "crypto/sha256" "encoding/binary" "encoding/hex" + "encoding/json" "errors" "fmt" "math" @@ -26,6 +27,8 @@ import ( "github.com/btcsuite/btcutil" ) +const methodGetBlockchainInfo = "getblockchaininfo" + // Driver implements asset.Driver. type Driver struct{} @@ -70,6 +73,7 @@ type btcNode interface { GetBlockVerbose(blockHash *chainhash.Hash) (*btcjson.GetBlockVerboseResult, error) GetBlockHash(blockHeight int64) (*chainhash.Hash, error) GetBestBlockHash() (*chainhash.Hash, error) + RawRequest(method string, params []json.RawMessage) (json.RawMessage, error) } // Backend is a dex backend for Bitcoin or a Bitcoin clone. It has methods for @@ -205,6 +209,15 @@ func (btc *Backend) ValidateSecret(secret, contract []byte) bool { return bytes.Equal(h[:], secretHash) } +// Synced is true if the blockchain is ready for action. +func (btc *Backend) Synced() (bool, error) { + chainInfo, err := btc.getBlockchainInfo() + if err != nil { + return false, fmt.Errorf("GetBlockChainInfo error: %w", err) + } + return !chainInfo.InitialBlockDownload && chainInfo.Headers-chainInfo.Blocks <= 1, nil +} + // Redemption is an input that redeems a swap contract. func (btc *Backend) Redemption(redemptionID, contractID []byte) (asset.Coin, error) { txHash, vin, err := decodeCoinID(redemptionID) @@ -363,6 +376,51 @@ func (btc *Backend) blockInfo(verboseTx *btcjson.TxRawResult) (blockHeight uint3 return } +// anylist is a list of RPC parameters to be converted to []json.RawMessage and +// sent via RawRequest. +type anylist []interface{} + +// call is used internally to marshal parmeters and send requests to the RPC +// server via (*rpcclient.Client).RawRequest. If `thing` is non-nil, the result +// will be marshaled into `thing`. +func (btc *Backend) call(method string, args anylist, thing interface{}) error { + params := make([]json.RawMessage, 0, len(args)) + for i := range args { + p, err := json.Marshal(args[i]) + if err != nil { + return err + } + params = append(params, p) + } + b, err := btc.node.RawRequest(method, params) + if err != nil { + return fmt.Errorf("rawrequest error: %v", err) + } + if thing != nil { + return json.Unmarshal(b, thing) + } + return nil +} + +// getBlockchainInfoResult models the data returned from the getblockchaininfo +// command. +type getBlockchainInfoResult struct { + Blocks int64 `json:"blocks"` + Headers int64 `json:"headers"` + BestBlockHash string `json:"bestblockhash"` + InitialBlockDownload bool `json:"initialblockdownload"` +} + +// getBlockchainInfo sends the getblockchaininfo request and returns the result. +func (btc *Backend) getBlockchainInfo() (*getBlockchainInfoResult, error) { + chainInfo := new(getBlockchainInfoResult) + err := btc.call(methodGetBlockchainInfo, nil, chainInfo) + if err != nil { + return nil, err + } + return chainInfo, nil +} + // Get the UTXO data and perform some checks for script support. func (btc *Backend) utxo(txHash *chainhash.Hash, vout uint32, redeemScript []byte) (*UTXO, error) { txOut, verboseTx, pkScript, err := btc.getTxOutInfo(txHash, vout) @@ -738,6 +796,11 @@ func (btc *Backend) auditContract(contract *Contract) error { func (btc *Backend) Run(ctx context.Context) { defer btc.shutdown() + _, err := btc.FeeRate() + if err != nil { + btc.log.Warnf("%s backend started without fee estimation available: %v", btc.name, err) + } + blockPoll := time.NewTicker(blockPollInterval) defer blockPoll.Stop() addBlock := func(block *btcjson.GetBlockVerboseResult, reorg bool) { diff --git a/server/asset/btc/btc_test.go b/server/asset/btc/btc_test.go index 06c5de4820..da95b9c1c7 100644 --- a/server/asset/btc/btc_test.go +++ b/server/asset/btc/btc_test.go @@ -10,6 +10,7 @@ import ( "context" "crypto/sha256" "encoding/hex" + "encoding/json" "errors" "fmt" "io/ioutil" @@ -292,7 +293,10 @@ func cleanTestChain() { } // A stub to replace rpcclient.Client for offline testing. -type testNode struct{} +type testNode struct { + rawResult []byte + rawErr error +} // Encode utxo info as a concatenated string hash:vout. func txOutID(txHash *chainhash.Hash, index uint32) string { @@ -301,7 +305,7 @@ func txOutID(txHash *chainhash.Hash, index uint32) string { const optimalFeeRate uint64 = 24 -func (testNode) EstimateSmartFee(confTarget int64, mode *btcjson.EstimateSmartFeeMode) (*btcjson.EstimateSmartFeeResult, error) { +func (*testNode) EstimateSmartFee(confTarget int64, mode *btcjson.EstimateSmartFeeMode) (*btcjson.EstimateSmartFeeResult, error) { optimalRate := float64(optimalFeeRate) * 1e-5 // fmt.Println((float64(optimalFeeRate)*1e-5)-0.00024) return &btcjson.EstimateSmartFeeResult{ @@ -311,7 +315,7 @@ func (testNode) EstimateSmartFee(confTarget int64, mode *btcjson.EstimateSmartFe } // Part of the btcNode interface. -func (t testNode) GetTxOut(txHash *chainhash.Hash, index uint32, _ bool) (*btcjson.GetTxOutResult, error) { +func (t *testNode) GetTxOut(txHash *chainhash.Hash, index uint32, _ bool) (*btcjson.GetTxOutResult, error) { testChainMtx.RLock() defer testChainMtx.RUnlock() outID := txOutID(txHash, index) @@ -321,7 +325,7 @@ func (t testNode) GetTxOut(txHash *chainhash.Hash, index uint32, _ bool) (*btcjs } // Part of the btcNode interface. -func (t testNode) GetRawTransactionVerbose(txHash *chainhash.Hash) (*btcjson.TxRawResult, error) { +func (t *testNode) GetRawTransactionVerbose(txHash *chainhash.Hash) (*btcjson.TxRawResult, error) { testChainMtx.RLock() defer testChainMtx.RUnlock() tx, found := testChain.txRaws[*txHash] @@ -332,7 +336,7 @@ func (t testNode) GetRawTransactionVerbose(txHash *chainhash.Hash) (*btcjson.TxR } // Part of the btcNode interface. -func (t testNode) GetBlockVerbose(blockHash *chainhash.Hash) (*btcjson.GetBlockVerboseResult, error) { +func (t *testNode) GetBlockVerbose(blockHash *chainhash.Hash) (*btcjson.GetBlockVerboseResult, error) { testChainMtx.RLock() defer testChainMtx.RUnlock() block, found := testChain.blocks[*blockHash] @@ -343,7 +347,7 @@ func (t testNode) GetBlockVerbose(blockHash *chainhash.Hash) (*btcjson.GetBlockV } // Part of the btcNode interface. -func (t testNode) GetBlockHash(blockHeight int64) (*chainhash.Hash, error) { +func (t *testNode) GetBlockHash(blockHeight int64) (*chainhash.Hash, error) { testChainMtx.RLock() defer testChainMtx.RUnlock() hash, found := testChain.hashes[blockHeight] @@ -354,13 +358,20 @@ func (t testNode) GetBlockHash(blockHeight int64) (*chainhash.Hash, error) { } // Part of the btcNode interface. -func (t testNode) GetBestBlockHash() (*chainhash.Hash, error) { +func (t *testNode) GetBestBlockHash() (*chainhash.Hash, error) { testChainMtx.RLock() defer testChainMtx.RUnlock() bbHash := testBestBlock.hash return &bbHash, nil } +func (t *testNode) RawRequest(string, []json.RawMessage) (json.RawMessage, error) { + if t.rawErr != nil { + return nil, t.rawErr + } + return t.rawResult, nil +} + // Create a btcjson.GetTxOutResult such as is returned from GetTxOut. func testGetTxOut(confirmations, value int64, pkScript []byte) *btcjson.GetTxOutResult { return &btcjson.GetTxOutResult{ @@ -728,7 +739,7 @@ func testMsgTxP2SHMofN(m, n int, segwit bool) *testMsgTxP2SH { // Make a backend that logs to stdout. func testBackend(segwit bool) (*Backend, func()) { logger := dex.StdOutLogger("TEST", dex.LevelTrace) - btc := newBTC("btc", segwit, testParams, logger, testNode{}) + btc := newBTC("btc", segwit, testParams, logger, &testNode{}) ctx, cancel := context.WithCancel(context.Background()) var wg sync.WaitGroup shutdown := func() { @@ -1429,3 +1440,39 @@ func TestDriver_DecodeCoinID(t *testing.T) { }) } } + +func TestSynced(t *testing.T) { + btc, shutdown := testBackend(true) + defer shutdown() + tNode := btc.node.(*testNode) + tNode.rawResult, _ = json.Marshal(&btcjson.GetBlockChainInfoResult{ + Headers: 100, + Blocks: 99, + }) + synced, err := btc.Synced() + if err != nil { + t.Fatalf("Synced error: %v", err) + } + if !synced { + t.Fatalf("not synced when should be synced") + } + + tNode.rawResult, _ = json.Marshal(&btcjson.GetBlockChainInfoResult{ + Headers: 100, + Blocks: 50, + }) + synced, err = btc.Synced() + if err != nil { + t.Fatalf("Synced error: %v", err) + } + if synced { + t.Fatalf("synced when shouldn't be synced") + } + + tNode.rawErr = fmt.Errorf("test error") + _, err = btc.Synced() + if err == nil { + t.Fatalf("getblockchaininfo error not propagated") + } + tNode.rawErr = nil +} diff --git a/server/asset/common.go b/server/asset/common.go index eb0e407318..7b14dccbd6 100644 --- a/server/asset/common.go +++ b/server/asset/common.go @@ -55,6 +55,9 @@ type Backend interface { VerifyUnspentCoin(coinID []byte) error // FeeRate returns the current optimal fee rate in atoms / byte. FeeRate() (uint64, error) + // Synced should return true when the blockchain is synced and ready for + // fee rate estimation. + Synced() (bool, error) } // Coin represents a transaction input or output. diff --git a/server/asset/dcr/dcr.go b/server/asset/dcr/dcr.go index e1658e618e..ed63a02220 100644 --- a/server/asset/dcr/dcr.go +++ b/server/asset/dcr/dcr.go @@ -75,6 +75,7 @@ type dcrNode interface { GetBlockVerbose(blockHash *chainhash.Hash, verboseTx bool) (*chainjson.GetBlockVerboseResult, error) GetBlockHash(blockHeight int64) (*chainhash.Hash, error) GetBestBlockHash() (*chainhash.Hash, error) + GetBlockChainInfo() (*chainjson.GetBlockChainInfoResult, error) } // Backend is an asset backend for Decred. It has methods for fetching output @@ -251,6 +252,15 @@ func (dcr *Backend) ValidateSecret(secret, contract []byte) bool { return bytes.Equal(h[:], secretHash) } +// Synced is true if the blockchain is ready for action. +func (dcr *Backend) Synced() (bool, error) { + chainInfo, err := dcr.node.GetBlockChainInfo() + if err != nil { + return false, fmt.Errorf("GetBlockChainInfo error: %w", err) + } + return !chainInfo.InitialBlockDownload && chainInfo.Headers-chainInfo.Blocks <= 1, nil +} + // Redemption is an input that redeems a swap contract. func (dcr *Backend) Redemption(redemptionID, contractID []byte) (asset.Coin, error) { txHash, vin, err := decodeCoinID(redemptionID) @@ -513,6 +523,12 @@ func (dcr *Backend) Run(ctx context.Context) { dcr.shutdown() wg.Done() }() + + _, err := dcr.FeeRate() + if err != nil { + dcr.log.Warnf("Decred backend started without fee estimation available: %v", err) + } + blockPoll := time.NewTicker(blockPollInterval) defer blockPoll.Stop() addBlock := func(block *chainjson.GetBlockVerboseResult, reorg bool) { diff --git a/server/asset/dcr/dcr_test.go b/server/asset/dcr/dcr_test.go index b3669a71cc..6122117944 100644 --- a/server/asset/dcr/dcr_test.go +++ b/server/asset/dcr/dcr_test.go @@ -176,7 +176,10 @@ func cleanTestChain() { } // A stub to replace rpcclient.Client for offline testing. -type testNode struct{} +type testNode struct { + blockchainInfo *chainjson.GetBlockChainInfoResult + blockchainInfoErr error +} // Store utxo info as a concatenated string hash:vout. func txOutID(txHash *chainhash.Hash, index uint32) string { @@ -185,14 +188,14 @@ func txOutID(txHash *chainhash.Hash, index uint32) string { const optimalFeeRate uint64 = 22 -func (testNode) EstimateSmartFee(confirmations int64, mode chainjson.EstimateSmartFeeMode) (float64, error) { +func (*testNode) EstimateSmartFee(confirmations int64, mode chainjson.EstimateSmartFeeMode) (float64, error) { optimalRate := float64(optimalFeeRate) * 1e-5 // fmt.Println((float64(optimalFeeRate)*1e-5)-0.00022) return optimalRate, nil // optimalFeeRate: 22 atoms/byte = 0.00022 DCR/KB * 1e8 atoms/DCR * 1e-3 KB/Byte } // Part of the dcrNode interface. -func (testNode) GetTxOut(txHash *chainhash.Hash, index uint32, _ bool) (*chainjson.GetTxOutResult, error) { +func (*testNode) GetTxOut(txHash *chainhash.Hash, index uint32, _ bool) (*chainjson.GetTxOutResult, error) { outID := txOutID(txHash, index) testChainMtx.RLock() defer testChainMtx.RUnlock() @@ -202,7 +205,7 @@ func (testNode) GetTxOut(txHash *chainhash.Hash, index uint32, _ bool) (*chainjs } // Part of the dcrNode interface. -func (testNode) GetRawTransactionVerbose(txHash *chainhash.Hash) (*chainjson.TxRawResult, error) { +func (*testNode) GetRawTransactionVerbose(txHash *chainhash.Hash) (*chainjson.TxRawResult, error) { testChainMtx.RLock() defer testChainMtx.RUnlock() tx, found := testChain.txRaws[*txHash] @@ -213,7 +216,7 @@ func (testNode) GetRawTransactionVerbose(txHash *chainhash.Hash) (*chainjson.TxR } // Part of the dcrNode interface. -func (testNode) GetBlockVerbose(blockHash *chainhash.Hash, verboseTx bool) (*chainjson.GetBlockVerboseResult, error) { +func (*testNode) GetBlockVerbose(blockHash *chainhash.Hash, verboseTx bool) (*chainjson.GetBlockVerboseResult, error) { testChainMtx.RLock() defer testChainMtx.RUnlock() block, found := testChain.blocks[*blockHash] @@ -224,7 +227,7 @@ func (testNode) GetBlockVerbose(blockHash *chainhash.Hash, verboseTx bool) (*cha } // Part of the dcrNode interface. -func (testNode) GetBlockHash(blockHeight int64) (*chainhash.Hash, error) { +func (*testNode) GetBlockHash(blockHeight int64) (*chainhash.Hash, error) { testChainMtx.RLock() defer testChainMtx.RUnlock() hash, found := testChain.hashes[blockHeight] @@ -235,7 +238,7 @@ func (testNode) GetBlockHash(blockHeight int64) (*chainhash.Hash, error) { } // Part of the dcrNode interface. -func (testNode) GetBestBlockHash() (*chainhash.Hash, error) { +func (*testNode) GetBestBlockHash() (*chainhash.Hash, error) { testChainMtx.RLock() defer testChainMtx.RUnlock() if len(testChain.hashes) == 0 { @@ -250,6 +253,13 @@ func (testNode) GetBestBlockHash() (*chainhash.Hash, error) { return testChain.hashes[bestHeight], nil } +func (t *testNode) GetBlockChainInfo() (*chainjson.GetBlockChainInfoResult, error) { + if t.blockchainInfoErr != nil { + return nil, t.blockchainInfoErr + } + return t.blockchainInfo, nil +} + // Create a chainjson.GetTxOutResult such as is returned from GetTxOut. func testGetTxOut(confirmations int64, pkScript []byte) *chainjson.GetTxOutResult { return &chainjson.GetTxOutResult{ @@ -686,7 +696,7 @@ func testMsgTxRevocation() *testMsgTx { // Make a backend that logs to stdout. func testBackend() (*Backend, func()) { dcr := unconnectedDCR(tLogger) - dcr.node = testNode{} + dcr.node = &testNode{} ctx, cancel := context.WithCancel(context.Background()) var wg sync.WaitGroup @@ -1569,3 +1579,39 @@ func TestDriver_DecodeCoinID(t *testing.T) { }) } } + +func TestSynced(t *testing.T) { + dcr, shutdown := testBackend() + defer shutdown() + tNode := dcr.node.(*testNode) + tNode.blockchainInfo = &chainjson.GetBlockChainInfoResult{ + Headers: 100, + Blocks: 99, + } + synced, err := dcr.Synced() + if err != nil { + t.Fatalf("Synced error: %v", err) + } + if !synced { + t.Fatalf("not synced when should be synced") + } + + tNode.blockchainInfo = &chainjson.GetBlockChainInfoResult{ + Headers: 100, + Blocks: 50, + } + synced, err = dcr.Synced() + if err != nil { + t.Fatalf("Synced error: %v", err) + } + if synced { + t.Fatalf("synced when shouldn't be synced") + } + + tNode.blockchainInfoErr = fmt.Errorf("test error") + _, err = dcr.Synced() + if err == nil { + t.Fatalf("getblockchaininfo error not propagated") + } + tNode.blockchainInfoErr = nil +} diff --git a/server/market/market.go b/server/market/market.go index f362a1c21e..d86f2d3143 100644 --- a/server/market/market.go +++ b/server/market/market.go @@ -46,7 +46,9 @@ const ( ErrQuantityTooHigh = Error("order quantity exceeds user limit") ErrDuplicateCancelOrder = Error("equivalent cancel order already in epoch") ErrTooManyCancelOrders = Error("too many cancel orders in current epoch") - ErrInvalidCancelOrder = Error("cancel order account does not match targeted order account") + ErrCancelNotPermitted = Error("cancel order account does not match targeted order account") + ErrTargetNotActive = Error("target order not active on this market") + ErrTargetNotCancelable = Error("targeted order is not a limit order with standing time-in-force") ErrSuspendedAccount = Error("suspended account") ErrMalformedOrderResponse = Error("malformed order response") ErrInternalServer = Error("internal server error") @@ -57,6 +59,7 @@ type Swapper interface { Negotiate(matchSets []*order.MatchSet, offBook map[order.OrderID]bool) CheckUnspent(asset uint32, coinID []byte) error UserSwappingAmt(user account.AccountID, base, quote uint32) (amt, count uint64) + ChainsSynced(base, quote uint32) (bool, error) } // Market is the market manager. It should not be overly involved with details @@ -599,10 +602,13 @@ func (m *Market) Cancelable(oid order.OrderID) bool { // means: (1) an order in the book or epoch queue, (2) type limit with // time-in-force standing (implied for book orders), and (3) AccountID field // matching the provided account ID. -func (m *Market) CancelableBy(oid order.OrderID, aid account.AccountID) bool { +func (m *Market) CancelableBy(oid order.OrderID, aid account.AccountID) (bool, error) { // All book orders are standing limit orders. if lo := m.book.Order(oid); lo != nil { - return lo.AccountID == aid + if lo.AccountID == aid { + return true, nil + } + return false, ErrCancelNotPermitted } // Check the active epochs (includes current and next). @@ -610,10 +616,21 @@ func (m *Market) CancelableBy(oid order.OrderID, aid account.AccountID) bool { ord := m.epochOrders[oid] m.epochMtx.RUnlock() - if lo, ok := ord.(*order.LimitOrder); ok { - return lo.Force == order.StandingTiF && lo.AccountID == aid + if ord == nil { + return false, ErrTargetNotActive } - return false + + lo, ok := ord.(*order.LimitOrder) + if !ok { + return false, ErrTargetNotCancelable + } + if lo.Force != order.StandingTiF { + return false, ErrTargetNotCancelable + } + if lo.AccountID != aid { + return false, ErrCancelNotPermitted + } + return true, nil } func (m *Market) checkUnfilledOrders(assetID uint32, unfilled []*order.LimitOrder) (unbooked []*order.LimitOrder) { @@ -893,19 +910,27 @@ func (m *Market) Run(ctx context.Context) { m.activeEpochIdx = currentEpoch.Epoch if !running { - // Open up SubmitOrderAsync. - close(m.running) - running = true - log.Infof("Market %s now accepting orders, epoch %d:%d", m.marketInfo.Name, - currentEpoch.Epoch, epochDuration) - // Signal to the book router if this is a resume. - if m.suspendEpochIdx != 0 { - notifyChan <- &updateSignal{ - action: resumeAction, - data: sigDataResume{ - epochIdx: currentEpoch.Epoch, - // TODO: signal config or new config - }, + // Check that both blockchains are synced before actually starting. + synced, err := m.swapper.ChainsSynced(m.marketInfo.Base, m.marketInfo.Quote) + if err != nil { + log.Errorf("Not starting %s market because of ChainsSynced error: %v", m.marketInfo.Name, err) + } else if !synced { + log.Debugf("Delaying start of %s market because chains aren't synced", m.marketInfo.Name) + } else { + // Open up SubmitOrderAsync. + close(m.running) + running = true + log.Infof("Market %s now accepting orders, epoch %d:%d", m.marketInfo.Name, + currentEpoch.Epoch, epochDuration) + // Signal to the book router if this is a resume. + if m.suspendEpochIdx != 0 { + notifyChan <- &updateSignal{ + action: resumeAction, + data: sigDataResume{ + epochIdx: currentEpoch.Epoch, + // TODO: signal config or new config + }, + } } } } @@ -1177,10 +1202,11 @@ func (m *Market) processOrder(rec *orderRecord, epoch *EpochQueue, notifyChan ch // Verify that the target order is on the books or in the epoch queue, // and that the account of the CancelOrder is the same as the account of // the target order. - if !m.CancelableBy(co.TargetOrderID, co.AccountID) { - log.Debugf("Cancel order %v (account=%v) does not own target order %v.", - co, co.AccountID, co.TargetOrderID) - errChan <- ErrInvalidCancelOrder + cancelable, err := m.CancelableBy(co.TargetOrderID, co.AccountID) + if !cancelable { + log.Debugf("Cancel order %v (account=%v) target order %v: %v", + co, co.AccountID, co.TargetOrderID, err) + errChan <- err return nil } } else if likelyTaker(ord) { // Likely-taker trade order. Check the quantity against user's limit. diff --git a/server/market/market_test.go b/server/market/market_test.go index 6fbe4b206f..1dfb3683d0 100644 --- a/server/market/market_test.go +++ b/server/market/market_test.go @@ -16,6 +16,7 @@ import ( "runtime" "strings" "sync" + "sync/atomic" "testing" "time" @@ -738,12 +739,23 @@ func TestMarket_Run(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() - startEpochIdx := 2 + encode.UnixMilli(time.Now())/epochDurationMSec + + // Check that start is delayed by an unsynced backend. Tell the Market to + // start + atomic.StoreUint32(&oRig.dcr.synced, 0) + nowEpochIdx := encode.UnixMilli(time.Now())/epochDurationMSec + 1 + + unsyncedEpochIdx := nowEpochIdx + 1 + unsyncedEpochTime := encode.UnixTimeMilli(unsyncedEpochIdx * epochDurationMSec) + + startEpochIdx := unsyncedEpochIdx + 1 + startEpochTime := encode.UnixTimeMilli(startEpochIdx * epochDurationMSec) + var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() - mkt.Start(ctx, startEpochIdx) + mkt.Start(ctx, unsyncedEpochIdx) }() // Make an order for the first epoch. @@ -828,11 +840,21 @@ func TestMarket_Run(t *testing.T) { t.Fatalf("Market should not be running yet") } - mkt.waitForEpochOpen() + halfEpoch := time.Duration(epochDurationMSec/2) * time.Millisecond - mktStatus = mkt.Status() - if !mktStatus.Running { - t.Fatalf("Market should be running now") + <-time.After(time.Until(unsyncedEpochTime.Add(halfEpoch))) + + if mkt.Running() { + t.Errorf("market running with an unsynced backend") + } + + atomic.StoreUint32(&oRig.dcr.synced, 1) + + <-time.After(time.Until(startEpochTime.Add(halfEpoch))) + <-storage.epochInserted + + if !mkt.Running() { + t.Errorf("market not running after backend sync finished") } // Submit again @@ -912,8 +934,8 @@ func TestMarket_Run(t *testing.T) { err = mkt.SubmitOrder(&coRecordWrongAccount) if err == nil { t.Errorf("An invalid order was processed, but it should not have been.") - } else if !errors.Is(err, ErrInvalidCancelOrder) { - t.Errorf(`expected ErrInvalidCancelOrder ("%v"), got "%v"`, ErrInvalidCancelOrder, err) + } else if !errors.Is(err, ErrCancelNotPermitted) { + t.Errorf(`expected ErrCancelNotPermitted ("%v"), got "%v"`, ErrCancelNotPermitted, err) } // Valid cancel order @@ -968,9 +990,7 @@ func TestMarket_Run(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - mkt.Run(ctx) // begin on next epoch start - // /startEpochIdx = 1 + encode.UnixMilli(time.Now())/epochDurationMSec - //mkt.Start(ctx, startEpochIdx) + mkt.Run(ctx) }() mkt.waitForEpochOpen() diff --git a/server/market/routers_test.go b/server/market/routers_test.go index 012ff6c3d3..7fc431fb54 100644 --- a/server/market/routers_test.go +++ b/server/market/routers_test.go @@ -324,12 +324,15 @@ type TBackend struct { utxoErr error utxos map[string]uint64 addrChecks bool + synced uint32 + syncedErr error } func tNewBackend() *TBackend { return &TBackend{ utxos: make(map[string]uint64), addrChecks: true, + synced: 1, } } @@ -374,6 +377,13 @@ func (b *TBackend) FeeRate() (uint64, error) { return 9, nil } +func (b *TBackend) Synced() (bool, error) { + if b.syncedErr != nil { + return false, b.syncedErr + } + return atomic.LoadUint32(&oRig.dcr.synced) == 1, nil +} + type tUTXO struct { val uint64 decoded string diff --git a/server/swap/swap.go b/server/swap/swap.go index 1c54492f5a..8c3e953ac9 100644 --- a/server/swap/swap.go +++ b/server/swap/swap.go @@ -530,6 +530,30 @@ func (s *Swapper) UserSwappingAmt(user account.AccountID, base, quote uint32) (a return } +// ChainsSynced will return true if both specified asset's backends are synced. +func (s *Swapper) ChainsSynced(base, quote uint32) (bool, error) { + b, found := s.coins[base] + if !found { + return false, fmt.Errorf("No backend found for %d", base) + } + baseSynced, err := b.Backend.Synced() + if err != nil { + return false, fmt.Errorf("Error checking sync status for %d", base) + } + if !baseSynced { + return false, nil + } + q, found := s.coins[quote] + if !found { + return false, fmt.Errorf("No backend found for %d", base) + } + quoteSynced, err := q.Backend.Synced() + if err != nil { + return false, fmt.Errorf("Error checking sync status for %d", base) + } + return quoteSynced, nil +} + func (s *Swapper) restoreState(state *State, allowPartial bool) error { // State binary version check should be done when State is loaded. diff --git a/server/swap/swap_test.go b/server/swap/swap_test.go index e97464fc62..eb2fa3e824 100644 --- a/server/swap/swap_test.go +++ b/server/swap/swap_test.go @@ -421,6 +421,7 @@ func (a *TAsset) CheckAddress(string) bool { return true func (a *TAsset) Run(context.Context) {} func (a *TAsset) ValidateSecret(secret, contract []byte) bool { return true } func (a *TAsset) VerifyUnspentCoin(coinID []byte) error { return nil } +func (a *TAsset) Synced() (bool, error) { return true, nil } func (a *TAsset) setContractErr(err error) { a.mtx.Lock()