LBRYCommunity/lbcwallet
1
0
Fork 0
Code Issues 5 Pull requests Projects Releases 14 Packages Wiki Activity

Remove account support, fix races on btcd connect.

Browse source 
This commit is the result of several big changes being made to the
wallet.  In particular, the "handshake" (initial sync to the chain
server) was quite racy and required proper synchronization.  To make
fixing this race easier, several other changes were made to the
internal wallet data structures and much of the RPC server ended up
being rewritten.

First, all account support has been removed.  The previous Account
struct has been replaced with a Wallet structure, which includes a
keystore for saving keys, and a txstore for storing relevant
transactions.  This decision has been made since it is the opinion of
myself and other developers that bitcoind accounts are fundamentally
broken (as accounts implemented by bitcoind support both arbitrary
address groupings as well as moving balances between accounts -- these
are fundamentally incompatible features), and since a BIP0032 keystore
is soon planned to be implemented (at which point, "accounts" can
return as HD extended keys).  With the keystore handling the grouping
of related keys, there is no reason have many different Account
structs, and the AccountManager has been removed as well.  All RPC
handlers that take an account option will only work with "" (the
default account) or "*" if the RPC allows specifying all accounts.

Second, much of the RPC server has been cleaned up.  The global
variables for the RPC server and chain server client have been moved
to part of the rpcServer struct, and the handlers for each RPC method
that are looked up change depending on which components have been set.
Passthrough requests are also no longer handled specially, but when
the chain server is set, a handler to perform the passthrough will be
returned if the method is not otherwise a wallet RPC.  The
notification system for websocket clients has also been rewritten so
wallet components can send notifications through channels, rather than
requiring direct access to the RPC server itself, or worse still,
sending directly to a websocket client's send channel.  In the future,
this will enable proper registration of notifications, rather than
unsolicited broadcasts to every connected websocket client (see
issue #84).

Finally, and the main reason why much of this cleanup was necessary,
the races during intial sync with the chain server have been fixed.
Previously, when the 'Handshake' was run, a rescan would occur which
would perform modifications to Account data structures as
notifications were received.  Synchronization was provided with a
single binary semaphore which serialized all access to wallet and
account data.  However, the Handshake itself was not able to run with
this lock (or else notifications would block), and many data races
would occur as both notifications were being handled.  If GOMAXPROCS
was ever increased beyond 1, btcwallet would always immediately crash
due to invalid addresses caused by the data races on startup.  To fix
this, the single lock for all wallet access has been replaced with
mutexes for both the keystore and txstore.  Handling of btcd
notifications and client requests may now occur simultaneously.
GOMAXPROCS has also been set to the number of logical CPUs at the
beginning of main, since with the data races fixed, there's no reason
to prevent the extra parallelism gained by increasing it.

Closes #78.

Closes #101.

Closes #110.
This commit is contained in:
Josh Rickmar 2014-07-08 22:17:38 -05:00
parent a4dab8a713
commit b9fd527d33
27 changed files with 4427 additions and 5313 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

778
account.go
View file

@ -1,778 +0,0 @@
/*
* Copyright (c) 2013, 2014 Conformal Systems LLC <info@conformal.com>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
package main
import (
"bytes"
"encoding/base64"
"fmt"
"path/filepath"
"github.com/conformal/btcjson"
"github.com/conformal/btcutil"
"github.com/conformal/btcwallet/keystore"
"github.com/conformal/btcwallet/txstore"
"github.com/conformal/btcwire"
)
// Account is a structure containing all the components for a
// complete wallet. It contains the Armory-style wallet (to store
// addresses and keys), and tx and utxo stores, and a mutex to prevent
// incorrect multiple access.
type Account struct {
name string
KeyStore *keystore.Store
TxStore *txstore.Store
lockedOutpoints map[btcwire.OutPoint]struct{}
FeeIncrement btcutil.Amount
}
func newAccount(name string, keys *keystore.Store, txs *txstore.Store) *Account {
return &Account{
name: name,
KeyStore: keys,
TxStore: txs,
lockedOutpoints: map[btcwire.OutPoint]struct{}{},
FeeIncrement: defaultFeeIncrement,
}
}
// Lock locks the underlying wallet for an account.
func (a *Account) Lock() error {
switch err := a.KeyStore.Lock(); err {
case nil:
server.NotifyWalletLockStateChange(a.KeyStore.Name(), true)
return nil
case keystore.ErrLocked:
// Do not pass wallet already locked errors to the caller.
return nil
default:
return err
}
}
// Unlock unlocks the underlying wallet for an account.
func (a *Account) Unlock(passphrase []byte) error {
if err := a.KeyStore.Unlock(passphrase); err != nil {
return err
}
server.NotifyWalletLockStateChange(a.KeyStore.Name(), false)
return nil
}
// AddressUsed returns whether there are any recorded transactions spending to
// a given address. Assumming correct TxStore usage, this will return true iff
// there are any transactions with outputs to this address in the blockchain or
// the btcd mempool.
func (a *Account) AddressUsed(addr btcutil.Address) bool {
// This not only can be optimized by recording this data as it is
// read when opening an account, and keeping it up to date each time a
// new received tx arrives, but it probably should in case an address is
// used in a tx (made public) but the tx is eventually removed from the
// store (consider a chain reorg).
pkHash := addr.ScriptAddress()
for _, r := range a.TxStore.Records() {
credits := r.Credits()
for _, c := range credits {
// Errors don't matter here. If addrs is nil, the
// range below does nothing.
_, addrs, _, _ := c.Addresses(activeNet.Params)
for _, a := range addrs {
if bytes.Equal(a.ScriptAddress(), pkHash) {
return true
}
}
}
}
return false
}
// CalculateBalance sums the amounts of all unspent transaction
// outputs to addresses of a wallet and returns the balance as a
// float64.
//
// If confirmations is 0, all UTXOs, even those not present in a
// block (height -1), will be used to get the balance. Otherwise,
// a UTXO must be in a block. If confirmations is 1 or greater,
// the balance will be calculated based on how many how many blocks
// include a UTXO.
func (a *Account) CalculateBalance(confirms int) (btcutil.Amount, error) {
rpcc, err := accessClient()
if err != nil {
return 0, err
}
bs, err := rpcc.BlockStamp()
if err != nil {
return 0, err
}
return a.TxStore.Balance(confirms, bs.Height)
}
// CalculateAddressBalance sums the amounts of all unspent transaction
// outputs to a single address's pubkey hash and returns the balance
// as a float64.
//
// If confirmations is 0, all UTXOs, even those not present in a
// block (height -1), will be used to get the balance. Otherwise,
// a UTXO must be in a block. If confirmations is 1 or greater,
// the balance will be calculated based on how many how many blocks
// include a UTXO.
func (a *Account) CalculateAddressBalance(addr btcutil.Address, confirms int) (btcutil.Amount, error) {
rpcc, err := accessClient()
if err != nil {
return 0, err
}
bs, err := rpcc.BlockStamp()
if err != nil {
return 0, err
}
var bal btcutil.Amount
unspent, err := a.TxStore.UnspentOutputs()
if err != nil {
return 0, err
}
for _, credit := range unspent {
if credit.Confirmed(confirms, bs.Height) {
// We only care about the case where len(addrs) == 1, and err
// will never be non-nil in that case
_, addrs, _, _ := credit.Addresses(activeNet.Params)
if len(addrs) != 1 {
continue
}
if addrs[0].EncodeAddress() == addr.EncodeAddress() {
bal += credit.Amount()
}
}
}
return bal, nil
}
// CurrentAddress gets the most recently requested Bitcoin payment address
// from an account. If the address has already been used (there is at least
// one transaction spending to it in the blockchain or btcd mempool), the next
// chained address is returned.
func (a *Account) CurrentAddress() (btcutil.Address, error) {
addr := a.KeyStore.LastChainedAddress()
// Get next chained address if the last one has already been used.
if a.AddressUsed(addr) {
return a.NewAddress()
}
return addr, nil
}
// ListSinceBlock returns a slice of objects with details about transactions
// since the given block. If the block is -1 then all transactions are included.
// This is intended to be used for listsinceblock RPC replies.
func (a *Account) ListSinceBlock(since, curBlockHeight int32,
minconf int) ([]btcjson.ListTransactionsResult, error) {
txList := []btcjson.ListTransactionsResult{}
for _, txRecord := range a.TxStore.Records() {
// Transaction records must only be considered if they occur
// after the block height since.
if since != -1 && txRecord.BlockHeight <= since {
continue
}
// Transactions that have not met minconf confirmations are to
// be ignored.
if !txRecord.Confirmed(minconf, curBlockHeight) {
continue
}
jsonResults, err := txRecord.ToJSON(a.name, curBlockHeight,
a.KeyStore.Net())
if err != nil {
return nil, err
}
txList = append(txList, jsonResults...)
}
return txList, nil
}
// ListTransactions returns a slice of objects with details about a recorded
// transaction. This is intended to be used for listtransactions RPC
// replies.
func (a *Account) ListTransactions(from, count int) ([]btcjson.ListTransactionsResult, error) {
txList := []btcjson.ListTransactionsResult{}
// Get current block. The block height used for calculating
// the number of tx confirmations.
rpcc, err := accessClient()
if err != nil {
return txList, err
}
bs, err := rpcc.BlockStamp()
if err != nil {
return txList, err
}
records := a.TxStore.Records()
lastLookupIdx := len(records) - count
// Search in reverse order: lookup most recently-added first.
for i := len(records) - 1; i >= from && i >= lastLookupIdx; i-- {
jsonResults, err := records[i].ToJSON(a.name, bs.Height,
a.KeyStore.Net())
if err != nil {
return nil, err
}
txList = append(txList, jsonResults...)
}
return txList, nil
}
// ListAddressTransactions returns a slice of objects with details about
// recorded transactions to or from any address belonging to a set. This is
// intended to be used for listaddresstransactions RPC replies.
func (a *Account) ListAddressTransactions(pkHashes map[string]struct{}) (
[]btcjson.ListTransactionsResult, error) {
txList := []btcjson.ListTransactionsResult{}
// Get current block. The block height used for calculating
// the number of tx confirmations.
rpcc, err := accessClient()
if err != nil {
return txList, err
}
bs, err := rpcc.BlockStamp()
if err != nil {
return txList, err
}
for _, r := range a.TxStore.Records() {
for _, c := range r.Credits() {
// We only care about the case where len(addrs) == 1,
// and err will never be non-nil in that case.
_, addrs, _, _ := c.Addresses(activeNet.Params)
if len(addrs) != 1 {
continue
}
apkh, ok := addrs[0].(*btcutil.AddressPubKeyHash)
if !ok {
continue
}
if _, ok := pkHashes[string(apkh.ScriptAddress())]; !ok {
continue
}
jsonResult, err := c.ToJSON(a.name, bs.Height,
a.KeyStore.Net())
if err != nil {
return nil, err
}
txList = append(txList, jsonResult)
}
}
return txList, nil
}
// ListAllTransactions returns a slice of objects with details about a recorded
// transaction. This is intended to be used for listalltransactions RPC
// replies.
func (a *Account) ListAllTransactions() ([]btcjson.ListTransactionsResult, error) {
txList := []btcjson.ListTransactionsResult{}
// Get current block. The block height used for calculating
// the number of tx confirmations.
rpcc, err := accessClient()
if err != nil {
return txList, err
}
bs, err := rpcc.BlockStamp()
if err != nil {
return txList, err
}
// Search in reverse order: lookup most recently-added first.
records := a.TxStore.Records()
for i := len(records) - 1; i >= 0; i-- {
jsonResults, err := records[i].ToJSON(a.name, bs.Height,
a.KeyStore.Net())
if err != nil {
return nil, err
}
txList = append(txList, jsonResults...)
}
return txList, nil
}
// DumpPrivKeys returns the WIF-encoded private keys for all addresses with
// private keys in a wallet.
func (a *Account) DumpPrivKeys() ([]string, error) {
// Iterate over each active address, appending the private
// key to privkeys.
privkeys := []string{}
for _, info := range a.KeyStore.ActiveAddresses() {
// Only those addresses with keys needed.
pka, ok := info.(keystore.PubKeyAddress)
if !ok {
continue
}
wif, err := pka.ExportPrivKey()
if err != nil {
// It would be nice to zero out the array here. However,
// since strings in go are immutable, and we have no
// control over the caller I don't think we can. :(
return nil, err
}
privkeys = append(privkeys, wif.String())
}
return privkeys, nil
}
// DumpWIFPrivateKey returns the WIF encoded private key for a
// single wallet address.
func (a *Account) DumpWIFPrivateKey(addr btcutil.Address) (string, error) {
// Get private key from wallet if it exists.
address, err := a.KeyStore.Address(addr)
if err != nil {
return "", err
}
pka, ok := address.(keystore.PubKeyAddress)
if !ok {
return "", fmt.Errorf("address %s is not a key type", addr)
}
wif, err := pka.ExportPrivKey()
if err != nil {
return "", err
}
return wif.String(), nil
}
// ImportPrivateKey imports a private key to the account's wallet and
// writes the new wallet to disk.
func (a *Account) ImportPrivateKey(wif *btcutil.WIF, bs *keystore.BlockStamp,
rescan bool) (string, error) {
// Attempt to import private key into wallet.
addr, err := a.KeyStore.ImportPrivateKey(wif, bs)
if err != nil {
return "", err
}
// Immediately write wallet to disk.
AcctMgr.ds.ScheduleWalletWrite(a)
if err := AcctMgr.ds.FlushAccount(a); err != nil {
return "", fmt.Errorf("cannot write account: %v", err)
}
addrStr := addr.EncodeAddress()
// Rescan blockchain for transactions with txout scripts paying to the
// imported address.
if rescan {
addrs := []btcutil.Address{addr}
job := &RescanJob{
Addresses: map[*Account][]btcutil.Address{a: addrs},
OutPoints: nil,
StartHeight: 0,
}
// Submit rescan job and log when the import has completed.
// Do not block on finishing the rescan.
doneChan := AcctMgr.rm.SubmitJob(job)
go func() {
<-doneChan
log.Infof("Finished import for address %s", addrStr)
}()
}
// Associate the imported address with this account.
AcctMgr.MarkAddressForAccount(addr, a)
log.Infof("Imported payment address %s", addrStr)
// Return the payment address string of the imported private key.
return addrStr, nil
}
// ExportToDirectory writes an account to a special export directory. Any
// previous files are overwritten.
func (a *Account) ExportToDirectory(dirBaseName string) error {
dir := filepath.Join(networkDir(activeNet.Params), dirBaseName)
if err := checkCreateDir(dir); err != nil {
return err
}
return AcctMgr.ds.ExportAccount(a, dir)
}
// ExportWatchingWallet returns a new account with a watching wallet
// exported by this a's wallet. Both wallets share the same tx and utxo
// stores, so locking one will lock the other as well. The returned account
// should be exported quickly, either to file or to an rpc caller, and then
// dropped from scope.
func (a *Account) ExportWatchingWallet() (*Account, error) {
ww, err := a.KeyStore.ExportWatchingWallet()
if err != nil {
return nil, err
}
wa := *a
wa.KeyStore = ww
return &wa, nil
}
// exportBase64 exports an account's serialized wallet, tx, and utxo
// stores as base64-encoded values in a map.
func (a *Account) exportBase64() (map[string]string, error) {
buf := bytes.Buffer{}
m := make(map[string]string)
_, err := a.KeyStore.WriteTo(&buf)
if err != nil {
return nil, err
}
m["wallet"] = base64.StdEncoding.EncodeToString(buf.Bytes())
buf.Reset()
if _, err = a.TxStore.WriteTo(&buf); err != nil {
return nil, err
}
m["tx"] = base64.StdEncoding.EncodeToString(buf.Bytes())
buf.Reset()
return m, nil
}
// LockedOutpoint returns whether an outpoint has been marked as locked and
// should not be used as an input for created transactions.
func (a *Account) LockedOutpoint(op btcwire.OutPoint) bool {
_, locked := a.lockedOutpoints[op]
return locked
}
// LockOutpoint marks an outpoint as locked, that is, it should not be used as
// an input for newly created transactions.
func (a *Account) LockOutpoint(op btcwire.OutPoint) {
a.lockedOutpoints[op] = struct{}{}
}
// UnlockOutpoint marks an outpoint as unlocked, that is, it may be used as an
// input for newly created transactions.
func (a *Account) UnlockOutpoint(op btcwire.OutPoint) {
delete(a.lockedOutpoints, op)
}
// ResetLockedOutpoints resets the set of locked outpoints so all may be used
// as inputs for new transactions.
func (a *Account) ResetLockedOutpoints() {
a.lockedOutpoints = map[btcwire.OutPoint]struct{}{}
}
// LockedOutpoints returns a slice of currently locked outpoints. This is
// intended to be used by marshaling the result as a JSON array for
// listlockunspent RPC results.
func (a *Account) LockedOutpoints() []btcjson.TransactionInput {
locked := make([]btcjson.TransactionInput, len(a.lockedOutpoints))
i := 0
for op := range a.lockedOutpoints {
locked[i] = btcjson.TransactionInput{
Txid: op.Hash.String(),
Vout: op.Index,
}
i++
}
return locked
}
// Track requests btcd to send notifications of new transactions for
// each address stored in a wallet.
func (a *Account) Track() {
rpcc, err := accessClient()
if err != nil {
log.Errorf("No chain server client to track addresses.")
return
}
// Request notifications for transactions sending to all wallet
// addresses.
//
// TODO: return as slice? (doesn't have to be ordered, or
// SortedActiveAddresses would be fine.)
addrMap := a.KeyStore.ActiveAddresses()
addrs := make([]btcutil.Address, 0, len(addrMap))
for addr := range addrMap {
addrs = append(addrs, addr)
}
if err := rpcc.NotifyReceived(addrs); err != nil {
log.Error("Unable to request transaction updates for address.")
}
unspent, err := a.TxStore.UnspentOutputs()
if err != nil {
log.Errorf("Unable to access unspent outputs: %v", err)
return
}
ReqSpentUtxoNtfns(unspent)
}
// RescanActiveJob creates a RescanJob for all active addresses in the
// account. This is needed for catching btcwallet up to a long-running
// btcd process, as otherwise it would have missed notifications as
// blocks are attached to the main chain.
func (a *Account) RescanActiveJob() (*RescanJob, error) {
// Determine the block necesary to start the rescan for all active
// addresses.
height := a.KeyStore.SyncHeight()
actives := a.KeyStore.SortedActiveAddresses()
addrs := make([]btcutil.Address, 0, len(actives))
for i := range actives {
addrs = append(addrs, actives[i].Address())
}
unspents, err := a.TxStore.UnspentOutputs()
if err != nil {
return nil, err
}
outpoints := make([]*btcwire.OutPoint, 0, len(unspents))
for _, c := range unspents {
outpoints = append(outpoints, c.OutPoint())
}
job := &RescanJob{
Addresses: map[*Account][]btcutil.Address{a: addrs},
OutPoints: outpoints,
StartHeight: height,
}
return job, nil
}
// ResendUnminedTxs iterates through all transactions that spend from wallet
// credits that are not known to have been mined into a block, and attempts
// to send each to the chain server for relay.
func (a *Account) ResendUnminedTxs() {
rpcc, err := accessClient()
if err != nil {
log.Errorf("No chain server client to resend txs.")
return
}
txs := a.TxStore.UnminedDebitTxs()
for _, tx := range txs {
_, err := rpcc.SendRawTransaction(tx.MsgTx(), false)
if err != nil {
// TODO(jrick): Check error for if this tx is a double spend,
// remove it if so.
log.Warnf("Could not resend transaction %v: %v",
tx.Sha(), err)
continue
}
log.Debugf("Resent unmined transaction %v", tx.Sha())
}
}
// SortedActivePaymentAddresses returns a slice of all active payment
// addresses in an account.
func (a *Account) SortedActivePaymentAddresses() []string {
infos := a.KeyStore.SortedActiveAddresses()
addrs := make([]string, len(infos))
for i, info := range infos {
addrs[i] = info.Address().EncodeAddress()
}
return addrs
}
// ActivePaymentAddresses returns a set of all active pubkey hashes
// in an account.
func (a *Account) ActivePaymentAddresses() map[string]struct{} {
infos := a.KeyStore.ActiveAddresses()
addrs := make(map[string]struct{}, len(infos))
for _, info := range infos {
addrs[info.Address().EncodeAddress()] = struct{}{}
}
return addrs
}
// NewAddress returns a new payment address for an account.
func (a *Account) NewAddress() (btcutil.Address, error) {
// Get current block's height and hash.
rpcc, err := accessClient()
if err != nil {
return nil, err
}
bs, err := rpcc.BlockStamp()
if err != nil {
return nil, err
}
// Get next address from wallet.
addr, err := a.KeyStore.NextChainedAddress(&bs, cfg.KeypoolSize)
if err != nil {
return nil, err
}
// Immediately write updated wallet to disk.
AcctMgr.ds.ScheduleWalletWrite(a)
if err := AcctMgr.ds.FlushAccount(a); err != nil {
return nil, fmt.Errorf("account write failed: %v", err)
}
// Mark this new address as belonging to this account.
AcctMgr.MarkAddressForAccount(addr, a)
// Request updates from btcd for new transactions sent to this address.
if err := rpcc.NotifyReceived([]btcutil.Address{addr}); err != nil {
return nil, err
}
return addr, nil
}
// NewChangeAddress returns a new change address for an account.
func (a *Account) NewChangeAddress() (btcutil.Address, error) {
// Get current block's height and hash.
rpcc, err := accessClient()
if err != nil {
return nil, err
}
bs, err := rpcc.BlockStamp()
if err != nil {
return nil, err
}
// Get next chained change address from wallet.
addr, err := a.KeyStore.ChangeAddress(&bs, cfg.KeypoolSize)
if err != nil {
return nil, err
}
// Immediately write updated wallet to disk.
AcctMgr.ds.ScheduleWalletWrite(a)
if err := AcctMgr.ds.FlushAccount(a); err != nil {
return nil, fmt.Errorf("account write failed: %v", err)
}
// Mark this new address as belonging to this account.
AcctMgr.MarkAddressForAccount(addr, a)
// Request updates from btcd for new transactions sent to this address.
if err := rpcc.NotifyReceived([]btcutil.Address{addr}); err != nil {
return nil, err
}
return addr, nil
}
// RecoverAddresses recovers the next n chained addresses of a wallet.
func (a *Account) RecoverAddresses(n int) error {
// Get info on the last chained address. The rescan starts at the
// earliest block height the last chained address might appear at.
last := a.KeyStore.LastChainedAddress()
lastInfo, err := a.KeyStore.Address(last)
if err != nil {
return err
}
addrs, err := a.KeyStore.ExtendActiveAddresses(n, cfg.KeypoolSize)
if err != nil {
return err
}
// Run a goroutine to rescan blockchain for recovered addresses.
go func() {
rpcc, err := accessClient()
if err != nil {
log.Errorf("Cannot access chain server client to " +
"rescan recovered addresses.")
return
}
err = rpcc.Rescan(lastInfo.FirstBlock(), addrs, nil)
if err != nil {
log.Errorf("Rescanning for recovered addresses "+
"failed: %v", err)
}
}()
return nil
}
// ReqSpentUtxoNtfns sends a message to btcd to request updates for when
// a stored UTXO has been spent.
func ReqSpentUtxoNtfns(credits []txstore.Credit) {
ops := make([]*btcwire.OutPoint, 0, len(credits))
for _, c := range credits {
op := c.OutPoint()
log.Debugf("Requesting spent UTXO notifications for Outpoint "+
"hash %s index %d", op.Hash, op.Index)
ops = append(ops, op)
}
rpcc, err := accessClient()
if err != nil {
log.Errorf("Cannot access chain server client to " +
"request spent output notifications.")
return
}
if err := rpcc.NotifySpent(ops); err != nil {
log.Errorf("Cannot request notifications for spent outputs: %v",
err)
}
}
// TotalReceived iterates through an account's transaction history, returning the
// total amount of bitcoins received for any account address. Amounts received
// through multisig transactions are ignored.
func (a *Account) TotalReceived(confirms int) (btcutil.Amount, error) {
rpcc, err := accessClient()
if err != nil {
return 0, err
}
bs, err := rpcc.BlockStamp()
if err != nil {
return 0, err
}
var amount btcutil.Amount
for _, r := range a.TxStore.Records() {
for _, c := range r.Credits() {
// Ignore change.
if c.Change() {
continue
}
// Tally if the appropiate number of block confirmations have passed.
if c.Confirmed(confirms, bs.Height) {
amount += c.Amount()
}
}
}
return amount, nil
}

1130
acctmgr.go
View file

File diff suppressed because it is too large Load diff

168
btcwallet.go Normal file
View file

@ -0,0 +1,168 @@
/*
* Copyright (c) 2013, 2014 Conformal Systems LLC <info@conformal.com>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
package main
import (
"io/ioutil"
"net"
"net/http"
_ "net/http/pprof"
"os"
"runtime"
"github.com/conformal/btcwallet/chain"
)
var (
cfg *config
shutdownChan = make(chan struct{})
)
func main() {
// Use all processor cores.
runtime.GOMAXPROCS(runtime.NumCPU())
// Work around defer not working after os.Exit.
if err := walletMain(); err != nil {
os.Exit(1)
}
}
// walletMain is a work-around main function that is required since deferred
// functions (such as log flushing) are not called with calls to os.Exit.
// Instead, main runs this function and checks for a non-nil error, at which
// point any defers have already run, and if the error is non-nil, the program
// can be exited with an error exit status.
func walletMain() error {
// Load configuration and parse command line. This function also
// initializes logging and configures it accordingly.
tcfg, _, err := loadConfig()
if err != nil {
return err
}
cfg = tcfg
defer backendLog.Flush()
if cfg.Profile != "" {
go func() {
listenAddr := net.JoinHostPort("", cfg.Profile)
log.Infof("Profile server listening on %s", listenAddr)
profileRedirect := http.RedirectHandler("/debug/pprof",
http.StatusSeeOther)
http.Handle("/", profileRedirect)
log.Errorf("%v", http.ListenAndServe(listenAddr, nil))
}()
}
// Create and start HTTP server to serve wallet client connections.
// This will be updated with the wallet and chain server RPC client
// created below after each is created.
server, err := newRPCServer(cfg.SvrListeners, cfg.RPCMaxClients,
cfg.RPCMaxWebsockets)
if err != nil {
log.Errorf("Unable to create HTTP server: %v", err)
return err
}
server.Start()
// Shutdown the server if an interrupt signal is received.
addInterruptHandler(server.Stop)
// Create channel so that the goroutine which opens the chain server
// connection can pass the conn to the goroutine which opens the wallet.
// Buffer the channel so sends are not blocked, since if the wallet is
// not yet created, the wallet open goroutine does not read this.
chainSvrChan := make(chan *chain.Client, 1)
go func() {
// Read CA certs and create the RPC client.
certs, err := ioutil.ReadFile(cfg.CAFile)
if err != nil {
log.Errorf("Cannot open CA file: %v", err)
close(chainSvrChan)
return
}
rpcc, err := chain.NewClient(activeNet.Params, cfg.RPCConnect,
cfg.Username, cfg.Password, certs)
if err != nil {
log.Errorf("Cannot create chain server RPC client: %v", err)
close(chainSvrChan)
return
}
err = rpcc.Start()
if err != nil {
log.Warnf("Connection to Bitcoin RPC chain server " +
"unsuccessful -- available RPC methods will be limited")
}
// Even if Start errored, we still add the server disconnected.
// All client methods will then error, so it's obvious to a
// client that the there was a connection problem.
server.SetChainServer(rpcc)
chainSvrChan <- rpcc
}()
// Create a channel to report unrecoverable errors during the loading of
// the wallet files. These may include OS file handling errors or
// issues deserializing the wallet files, but does not include missing
// wallet files (as that must be handled by creating a new wallet).
walletOpenErrors := make(chan error)
go func() {
defer close(walletOpenErrors)
// Open wallet structures from disk.
w, err := openWallet()
if err != nil {
if os.IsNotExist(err) {
// If the keystore file is missing, notify the server
// that generating new wallets is ok.
server.SetWallet(nil)
return
} else {
// If the keystore file exists but another error was
// encountered, we cannot continue.
log.Errorf("Cannot load wallet files: %v", err)
walletOpenErrors <- err
return
}
}
server.SetWallet(w)
// Start wallet goroutines and handle RPC client notifications
// if the chain server connection was opened.
select {
case chainSvr := <-chainSvrChan:
w.Start(chainSvr)
case <-server.quit:
}
}()
// Check for unrecoverable errors during the wallet startup, and return
// the error, if any.
err, ok := <-walletOpenErrors
if ok {
return err
}
// Wait for the server to shutdown either due to a stop RPC request
// or an interrupt.
server.WaitForShutdown()
log.Info("Shutdown complete")
return nil
}

306
chain/chain.go Normal file
View file

@ -0,0 +1,306 @@
/*
* Copyright (c) 2013, 2014 Conformal Systems LLC <info@conformal.com>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
package chain
import (
"errors"
"sync"
"time"
"github.com/conformal/btcnet"
"github.com/conformal/btcrpcclient"
"github.com/conformal/btcutil"
"github.com/conformal/btcwallet/keystore"
"github.com/conformal/btcwallet/txstore"
"github.com/conformal/btcwire"
"github.com/conformal/btcws"
)
type Client struct {
*btcrpcclient.Client
netParams *btcnet.Params
enqueueNotification chan interface{}
dequeueNotification chan interface{}
currentBlock chan *keystore.BlockStamp
quit chan struct{}
wg sync.WaitGroup
started bool
quitMtx sync.Mutex
}
func NewClient(net *btcnet.Params, connect, user, pass string, certs []byte) (*Client, error) {
client := Client{
netParams: net,
enqueueNotification: make(chan interface{}),
dequeueNotification: make(chan interface{}),
currentBlock: make(chan *keystore.BlockStamp),
quit: make(chan struct{}),
}
initializedClient := make(chan struct{})
ntfnCallbacks := btcrpcclient.NotificationHandlers{
OnClientConnected: func() {
log.Info("Established connection to btcd")
},
OnBlockConnected: client.onBlockConnected,
OnBlockDisconnected: client.onBlockDisconnected,
OnRecvTx: client.onRecvTx,
OnRedeemingTx: client.onRedeemingTx,
OnRescanFinished: client.onRescanFinished,
OnRescanProgress: client.onRescanProgress,
}
conf := btcrpcclient.ConnConfig{
Host: connect,
Endpoint: "ws",
User: user,
Pass: pass,
Certificates: certs,
DisableConnectOnNew: true,
}
c, err := btcrpcclient.New(&conf, &ntfnCallbacks)
if err != nil {
return nil, err
}
client.Client = c
close(initializedClient)
return &client, nil
}
func (c *Client) Start() error {
err := c.Connect(5) // attempt connection 5 tries at most
if err != nil {
return err
}
// Verify that the server is running on the expected network.
net, err := c.GetCurrentNet()
if err != nil {
c.Disconnect()
return err
}
if net != c.netParams.Net {
c.Disconnect()
return errors.New("mismatched networks")
}
c.quitMtx.Lock()
c.started = true
c.quitMtx.Unlock()
c.wg.Add(1)
go c.handler()
return nil
}
func (c *Client) Stop() {
c.quitMtx.Lock()
defer c.quitMtx.Unlock()
select {
case <-c.quit:
default:
close(c.quit)
c.Client.Shutdown()
if !c.started {
close(c.dequeueNotification)
}
}
}
func (c *Client) WaitForShutdown() {
c.Client.WaitForShutdown()
c.wg.Wait()
}
func (c *Client) Notifications() <-chan interface{} {
return c.dequeueNotification
}
func (c *Client) BlockStamp() (*keystore.BlockStamp, error) {
select {
case bs := <-c.currentBlock:
return bs, nil
case <-c.quit:
return nil, errors.New("disconnected")
}
}
// Notification types. These are defined here and processed from from reading
// a notificationChan to avoid handling these notifications directly in
// btcrpcclient callbacks, which isn't very Go-like and doesn't allow
// blocking client calls.
type (
BlockConnected keystore.BlockStamp
BlockDisconnected keystore.BlockStamp
RecvTx struct {
Tx *btcutil.Tx // Index is guaranteed to be set.
Block *txstore.Block // nil if unmined
}
RedeemingTx struct {
Tx *btcutil.Tx // Index is guaranteed to be set.
Block *txstore.Block // nil if unmined
}
RescanProgress struct {
Hash *btcwire.ShaHash
Height int32
Time time.Time
}
RescanFinished struct {
Hash *btcwire.ShaHash
Height int32
Time time.Time
}
)
// parseBlock parses a btcws definition of the block a tx is mined it to the
// Block structure of the txstore package, and the block index. This is done
// here since btcrpcclient doesn't parse this nicely for us.
func parseBlock(block *btcws.BlockDetails) (blk *txstore.Block, idx int, err error) {
if block == nil {
return nil, btcutil.TxIndexUnknown, nil
}
blksha, err := btcwire.NewShaHashFromStr(block.Hash)
if err != nil {
return nil, btcutil.TxIndexUnknown, err
}
blk = &txstore.Block{
Height: block.Height,
Hash: *blksha,
Time: time.Unix(block.Time, 0),
}
return blk, block.Index, nil
}
func (c *Client) onBlockConnected(hash *btcwire.ShaHash, height int32) {
c.enqueueNotification <- BlockConnected{Hash: hash, Height: height}
}
func (c *Client) onBlockDisconnected(hash *btcwire.ShaHash, height int32) {
c.enqueueNotification <- BlockDisconnected{Hash: hash, Height: height}
}
func (c *Client) onRecvTx(tx *btcutil.Tx, block *btcws.BlockDetails) {
var blk *txstore.Block
index := btcutil.TxIndexUnknown
if block != nil {
var err error
blk, index, err = parseBlock(block)
if err != nil {
// Log and drop improper notification.
log.Errorf("recvtx notification bad block: %v", err)
return
}
}
tx.SetIndex(index)
c.enqueueNotification <- RecvTx{tx, blk}
}
func (c *Client) onRedeemingTx(tx *btcutil.Tx, block *btcws.BlockDetails) {
var blk *txstore.Block
index := btcutil.TxIndexUnknown
if block != nil {
var err error
blk, index, err = parseBlock(block)
if err != nil {
// Log and drop improper notification.
log.Errorf("recvtx notification bad block: %v", err)
return
}
}
tx.SetIndex(index)
c.enqueueNotification <- RedeemingTx{tx, blk}
}
func (c *Client) onRescanProgress(hash *btcwire.ShaHash, height int32, blkTime time.Time) {
c.enqueueNotification <- &RescanProgress{hash, height, blkTime}
}
func (c *Client) onRescanFinished(hash *btcwire.ShaHash, height int32, blkTime time.Time) {
c.enqueueNotification <- &RescanFinished{hash, height, blkTime}
}
// handler maintains a queue of notifications and the current state (best
// block) of the chain.
func (c *Client) handler() {
hash, height, err := c.GetBestBlock()
if err != nil {
close(c.quit)
c.wg.Done()
}
bs := &keystore.BlockStamp{Hash: hash, Height: height}
// TODO: Rather than leaving this as an unbounded queue for all types of
// notifications, try dropping ones where a later enqueued notification
// can fully invalidate one waiting to be processed. For example,
// blockconnected notifications for greater block heights can remove the
// need to process earlier blockconnected notifications still waiting
// here.
var notifications []interface{}
enqueue := c.enqueueNotification
var dequeue chan interface{}
var next interface{}
out:
for {
select {
case n, ok := <-enqueue:
if !ok {
// If no notifications are queued for handling,
// the queue is finished.
if len(notifications) == 0 {
break out
}
// nil channel so no more reads can occur.
enqueue = nil
continue
}
if len(notifications) == 0 {
next = n
dequeue = c.dequeueNotification
}
notifications = append(notifications, n)
case dequeue <- next:
if n, ok := next.(BlockConnected); ok {
bs = (*keystore.BlockStamp)(&n)
}
notifications[0] = nil
notifications = notifications[1:]
if len(notifications) != 0 {
next = notifications[0]
} else {
// If no more notifications can be enqueued, the
// queue is finished.
if enqueue == nil {
break out
}
dequeue = nil
}
case c.currentBlock <- bs:
case <-c.quit:
break out
}
}
close(c.dequeueNotification)
c.wg.Done()
}

68
chain/log.go Normal file
View file

@ -0,0 +1,68 @@
/*
* Copyright (c) 2013, 2014 Conformal Systems LLC <info@conformal.com>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
package chain
import "github.com/conformal/btclog"
// 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 btclog.Logger
// The default amount of logging is none.
func init() {
DisableLog()
}
// DisableLog disables all library log output. Logging output is disabled
// by default until either UseLogger or SetLogWriter are called.
func DisableLog() {
log = btclog.Disabled
}
// UseLogger uses a specified Logger to output package logging info.
// This should be used in preference to SetLogWriter if the caller is also
// using btclog.
func UseLogger(logger btclog.Logger) {
log = logger
}
// LogClosure is a closure that can be printed with %v to be used to
// generate expensive-to-create data for a detailed log level and avoid doing
// the work if the data isn't printed.
type logClosure func() string
// String invokes the log closure and returns the results string.
func (c logClosure) String() string {
return c()
}
// newLogClosure returns a new closure over the passed function which allows
// it to be used as a parameter in a logging function that is only invoked when
// the logging level is such that the message will actually be logged.
func newLogClosure(c func() string) logClosure {
return logClosure(c)
}
// pickNoun returns the singular or plural form of a noun depending
// on the count n.
func pickNoun(n int, singular, plural string) string {
if n == 1 {
return singular
}
return plural
}

182
chainntfns.go Normal file
View file

@ -0,0 +1,182 @@
/*
* Copyright (c) 2013, 2014 Conformal Systems LLC <info@conformal.com>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
package main
import (
"github.com/conformal/btcscript"
"github.com/conformal/btcutil"
"github.com/conformal/btcwallet/chain"
"github.com/conformal/btcwallet/keystore"
"github.com/conformal/btcwallet/txstore"
)
func (w *Wallet) handleChainNotifications() {
for n := range w.chainSvr.Notifications() {
var err error
switch n := n.(type) {
case chain.BlockConnected:
w.connectBlock(keystore.BlockStamp(n))
case chain.BlockDisconnected:
w.disconnectBlock(keystore.BlockStamp(n))
case chain.RecvTx:
err = w.addReceivedTx(n.Tx, n.Block)
case chain.RedeemingTx:
err = w.addRedeemingTx(n.Tx, n.Block)
// The following are handled by the wallet's rescan
// goroutines, so just pass them there.
case *chain.RescanProgress, *chain.RescanFinished:
w.rescanNotifications <- n
}
if err != nil {
log.Errorf("Cannot handle chain server "+
"notification: %v", err)
}
}
w.wg.Done()
}
// connectBlock handles a chain server notification by marking a wallet
// that's currently in-sync with the chain server as being synced up to
// the passed block.
func (w *Wallet) connectBlock(bs keystore.BlockStamp) {
if !w.ChainSynced() {
return
}
w.KeyStore.SetSyncedWith(&bs)
w.KeyStore.MarkDirty()
w.notifyConnectedBlock(bs)
w.notifyBalances(bs.Height)
}
// disconnectBlock handles a chain server reorganize by rolling back all
// block history from the reorged block for a wallet in-sync with the chain
// server.
func (w *Wallet) disconnectBlock(bs keystore.BlockStamp) {
if !w.ChainSynced() {
return
}
// Disconnect the last seen block from the keystore if it
// matches the removed block.
iter := w.KeyStore.NewIterateRecentBlocks()
if iter != nil && *iter.BlockStamp().Hash == *bs.Hash {
if iter.Prev() {
prev := iter.BlockStamp()
w.KeyStore.SetSyncedWith(&prev)
} else {
w.KeyStore.SetSyncedWith(nil)
}
w.KeyStore.MarkDirty()
}
w.notifyDisconnectedBlock(bs)
w.notifyBalances(bs.Height - 1)
}
func (w *Wallet) addReceivedTx(tx *btcutil.Tx, block *txstore.Block) error {
// For every output, if it pays to a wallet address, insert the
// transaction into the store (possibly moving it from unconfirmed to
// confirmed), and add a credit record if one does not already exist.
var txr *txstore.TxRecord
txInserted := false
for txOutIdx, txOut := range tx.MsgTx().TxOut {
// Errors don't matter here. If addrs is nil, the range below
// does nothing.
_, addrs, _, _ := btcscript.ExtractPkScriptAddrs(txOut.PkScript,
activeNet.Params)
insert := false
for _, addr := range addrs {
_, err := w.KeyStore.Address(addr)
if err == nil {
insert = true
break
}
}
if insert {
if !txInserted {
var err error
txr, err = w.TxStore.InsertTx(tx, block)
if err != nil {
return err
}
// InsertTx may have moved a previous unmined
// tx, so mark the entire store as dirty.
w.TxStore.MarkDirty()
txInserted = true
}
if txr.HasCredit(txOutIdx) {
continue
}
_, err := txr.AddCredit(uint32(txOutIdx), false)
if err != nil {
return err
}
w.TxStore.MarkDirty()
}
}
bs, err := w.chainSvr.BlockStamp()
if err == nil {
w.notifyBalances(bs.Height)
}
return nil
}
// addRedeemingTx inserts the notified spending transaction as a debit and
// schedules the transaction store for a future file write.
func (w *Wallet) addRedeemingTx(tx *btcutil.Tx, block *txstore.Block) error {
txr, err := w.TxStore.InsertTx(tx, block)
if err != nil {
return err
}
if _, err := txr.AddDebits(); err != nil {
return err
}
w.KeyStore.MarkDirty()
bs, err := w.chainSvr.BlockStamp()
if err == nil {
w.notifyBalances(bs.Height)
}
return nil
}
func (w *Wallet) notifyBalances(curHeight int32) {
// Don't notify unless wallet is synced to the chain server.
if !w.ChainSynced() {
return
}
// Notify any potential changes to the balance.
confirmed, err := w.TxStore.Balance(1, curHeight)
if err != nil {
log.Errorf("Cannot determine 1-conf balance: %v", err)
return
}
w.notifyConfirmedBalance(confirmed)
unconfirmed, err := w.TxStore.Balance(0, curHeight)
if err != nil {
log.Errorf("Cannot determine 0-conf balance: %v", err)
return
}
w.notifyUnconfirmedBalance(unconfirmed - confirmed)
}

161
cmd.go
View file

@ -1,161 +0,0 @@
/*
* Copyright (c) 2013, 2014 Conformal Systems LLC <info@conformal.com>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
package main
import (
"errors"
"io/ioutil"
"net"
"net/http"
_ "net/http/pprof"
"os"
"time"
)
var (
cfg *config
server *rpcServer
shutdownChan = make(chan struct{})
clientAccessChan = make(chan *rpcClient)
)
func clientAccess(newClient <-chan *rpcClient) {
var client *rpcClient
for {
select {
case c := <-newClient:
client = c
case clientAccessChan <- client:
}
}
}
func accessClient() (*rpcClient, error) {
c := <-clientAccessChan
if c == nil {
return nil, errors.New("chain server disconnected")
}
return c, nil
}
func clientConnect(certs []byte, newClient chan<- *rpcClient) {
const initialWait = 5 * time.Second
wait := initialWait
for {
select {
case <-server.quit:
return
default:
}
client, err := newRPCClient(certs)
if err != nil {
log.Warnf("Unable to open chain server client "+
"connection: %v", err)
time.Sleep(wait)
wait <<= 1
if wait > time.Minute {
wait = time.Minute
}
continue
}
wait = initialWait
client.Start()
newClient <- client
client.WaitForShutdown()
}
}
func main() {
// Work around defer not working after os.Exit.
if err := walletMain(); err != nil {
os.Exit(1)
}
}
// walletMain is a work-around main function that is required since deferred
// functions (such as log flushing) are not called with calls to os.Exit.
// Instead, main runs this function and checks for a non-nil error, at which
// point any defers have already run, and if the error is non-nil, the program
// can be exited with an error exit status.
func walletMain() error {
// Load configuration and parse command line. This function also
// initializes logging and configures it accordingly.
tcfg, _, err := loadConfig()
if err != nil {
return err
}
cfg = tcfg
defer backendLog.Flush()
if cfg.Profile != "" {
go func() {
listenAddr := net.JoinHostPort("", cfg.Profile)
log.Infof("Profile server listening on %s", listenAddr)
profileRedirect := http.RedirectHandler("/debug/pprof",
http.StatusSeeOther)
http.Handle("/", profileRedirect)
log.Errorf("%v", http.ListenAndServe(listenAddr, nil))
}()
}
// Read CA file to verify a btcd TLS connection.
certs, err := ioutil.ReadFile(cfg.CAFile)
if err != nil {
log.Errorf("cannot open CA file: %v", err)
return err
}
// Check and update any old file locations.
err = updateOldFileLocations()
if err != nil {
return err
}
// Start account manager and open accounts.
AcctMgr.Start()
server, err = newRPCServer(
cfg.SvrListeners,
cfg.RPCMaxClients,
cfg.RPCMaxWebsockets,
)
if err != nil {
log.Errorf("Unable to create HTTP server: %v", err)
return err
}
// Start HTTP server to serve wallet client connections.
server.Start()
// Shutdown the server if an interrupt signal is received.
addInterruptHandler(server.Stop)
// Start client connection to a btcd chain server. Attempt
// reconnections if the client could not be successfully connected.
clientChan := make(chan *rpcClient)
go clientAccess(clientChan)
go clientConnect(certs, clientChan)
// Wait for the server to shutdown either due to a stop RPC request
// or an interrupt.
server.WaitForShutdown()
log.Info("Shutdown complete")
return nil
}

4
config.go
View file

@ -36,7 +36,6 @@ const (
defaultLogLevel = "info"
defaultLogDirname = "logs"
defaultLogFilename = "btcwallet.log"
defaultKeypoolSize = 100
defaultDisallowFree = false
defaultRPCMaxClients = 10
defaultRPCMaxWebsockets = 25
@ -72,7 +71,7 @@ type config struct {
RPCMaxWebsockets int64 `long:"rpcmaxwebsockets" description:"Max number of RPC websocket connections"`
MainNet bool `long:"mainnet" description:"Use the main Bitcoin network (default testnet3)"`
SimNet bool `long:"simnet" description:"Use the simulation test network (default testnet3)"`
KeypoolSize uint `short:"k" long:"keypoolsize" description:"Maximum number of addresses in keypool"`
KeypoolSize uint `short:"k" long:"keypoolsize" description:"DEPRECATED -- Maximum number of addresses in keypool"`
DisallowFree bool `long:"disallowfree" description:"Force transactions to always include a fee"`
Proxy string `long:"proxy" description:"Connect via SOCKS5 proxy (eg. 127.0.0.1:9050)"`
ProxyUser string `long:"proxyuser" description:"Username for proxy server"`
@ -243,7 +242,6 @@ func loadConfig() (*config, []string, error) {
LogDir: defaultLogDir,
RPCKey: defaultRPCKeyFile,
RPCCert: defaultRPCCertFile,
KeypoolSize: defaultKeypoolSize,
DisallowFree: defaultDisallowFree,
RPCMaxClients: defaultRPCMaxClients,
RPCMaxWebsockets: defaultRPCMaxWebsockets,

52
createtx.go
View file

@ -66,8 +66,8 @@ const defaultFeeIncrement = 10000
type CreatedTx struct {
tx *btcutil.Tx
inputs []txstore.Credit
changeAddr btcutil.Address
changeIndex int // negative if no change
}
// ByAmount defines the methods needed to satisify sort.Interface to
@ -114,13 +114,17 @@ func selectInputs(eligible []txstore.Credit, amt, fee btcutil.Amount,
// address, changeUtxo will point to a unconfirmed (height = -1, zeroed
// block hash) Utxo. ErrInsufficientFunds is returned if there are not
// enough eligible unspent outputs to create the transaction.
func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
func (w *Wallet) txToPairs(pairs map[string]btcutil.Amount,
minconf int) (*CreatedTx, error) {
// Wallet must be unlocked to compose transaction.
if a.KeyStore.IsLocked() {
return nil, keystore.ErrLocked
// Key store must be unlocked to compose transaction. Grab the
// unlock if possible (to prevent future unlocks), or return the
// error if the keystore is already locked.
heldUnlock, err := w.HoldUnlock()
if err != nil {
return nil, err
}
defer heldUnlock.Release()
// Create a new transaction which will include all input scripts.
msgtx := btcwire.NewMsgTx()
@ -152,11 +156,7 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
}
// Get current block's height and hash.
rpcc, err := accessClient()
if err != nil {
return nil, err
}
bs, err := rpcc.BlockStamp()
bs, err := w.chainSvr.BlockStamp()
if err != nil {
return nil, err
}
@ -166,7 +166,7 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
// a higher fee if not enough was originally chosen.
txNoInputs := msgtx.Copy()
unspent, err := a.TxStore.UnspentOutputs()
unspent, err := w.TxStore.UnspentOutputs()
if err != nil {
return nil, err
}
@ -192,7 +192,7 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
}
// Locked unspent outputs are skipped.
if a.LockedOutpoint(*unspent[i].OutPoint()) {
if w.LockedOutpoint(*unspent[i].OutPoint()) {
continue
}
@ -208,12 +208,14 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
// changeAddr is nil/zeroed until a change address is needed, and reused
// again in case a change utxo has already been chosen.
var changeAddr btcutil.Address
var changeIdx int
// Get the number of satoshis to increment fee by when searching for
// the minimum tx fee needed.
fee := btcutil.Amount(0)
for {
msgtx = txNoInputs.Copy()
changeIdx = -1
// Select eligible outputs to be used in transaction based on the amount
// neededing to sent, and the current fee estimation.
@ -228,13 +230,16 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
if change > 0 {
// Get a new change address if one has not already been found.
if changeAddr == nil {
changeAddr, err = a.KeyStore.ChangeAddress(&bs, cfg.KeypoolSize)
changeAddr, err = w.KeyStore.ChangeAddress(bs)
if err != nil {
return nil, fmt.Errorf("failed to get next address: %s", err)
}
// Mark change address as belonging to this account.
AcctMgr.MarkAddressForAccount(changeAddr, a)
w.KeyStore.MarkDirty()
err = w.chainSvr.NotifyReceived([]btcutil.Address{changeAddr})
if err != nil {
return nil, fmt.Errorf("cannot request updates for "+
"change address: %v", err)
}
}
// Spend change.
@ -249,6 +254,7 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
r := rng.Int31n(int32(len(msgtx.TxOut))) // random index
c := len(msgtx.TxOut) - 1 // change index
msgtx.TxOut[r], msgtx.TxOut[c] = msgtx.TxOut[c], msgtx.TxOut[r]
changeIdx = int(r)
}
// Selected unspent outputs become new transaction's inputs.
@ -264,10 +270,10 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
}
apkh, ok := addrs[0].(*btcutil.AddressPubKeyHash)
if !ok {
continue // don't handle inputs to this yes
continue // don't handle inputs to this yet
}
ai, err := a.KeyStore.Address(apkh)
ai, err := w.KeyStore.Address(apkh)
if err != nil {
return nil, fmt.Errorf("cannot get address info: %v", err)
}
@ -275,10 +281,8 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
pka := ai.(keystore.PubKeyAddress)
privkey, err := pka.PrivKey()
if err == keystore.ErrLocked {
return nil, keystore.ErrLocked
} else if err != nil {
return nil, fmt.Errorf("cannot get address key: %v", err)
if err != nil {
return nil, fmt.Errorf("cannot get private key: %v", err)
}
sigscript, err := btcscript.SignatureScript(msgtx, i,
@ -294,7 +298,7 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
if !cfg.DisallowFree {
noFeeAllowed = allowFree(bs.Height, inputs, msgtx.SerializeSize())
}
if minFee := minimumFee(a.FeeIncrement, msgtx, noFeeAllowed); fee < minFee {
if minFee := minimumFee(w.FeeIncrement, msgtx, noFeeAllowed); fee < minFee {
fee = minFee
} else {
selectedInputs = inputs
@ -328,8 +332,8 @@ func (a *Account) txToPairs(pairs map[string]btcutil.Amount,
}
info := &CreatedTx{
tx: btcutil.NewTx(msgtx),
inputs: selectedInputs,
changeAddr: changeAddr,
changeIndex: changeIdx,
}
return info, nil
}

2
createtx_test.go
View file

@ -89,7 +89,7 @@ func TestFakeTxs(t *testing.T) {
t.Errorf("Can not create encrypted wallet: %s", err)
return
}
a := &Account{
a := &Wallet{
Wallet: w,
lockedOutpoints: map[btcwire.OutPoint]struct{}{},
}

396
disksync.go
View file

@ -18,68 +18,9 @@ package main
import (
"fmt"
"io/ioutil"
"os"
"path/filepath"
"time"
"github.com/conformal/btcnet"
"github.com/conformal/btcwire"
)
// networkDir returns the directory name of a network directory to hold account
// files.
func networkDir(net *btcnet.Params) string {
netname := net.Name
// For now, we must always name the testnet data directory as "testnet"
// and not "testnet3" or any other version, as the btcnet testnet3
// paramaters will likely be switched to being named "testnet3" in the
// future. This is done to future proof that change, and an upgrade
// plan to move the testnet3 data directory can be worked out later.
if net.Net == btcwire.TestNet3 {
netname = "testnet"
}
return filepath.Join(cfg.DataDir, netname)
}
// tmpNetworkDir returns the temporary directory name for a given network.
func tmpNetworkDir(net *btcnet.Params) string {
return networkDir(net) + "_tmp"
}
// freshDir creates a new directory specified by path if it does not
// exist. If the directory already exists, all files contained in the
// directory are removed.
func freshDir(path string) error {
if err := checkCreateDir(path); err != nil {
return err
}
// Remove all files in the directory.
fd, err := os.Open(path)
if err != nil {
return err
}
defer func() {
if err := fd.Close(); err != nil {
log.Warnf("Cannot close directory: %v", err)
}
}()
names, err := fd.Readdirnames(0)
if err != nil {
return err
}
for _, name := range names {
if err := os.RemoveAll(name); err != nil {
return err
}
}
return nil
}
// checkCreateDir checks that the path exists and is a directory.
// If path does not exist, it is created.
func checkCreateDir(path string) error {
@ -100,340 +41,3 @@ func checkCreateDir(path string) error {
return nil
}
// accountFilename returns the filepath of an account file given the
// filename suffix ("wallet.bin", or "tx.bin"), account name and the
// network directory holding the file.
func accountFilename(suffix, account, netdir string) string {
if account == "" {
// default account
return filepath.Join(netdir, suffix)
}
// non-default account
return filepath.Join(netdir, fmt.Sprintf("%v-%v", account, suffix))
}
// syncSchedule references the account files which have been
// scheduled to be written and the directory to write to.
type syncSchedule struct {
dir string
wallets map[*Account]struct{}
txs map[*Account]struct{}
}
func newSyncSchedule(dir string) *syncSchedule {
s := &syncSchedule{
dir: dir,
wallets: make(map[*Account]struct{}),
txs: make(map[*Account]struct{}),
}
return s
}
// flushAccount writes all scheduled account files to disk for
// a single account and removes them from the schedule.
func (s *syncSchedule) flushAccount(a *Account) error {
if _, ok := s.txs[a]; ok {
if err := a.writeTxStore(s.dir); err != nil {
return err
}
delete(s.txs, a)
}
if _, ok := s.wallets[a]; ok {
if err := a.writeWallet(s.dir); err != nil {
return err
}
delete(s.wallets, a)
}
return nil
}
// flush writes all scheduled account files and removes each
// from the schedule.
func (s *syncSchedule) flush() error {
for a := range s.txs {
if err := a.writeTxStore(s.dir); err != nil {
return err
}
delete(s.txs, a)
}
for a := range s.wallets {
if err := a.writeWallet(s.dir); err != nil {
return err
}
delete(s.wallets, a)
}
return nil
}
type flushAccountRequest struct {
a *Account
err chan error
}
type writeBatchRequest struct {
a []*Account
err chan error
}
type exportRequest struct {
dir string
a *Account
err chan error
}
// DiskSyncer manages all disk write operations for a collection of accounts.
type DiskSyncer struct {
// Flush scheduled account writes.
flushAccount chan *flushAccountRequest
// Schedule file writes for an account.
scheduleWallet chan *Account
scheduleTxStore chan *Account
// Write a collection of accounts all at once.
writeBatch chan *writeBatchRequest
// Write an account export.
exportAccount chan *exportRequest
// Account manager for this DiskSyncer. This is only
// needed to grab the account manager semaphore.
am *AccountManager
quit chan struct{}
shutdown chan struct{}
}
// NewDiskSyncer creates a new DiskSyncer.
func NewDiskSyncer(am *AccountManager) *DiskSyncer {
return &DiskSyncer{
flushAccount: make(chan *flushAccountRequest),
scheduleWallet: make(chan *Account),
scheduleTxStore: make(chan *Account),
writeBatch: make(chan *writeBatchRequest),
exportAccount: make(chan *exportRequest),
am: am,
quit: make(chan struct{}),
shutdown: make(chan struct{}),
}
}
// Start starts the goroutines required to run the DiskSyncer.
func (ds *DiskSyncer) Start() {
go ds.handler()
}
func (ds *DiskSyncer) Stop() {
close(ds.quit)
}
func (ds *DiskSyncer) WaitForShutdown() {
<-ds.shutdown
}
// handler runs the disk syncer. It manages a set of "dirty" account files
// which must be written to disk, and synchronizes all writes in a single
// goroutine. Periodic flush operations may be signaled by an AccountManager.
//
// This never returns and is should be called from a new goroutine.
func (ds *DiskSyncer) handler() {
netdir := networkDir(activeNet.Params)
if err := checkCreateDir(netdir); err != nil {
log.Errorf("Unable to create or write to account directory: %v", err)
}
tmpnetdir := tmpNetworkDir(activeNet.Params)
const wait = 10 * time.Second
var timer <-chan time.Time
var sem chan struct{}
schedule := newSyncSchedule(netdir)
out:
for {
select {
case <-sem: // Now have exclusive access of the account manager
err := schedule.flush()
if err != nil {
log.Errorf("Cannot write accounts: %v", err)
}
timer = nil
// Do not grab semaphore again until another flush is needed.
sem = nil
// Release semaphore.
ds.am.bsem <- struct{}{}
case <-timer:
// Grab AccountManager semaphore when ready so flush can occur.
sem = ds.am.bsem
case fr := <-ds.flushAccount:
fr.err <- schedule.flushAccount(fr.a)
case a := <-ds.scheduleWallet:
schedule.wallets[a] = struct{}{}
if timer == nil {
timer = time.After(wait)
}
case a := <-ds.scheduleTxStore:
schedule.txs[a] = struct{}{}
if timer == nil {
timer = time.After(wait)
}
case sr := <-ds.writeBatch:
err := batchWriteAccounts(sr.a, tmpnetdir, netdir)
if err == nil {
// All accounts have been synced, old schedule
// can be discarded.
schedule = newSyncSchedule(netdir)
timer = nil
}
sr.err <- err
case er := <-ds.exportAccount:
a := er.a
dir := er.dir
er.err <- a.writeAll(dir)
case <-ds.quit:
err := schedule.flush()
if err != nil {
log.Errorf("Cannot write accounts: %v", err)
}
break out
}
}
close(ds.shutdown)
}
// FlushAccount writes all scheduled account files to disk for a single
// account.
func (ds *DiskSyncer) FlushAccount(a *Account) error {
err := make(chan error)
ds.flushAccount <- &flushAccountRequest{a: a, err: err}
return <-err
}
// ScheduleWalletWrite schedules an account's wallet to be written to disk.
func (ds *DiskSyncer) ScheduleWalletWrite(a *Account) {
ds.scheduleWallet <- a
}
// ScheduleTxStoreWrite schedules an account's transaction store to be
// written to disk.
func (ds *DiskSyncer) ScheduleTxStoreWrite(a *Account) {
ds.scheduleTxStore <- a
}
// WriteBatch safely replaces all account files in the network directory
// with new files created from all accounts in a.
func (ds *DiskSyncer) WriteBatch(a []*Account) error {
err := make(chan error)
ds.writeBatch <- &writeBatchRequest{
a: a,
err: err,
}
return <-err
}
// ExportAccount writes all account files for a to a new directory.
func (ds *DiskSyncer) ExportAccount(a *Account, dir string) error {
err := make(chan error)
er := &exportRequest{
dir: dir,
a: a,
err: err,
}
ds.exportAccount <- er
return <-err
}
func batchWriteAccounts(accts []*Account, tmpdir, netdir string) error {
if err := freshDir(tmpdir); err != nil {
return err
}
for _, a := range accts {
if err := a.writeAll(tmpdir); err != nil {
return err
}
}
// This is technically NOT an atomic operation, but at startup, if the
// network directory is missing but the temporary network directory
// exists, the temporary is moved before accounts are opened.
if err := os.RemoveAll(netdir); err != nil {
return err
}
if err := Rename(tmpdir, netdir); err != nil {
return err
}
return nil
}
func (a *Account) writeAll(dir string) error {
if err := a.writeTxStore(dir); err != nil {
return err
}
if err := a.writeWallet(dir); err != nil {
return err
}
return nil
}
func (a *Account) writeWallet(dir string) error {
wfilepath := accountFilename("wallet.bin", a.name, dir)
_, filename := filepath.Split(wfilepath)
tmpfile, err := ioutil.TempFile(dir, filename)
if err != nil {
return err
}
if _, err = a.KeyStore.WriteTo(tmpfile); err != nil {
return err
}
tmppath := tmpfile.Name()
if err := tmpfile.Sync(); err != nil {
log.Warnf("Failed to sync temporary wallet file %s: %v",
tmppath, err)
}
if err := tmpfile.Close(); err != nil {
log.Warnf("Cannot close temporary wallet file %s: %v",
tmppath, err)
}
return Rename(tmppath, wfilepath)
}
func (a *Account) writeTxStore(dir string) error {
txfilepath := accountFilename("tx.bin", a.name, dir)
_, filename := filepath.Split(txfilepath)
tmpfile, err := ioutil.TempFile(dir, filename)
if err != nil {
return err
}
if _, err = a.TxStore.WriteTo(tmpfile); err != nil {
return err
}
tmppath := tmpfile.Name()
if err := tmpfile.Sync(); err != nil {
log.Warnf("Failed to sync temporary txstore file %s: %v",
tmppath, err)
}
if err := tmpfile.Close(); err != nil {
log.Warnf("Cannot close temporary txstore file %s: %v",
tmppath, err)
}
return Rename(tmppath, txfilepath)
}

693
keystore/keystore.go
View file

File diff suppressed because it is too large Load diff

178
keystore/keystore_test.go
View file

@ -28,12 +28,22 @@ import (
"github.com/conformal/btcnet"
"github.com/conformal/btcscript"
"github.com/conformal/btcutil"
"github.com/conformal/btcwire"
"github.com/davecgh/go-spew/spew"
)
const dummyDir = ""
var tstNetParams = &btcnet.MainNetParams
func makeBS(height int32) *BlockStamp {
return &BlockStamp{
Hash: new(btcwire.ShaHash),
Height: height,
}
}
func TestBtcAddressSerializer(t *testing.T) {
fakeWallet := &Store{net: (*netParams)(tstNetParams)}
kdfp := &kdfParameters{
@ -44,14 +54,14 @@ func TestBtcAddressSerializer(t *testing.T) {
t.Error(err.Error())
return
}
key := Key([]byte("banana"), kdfp)
key := kdf([]byte("banana"), kdfp)
privKey := make([]byte, 32)
if _, err := rand.Read(privKey); err != nil {
t.Error(err.Error())
return
}
addr, err := newBtcAddress(fakeWallet, privKey, nil,
&BlockStamp{}, true)
makeBS(0), true)
if err != nil {
t.Error(err.Error())
return
@ -91,7 +101,7 @@ func TestScriptAddressSerializer(t *testing.T) {
fakeWallet := &Store{net: (*netParams)(tstNetParams)}
script := []byte{btcscript.OP_TRUE, btcscript.OP_DUP,
btcscript.OP_DROP}
addr, err := newScriptAddress(fakeWallet, script, &BlockStamp{})
addr, err := newScriptAddress(fakeWallet, script, makeBS(0))
if err != nil {
t.Error(err.Error())
return
@ -118,9 +128,9 @@ func TestScriptAddressSerializer(t *testing.T) {
}
func TestWalletCreationSerialization(t *testing.T) {
createdAt := &BlockStamp{}
w1, err := NewStore("banana wallet", "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt, 100)
createdAt := makeBS(0)
w1, err := New(dummyDir, "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt)
if err != nil {
t.Error("Error creating new wallet: " + err.Error())
return
@ -215,16 +225,16 @@ func TestChaining(t *testing.T) {
// Create next chained private keys, chained from both the uncompressed
// and compressed pubkeys.
nextPrivUncompressed, err := ChainedPrivKey(test.origPrivateKey,
nextPrivUncompressed, err := chainedPrivKey(test.origPrivateKey,
origPubUncompressed, test.cc)
if err != nil {
t.Errorf("%s: Uncompressed ChainedPrivKey failed: %v", test.name, err)
t.Errorf("%s: Uncompressed chainedPrivKey failed: %v", test.name, err)
return
}
nextPrivCompressed, err := ChainedPrivKey(test.origPrivateKey,
nextPrivCompressed, err := chainedPrivKey(test.origPrivateKey,
origPubCompressed, test.cc)
if err != nil {
t.Errorf("%s: Compressed ChainedPrivKey failed: %v", test.name, err)
t.Errorf("%s: Compressed chainedPrivKey failed: %v", test.name, err)
return
}
@ -247,14 +257,14 @@ func TestChaining(t *testing.T) {
// Create the next pubkeys by chaining directly off the original
// pubkeys (without using the original's private key).
nextPubUncompressedFromPub, err := ChainedPubKey(origPubUncompressed, test.cc)
nextPubUncompressedFromPub, err := chainedPubKey(origPubUncompressed, test.cc)
if err != nil {
t.Errorf("%s: Uncompressed ChainedPubKey failed: %v", test.name, err)
t.Errorf("%s: Uncompressed chainedPubKey failed: %v", test.name, err)
return
}
nextPubCompressedFromPub, err := ChainedPubKey(origPubCompressed, test.cc)
nextPubCompressedFromPub, err := chainedPubKey(origPubCompressed, test.cc)
if err != nil {
t.Errorf("%s: Compressed ChainedPubKey failed: %v", test.name, err)
t.Errorf("%s: Compressed chainedPubKey failed: %v", test.name, err)
return
}
@ -328,11 +338,8 @@ func TestChaining(t *testing.T) {
}
func TestWalletPubkeyChaining(t *testing.T) {
// Set a reasonable keypool size that isn't too big nor too small for testing.
const keypoolSize = 5
w, err := NewStore("banana wallet", "A wallet for testing.",
[]byte("banana"), tstNetParams, &BlockStamp{}, keypoolSize)
w, err := New(dummyDir, "A wallet for testing.",
[]byte("banana"), tstNetParams, makeBS(0))
if err != nil {
t.Error("Error creating new wallet: " + err.Error())
return
@ -341,20 +348,9 @@ func TestWalletPubkeyChaining(t *testing.T) {
t.Error("New wallet is not locked.")
}
// Wallet should have a total of 6 addresses, one for the root, plus 5 in
// the keypool with their private keys set. Ask for as many new addresses
// as needed to deplete the pool.
for i := 0; i < keypoolSize; i++ {
_, err := w.NextChainedAddress(&BlockStamp{}, keypoolSize)
if err != nil {
t.Errorf("Error getting next address from keypool: %v", err)
return
}
}
// Get next chained address after depleting the keypool. This will extend
// the chain based on the last pubkey, not privkey.
addrWithoutPrivkey, err := w.NextChainedAddress(&BlockStamp{}, keypoolSize)
// Get next chained address. The wallet is locked, so this will chain
// off the last pubkey, not privkey.
addrWithoutPrivkey, err := w.NextChainedAddress(makeBS(0))
if err != nil {
t.Errorf("Failed to extend address chain from pubkey: %v", err)
return
@ -456,13 +452,9 @@ func TestWalletPubkeyChaining(t *testing.T) {
return
}
// Test that normal keypool extension and address creation continues to
// work. With the wallet still unlocked, create a new address. This
// will cause the keypool to refill and return the first address from the
// keypool.
nextAddr, err := w.NextChainedAddress(&BlockStamp{}, keypoolSize)
nextAddr, err := w.NextChainedAddress(makeBS(0))
if err != nil {
t.Errorf("Unable to create next address or refill keypool after finding the privkey: %v", err)
t.Errorf("Unable to create next address after finding the privkey: %v", err)
return
}
@ -505,10 +497,9 @@ func TestWalletPubkeyChaining(t *testing.T) {
}
func TestWatchingWalletExport(t *testing.T) {
const keypoolSize = 10
createdAt := &BlockStamp{}
w, err := NewStore("banana wallet", "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt, keypoolSize)
createdAt := makeBS(0)
w, err := New(dummyDir, "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt)
if err != nil {
t.Error("Error creating new wallet: " + err.Error())
return
@ -520,19 +511,6 @@ func TestWatchingWalletExport(t *testing.T) {
// Add root address.
activeAddrs[getAddressKey(w.LastChainedAddress())] = struct{}{}
// Get as many new active addresses as necessary to deplete the keypool.
// This is done as we will want to test that new addresses created by
// the watching wallet do not pull from previous public keys in the
// original keypool.
for i := 0; i < keypoolSize; i++ {
apkh, err := w.NextChainedAddress(createdAt, keypoolSize)
if err != nil {
t.Errorf("unable to get next address: %v", err)
return
}
activeAddrs[getAddressKey(apkh)] = struct{}{}
}
// Create watching wallet from w.
ww, err := w.ExportWatchingWallet()
if err != nil {
@ -606,40 +584,23 @@ func TestWatchingWalletExport(t *testing.T) {
}
// Check that the new addresses created by each wallet match. The
// original wallet is unlocked so the keypool is refilled and chained
// addresses use the previous' privkey, not pubkey.
// original wallet is unlocked so addresses are chained with privkeys.
if err := w.Unlock([]byte("banana")); err != nil {
t.Errorf("Unlocking original wallet failed: %v", err)
}
for i := 0; i < keypoolSize; i++ {
addr, err := w.NextChainedAddress(createdAt, keypoolSize)
if err != nil {
t.Errorf("Cannot get next chained address for original wallet: %v", err)
return
}
waddr, err := ww.NextChainedAddress(createdAt, keypoolSize)
if err != nil {
t.Errorf("Cannot get next chained address for watching wallet: %v", err)
return
}
if addr.EncodeAddress() != waddr.EncodeAddress() {
t.Errorf("Next addresses for each wallet do not match eachother.")
return
}
}
// Test that ExtendActiveAddresses for the watching wallet match
// manually requested addresses of the original wallet.
newAddrs := make([]btcutil.Address, 0, keypoolSize)
for i := 0; i < keypoolSize; i++ {
addr, err := w.NextChainedAddress(createdAt, keypoolSize)
var newAddrs []btcutil.Address
for i := 0; i < 10; i++ {
addr, err := w.NextChainedAddress(createdAt)
if err != nil {
t.Errorf("Cannot get next chained address for original wallet: %v", err)
return
}
newAddrs = append(newAddrs, addr)
}
newWWAddrs, err := ww.ExtendActiveAddresses(keypoolSize, keypoolSize)
newWWAddrs, err := ww.ExtendActiveAddresses(10)
if err != nil {
t.Errorf("Cannot extend active addresses for watching wallet: %v", err)
return
@ -653,16 +614,16 @@ func TestWatchingWalletExport(t *testing.T) {
// Test ExtendActiveAddresses for the original wallet after manually
// requesting addresses for the watching wallet.
newWWAddrs = make([]btcutil.Address, 0, keypoolSize)
for i := 0; i < keypoolSize; i++ {
addr, err := ww.NextChainedAddress(createdAt, keypoolSize)
newWWAddrs = nil
for i := 0; i < 10; i++ {
addr, err := ww.NextChainedAddress(createdAt)
if err != nil {
t.Errorf("Cannot get next chained address for watching wallet: %v", err)
return
}
newWWAddrs = append(newWWAddrs, addr)
}
newAddrs, err = w.ExtendActiveAddresses(keypoolSize, keypoolSize)
newAddrs, err = w.ExtendActiveAddresses(10)
if err != nil {
t.Errorf("Cannot extend active addresses for original wallet: %v", err)
return
@ -728,11 +689,10 @@ func TestWatchingWalletExport(t *testing.T) {
}
func TestImportPrivateKey(t *testing.T) {
const keypoolSize = 10
createHeight := int32(100)
createdAt := &BlockStamp{Height: createHeight}
w, err := NewStore("banana wallet", "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt, keypoolSize)
createdAt := makeBS(createHeight)
w, err := New(dummyDir, "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt)
if err != nil {
t.Error("Error creating new wallet: " + err.Error())
return
@ -751,7 +711,7 @@ func TestImportPrivateKey(t *testing.T) {
// verify that the entire wallet's sync height matches the
// expected createHeight.
if h := w.SyncHeight(); h != createHeight {
if _, h := w.SyncedTo(); h != createHeight {
t.Errorf("Initial sync height %v does not match expected %v.", h, createHeight)
return
}
@ -762,7 +722,7 @@ func TestImportPrivateKey(t *testing.T) {
t.Fatal(err)
}
importHeight := int32(50)
importedAt := &BlockStamp{Height: importHeight}
importedAt := makeBS(importHeight)
address, err := w.ImportPrivateKey(wif, importedAt)
if err != nil {
t.Error("importing private key: " + err.Error())
@ -788,7 +748,7 @@ func TestImportPrivateKey(t *testing.T) {
}
// verify that the sync height now match the (smaller) import height.
if h := w.SyncHeight(); h != importHeight {
if _, h := w.SyncedTo(); h != importHeight {
t.Errorf("After import sync height %v does not match expected %v.", h, importHeight)
return
}
@ -810,7 +770,7 @@ func TestImportPrivateKey(t *testing.T) {
}
// Verify that the sync height match expected after the reserialization.
if h := w2.SyncHeight(); h != importHeight {
if _, h := w2.SyncedTo(); h != importHeight {
t.Errorf("After reserialization sync height %v does not match expected %v.", h, importHeight)
return
}
@ -822,7 +782,7 @@ func TestImportPrivateKey(t *testing.T) {
t.Errorf("Cannot mark address partially synced: %v", err)
return
}
if h := w2.SyncHeight(); h != partialHeight {
if _, h := w2.SyncedTo(); h != partialHeight {
t.Errorf("After address partial sync, sync height %v does not match expected %v.", h, partialHeight)
return
}
@ -842,7 +802,7 @@ func TestImportPrivateKey(t *testing.T) {
}
// Test correct partial height after serialization.
if h := w3.SyncHeight(); h != partialHeight {
if _, h := w3.SyncedTo(); h != partialHeight {
t.Errorf("After address partial sync and reserialization, sync height %v does not match expected %v.",
h, partialHeight)
return
@ -854,7 +814,7 @@ func TestImportPrivateKey(t *testing.T) {
t.Errorf("Cannot mark address synced: %v", err)
return
}
if h := w3.SyncHeight(); h != importHeight {
if _, h := w3.SyncedTo(); h != importHeight {
t.Errorf("After address unsync, sync height %v does not match expected %v.", h, importHeight)
return
}
@ -866,7 +826,7 @@ func TestImportPrivateKey(t *testing.T) {
t.Errorf("Cannot mark address synced: %v", err)
return
}
if h := w3.SyncHeight(); h != createHeight {
if _, h := w3.SyncedTo(); h != createHeight {
t.Errorf("After address sync, sync height %v does not match expected %v.", h, createHeight)
return
}
@ -898,11 +858,10 @@ func TestImportPrivateKey(t *testing.T) {
}
func TestImportScript(t *testing.T) {
const keypoolSize = 10
createHeight := int32(100)
createdAt := &BlockStamp{Height: createHeight}
w, err := NewStore("banana wallet", "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt, keypoolSize)
createdAt := makeBS(createHeight)
w, err := New(dummyDir, "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt)
if err != nil {
t.Error("Error creating new wallet: " + err.Error())
return
@ -915,7 +874,7 @@ func TestImportScript(t *testing.T) {
// verify that the entire wallet's sync height matches the
// expected createHeight.
if h := w.SyncHeight(); h != createHeight {
if _, h := w.SyncedTo(); h != createHeight {
t.Errorf("Initial sync height %v does not match expected %v.", h, createHeight)
return
}
@ -923,7 +882,7 @@ func TestImportScript(t *testing.T) {
script := []byte{btcscript.OP_TRUE, btcscript.OP_DUP,
btcscript.OP_DROP}
importHeight := int32(50)
stamp := &BlockStamp{Height: importHeight}
stamp := makeBS(importHeight)
address, err := w.ImportScript(script, stamp)
if err != nil {
t.Error("error importing script: " + err.Error())
@ -989,7 +948,7 @@ func TestImportScript(t *testing.T) {
}
// verify that the sync height now match the (smaller) import height.
if h := w.SyncHeight(); h != importHeight {
if _, h := w.SyncedTo(); h != importHeight {
t.Errorf("After import sync height %v does not match expected %v.", h, importHeight)
return
}
@ -1028,7 +987,7 @@ func TestImportScript(t *testing.T) {
}
// Verify that the sync height matches expected after the reserialization.
if h := w2.SyncHeight(); h != importHeight {
if _, h := w2.SyncedTo(); h != importHeight {
t.Errorf("After reserialization sync height %v does not match expected %v.", h, importHeight)
return
}
@ -1125,7 +1084,7 @@ func TestImportScript(t *testing.T) {
t.Errorf("Cannot mark address partially synced: %v", err)
return
}
if h := w2.SyncHeight(); h != partialHeight {
if _, h := w2.SyncedTo(); h != partialHeight {
t.Errorf("After address partial sync, sync height %v does not match expected %v.", h, partialHeight)
return
}
@ -1145,7 +1104,7 @@ func TestImportScript(t *testing.T) {
}
// Test correct partial height after serialization.
if h := w3.SyncHeight(); h != partialHeight {
if _, h := w3.SyncedTo(); h != partialHeight {
t.Errorf("After address partial sync and reserialization, sync height %v does not match expected %v.",
h, partialHeight)
return
@ -1157,7 +1116,7 @@ func TestImportScript(t *testing.T) {
t.Errorf("Cannot mark address synced: %v", err)
return
}
if h := w3.SyncHeight(); h != importHeight {
if _, h := w3.SyncedTo(); h != importHeight {
t.Errorf("After address unsync, sync height %v does not match expected %v.", h, importHeight)
return
}
@ -1169,7 +1128,7 @@ func TestImportScript(t *testing.T) {
t.Errorf("Cannot mark address synced: %v", err)
return
}
if h := w3.SyncHeight(); h != createHeight {
if _, h := w3.SyncedTo(); h != createHeight {
t.Errorf("After address sync, sync height %v does not match expected %v.", h, createHeight)
return
}
@ -1181,10 +1140,9 @@ func TestImportScript(t *testing.T) {
}
func TestChangePassphrase(t *testing.T) {
const keypoolSize = 10
createdAt := &BlockStamp{}
w, err := NewStore("banana wallet", "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt, keypoolSize)
createdAt := makeBS(0)
w, err := New(dummyDir, "A wallet for testing.",
[]byte("banana"), tstNetParams, createdAt)
if err != nil {
t.Error("Error creating new wallet: " + err.Error())
return

6
log.go
View file

@ -21,6 +21,7 @@ import (
"os"
"github.com/conformal/btclog"
"github.com/conformal/btcwallet/chain"
"github.com/conformal/btcwallet/txstore"
"github.com/conformal/seelog"
)
@ -42,12 +43,14 @@ var (
backendLog = seelog.Disabled
log = btclog.Disabled
txstLog = btclog.Disabled
chainLog = btclog.Disabled
)
// subsystemLoggers maps each subsystem identifier to its associated logger.
var subsystemLoggers = map[string]btclog.Logger{
"BTCW": log,
"TXST": txstLog,
"CHNS": chainLog,
}
// logClosure is used to provide a closure over expensive logging operations
@ -80,6 +83,9 @@ func useLogger(subsystemID string, logger btclog.Logger) {
case "TXST":
txstLog = logger
txstore.UseLogger(logger)
case "CHNS":
chainLog = logger
chain.UseLogger(logger)
}
}

6
rename_plan9.go → rename/rename_plan9.go
View file

@ -26,16 +26,16 @@
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
package main
package rename
import (
"os"
"path/filepath"
)
// Rename provides an atomic file rename. newpath is replaced if it
// Atomic provides an atomic file rename. newpath is replaced if it
// already exists.
func Rename(oldpath, newpath string) error {
func Atomic(oldpath, newpath string) error {
if _, err := os.Stat(newpath); err == nil {
if err := os.Remove(newpath); err != nil {
return err

6
rename_unix.go → rename/rename_unix.go
View file

@ -14,14 +14,14 @@
// +build !windows,!plan9
package main
package rename
import (
"os"
)
// Rename provides an atomic file rename. newpath is replaced if it
// Atomic provides an atomic file rename. newpath is replaced if it
// already exists.
func Rename(oldpath, newpath string) error {
func Atomic(oldpath, newpath string) error {
return os.Rename(oldpath, newpath)
}

6
rename_windows.go → rename/rename_windows.go
View file

@ -26,7 +26,7 @@
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
package main
package rename
import (
"syscall"
@ -56,9 +56,9 @@ func moveFileEx(from *uint16, to *uint16, flags uint32) error {
return nil
}
// Rename provides an atomic file rename. newpath is replaced if it
// Atomic provides an atomic file rename. newpath is replaced if it
// already exists.
func Rename(oldpath, newpath string) error {
func Atomic(oldpath, newpath string) error {
from, err := syscall.UTF16PtrFromString(oldpath)
if err != nil {
return err

454
rescan.go
View file

@ -17,282 +17,302 @@
package main
import (
"sync"
"github.com/conformal/btcutil"
"github.com/conformal/btcwallet/chain"
"github.com/conformal/btcwallet/keystore"
"github.com/conformal/btcwire"
)
// RescanMsg is the interface type for messages sent to the
// RescanManager's message channel.
type RescanMsg interface {
ImplementsRescanMsg()
}
// RescanStartedMsg reports the job being processed for a new
// rescan.
type RescanStartedMsg RescanJob
// ImplementsRescanMsg is implemented to satisify the RescanMsg
// interface.
func (r *RescanStartedMsg) ImplementsRescanMsg() {}
// RescanProgressMsg reports the current progress made by a rescan
// for a set of account's addresses.
// RescanProgressMsg reports the current progress made by a rescan for a
// set of wallet addresses.
type RescanProgressMsg struct {
Addresses map[*Account][]btcutil.Address
Height int32
Addresses []btcutil.Address
Notification *chain.RescanProgress
}
// ImplementsRescanMsg is implemented to satisify the RescanMsg
// interface.
func (r *RescanProgressMsg) ImplementsRescanMsg() {}
// RescanFinishedMsg reports the set of account's addresses of a
// possibly-finished rescan, or an error if the rescan failed.
// RescanFinishedMsg reports the addresses that were rescanned when a
// rescanfinished message was received rescanning a batch of addresses.
type RescanFinishedMsg struct {
Addresses map[*Account][]btcutil.Address
Error error
Addresses []btcutil.Address
Notification *chain.RescanFinished
WasInitialSync bool
}
// ImplementsRescanMsg is implemented to satisify the RescanMsg
// interface.
func (r *RescanFinishedMsg) ImplementsRescanMsg() {}
// RescanManager manages a set of current and to be processed account's
// addresses, batching waiting jobs together to minimize the total time
// needed to rescan many separate jobs. Rescan requests are processed
// one at a time, and the next batch does not run until the current
// has finished.
type RescanManager struct {
addJob chan *RescanJob
sendJob chan *RescanJob
status chan interface{} // rescanProgress and rescanFinished
msgs chan RescanMsg
jobCompleteChan chan chan struct{}
wg sync.WaitGroup
quit chan struct{}
}
// NewRescanManager creates a new RescanManger. If msgChan is non-nil,
// rescan messages are sent to the channel for additional processing by
// the caller.
func NewRescanManager(msgChan chan RescanMsg) *RescanManager {
return &RescanManager{
addJob: make(chan *RescanJob, 1),
sendJob: make(chan *RescanJob, 1),
status: make(chan interface{}, 1),
msgs: msgChan,
jobCompleteChan: make(chan chan struct{}, 1),
quit: make(chan struct{}),
}
}
// Start starts the goroutines to run the RescanManager.
func (m *RescanManager) Start() {
m.wg.Add(2)
go m.jobHandler()
go m.rpcHandler()
}
func (m *RescanManager) Stop() {
close(m.quit)
}
func (m *RescanManager) WaitForShutdown() {
m.wg.Wait()
// RescanJob is a job to be processed by the RescanManager. The job includes
// a set of wallet addresses, a starting height to begin the rescan, and
// outpoints spendable by the addresses thought to be unspent. After the
// rescan completes, the error result of the rescan RPC is sent on the Err
// channel.
type RescanJob struct {
InitialSync bool
Addrs []btcutil.Address
OutPoints []*btcwire.OutPoint
BlockStamp keystore.BlockStamp
err chan error
}
// rescanBatch is a collection of one or more RescanJobs that were merged
// together before a rescan is performed.
type rescanBatch struct {
addrs map[*Account][]btcutil.Address
outpoints map[btcwire.OutPoint]struct{}
height int32
complete chan struct{}
initialSync bool
addrs []btcutil.Address
outpoints []*btcwire.OutPoint
bs keystore.BlockStamp
errChans []chan error
}
func newRescanBatch() *rescanBatch {
// SubmitRescan submits a RescanJob to the RescanManager. A channel is
// returned with the final error of the rescan. The channel is buffered
// and does not need to be read to prevent a deadlock.
func (w *Wallet) SubmitRescan(job *RescanJob) <-chan error {
errChan := make(chan error, 1)
job.err = errChan
w.rescanAddJob <- job
return errChan
}
// batch creates the rescanBatch for a single rescan job.
func (job *RescanJob) batch() *rescanBatch {
return &rescanBatch{
addrs: map[*Account][]btcutil.Address{},
outpoints: map[btcwire.OutPoint]struct{}{},
height: -1,
complete: make(chan struct{}),
}
}
func (b *rescanBatch) done() {
close(b.complete)
}
func (b *rescanBatch) empty() bool {
return len(b.addrs) == 0
}
func (b *rescanBatch) job() *RescanJob {
// Create slice of outpoints from the batch's set.
outpoints := make([]*btcwire.OutPoint, 0, len(b.outpoints))
for outpoint := range b.outpoints {
opCopy := outpoint
outpoints = append(outpoints, &opCopy)
}
return &RescanJob{
Addresses: b.addrs,
OutPoints: outpoints,
StartHeight: b.height,
initialSync: job.InitialSync,
addrs: job.Addrs,
outpoints: job.OutPoints,
bs: job.BlockStamp,
errChans: []chan error{job.err},
}
}
// merge merges the work from k into j, setting the starting height to
// the minimum of the two jobs. This method does not check for
// duplicate addresses or outpoints.
func (b *rescanBatch) merge(job *RescanJob) {
for acct, addr := range job.Addresses {
b.addrs[acct] = append(b.addrs[acct], addr...)
if job.InitialSync {
b.initialSync = true
}
for _, op := range job.OutPoints {
b.outpoints[*op] = struct{}{}
b.addrs = append(b.addrs, job.Addrs...)
b.outpoints = append(b.outpoints, job.OutPoints...)
if job.BlockStamp.Height < b.bs.Height {
b.bs = job.BlockStamp
}
if b.height == -1 || job.StartHeight < b.height {
b.height = job.StartHeight
b.errChans = append(b.errChans, job.err)
}
// done iterates through all error channels, duplicating sending the error
// to inform callers that the rescan finished (or could not complete due
// to an error).
func (b *rescanBatch) done(err error) {
for _, c := range b.errChans {
c <- err
}
}
// jobHandler runs the RescanManager's for-select loop to manage rescan jobs
// and dispatch requests.
func (m *RescanManager) jobHandler() {
curBatch := newRescanBatch()
nextBatch := newRescanBatch()
// rescanBatchHandler handles incoming rescan request, serializing rescan
// submissions, and possibly batching many waiting requests together so they
// can be handled by a single rescan after the current one completes.
func (w *Wallet) rescanBatchHandler() {
var curBatch, nextBatch *rescanBatch
out:
for {
select {
case job := <-m.addJob:
if curBatch.empty() {
case job := <-w.rescanAddJob:
if curBatch == nil {
// Set current batch as this job and send
// request.
curBatch.merge(job)
m.sendJob <- job
// Send the channel that is closed when the
// current batch completes.
m.jobCompleteChan <- curBatch.complete
// Notify listener of a newly-started rescan.
if m.msgs != nil {
m.msgs <- (*RescanStartedMsg)(job)
}
curBatch = job.batch()
w.rescanBatch <- curBatch
} else {
// Create next batch if it doesn't exist, or
// merge the job.
if nextBatch == nil {
nextBatch = job.batch()
} else {
// Add job to waiting batch.
nextBatch.merge(job)
// Send the channel that is closed when the
// waiting batch completes.
m.jobCompleteChan <- nextBatch.complete
}
}
case status := <-m.status:
switch s := status.(type) {
case rescanProgress:
if m.msgs != nil {
m.msgs <- &RescanProgressMsg{
case n := <-w.rescanNotifications:
switch n := n.(type) {
case *chain.RescanProgress:
w.rescanProgress <- &RescanProgressMsg{
Addresses: curBatch.addrs,
Height: int32(s),
}
Notification: n,
}
case rescanFinished:
if m.msgs != nil {
m.msgs <- &RescanFinishedMsg{
case *chain.RescanFinished:
if curBatch == nil {
log.Warnf("Received rescan finished " +
"notification but no rescan " +
"currently running")
continue
}
w.rescanFinished <- &RescanFinishedMsg{
Addresses: curBatch.addrs,
Error: s.error,
Notification: n,
WasInitialSync: curBatch.initialSync,
}
}
curBatch.done()
curBatch, nextBatch = nextBatch, newRescanBatch()
curBatch, nextBatch = nextBatch, nil
if !curBatch.empty() {
job := curBatch.job()
m.sendJob <- job
if m.msgs != nil {
m.msgs <- (*RescanStartedMsg)(job)
}
if curBatch != nil {
w.rescanBatch <- curBatch
}
default:
// Unexpected status message
panic(s)
// Unexpected message
panic(n)
}
case <-m.quit:
case <-w.quit:
break out
}
}
close(m.sendJob)
if m.msgs != nil {
close(m.msgs)
}
m.wg.Done()
close(w.rescanBatch)
w.wg.Done()
}
// rpcHandler reads jobs sent by the jobHandler and sends the rpc requests
// to perform the rescan. New jobs are not read until a rescan finishes.
// The jobHandler is notified when the processing the rescan finishes.
func (m *RescanManager) rpcHandler() {
for job := range m.sendJob {
var addrs []btcutil.Address
for _, accountAddrs := range job.Addresses {
addrs = append(addrs, accountAddrs...)
}
client, err := accessClient()
// rescanProgressHandler handles notifications for paritally and fully completed
// rescans by marking each rescanned address as partially or fully synced and
// writing the keystore back to disk.
func (w *Wallet) rescanProgressHandler() {
out:
for {
// These can't be processed out of order since both chans are
// unbuffured and are sent from same context (the batch
// handler).
select {
case msg := <-w.rescanProgress:
n := msg.Notification
log.Infof("Rescanned through block %v (height %d)",
n.Hash, n.Height)
// TODO(jrick): save partial syncs should also include
// the block hash.
for _, addr := range msg.Addresses {
err := w.KeyStore.SetSyncStatus(addr,
keystore.PartialSync(n.Height))
if err != nil {
log.Errorf("Error marking address %v "+
"partially synced: %v", addr, err)
}
}
w.KeyStore.MarkDirty()
err := w.KeyStore.WriteIfDirty()
if err != nil {
log.Errorf("Could not write partial rescan "+
"progress to keystore: %v", err)
}
case msg := <-w.rescanFinished:
n := msg.Notification
addrs := msg.Addresses
noun := pickNoun(len(addrs), "address", "addresses")
if msg.WasInitialSync {
w.Track()
w.ResendUnminedTxs()
bs := keystore.BlockStamp{
Hash: n.Hash,
Height: n.Height,
}
w.KeyStore.SetSyncedWith(&bs)
w.notifyConnectedBlock(bs)
// Mark wallet as synced to chain so connected
// and disconnected block notifications are
// processed.
close(w.chainSynced)
}
log.Infof("Finished rescan for %d %s (synced to block "+
"%s, height %d)", len(addrs), noun, n.Hash,
n.Height)
for _, addr := range addrs {
err := w.KeyStore.SetSyncStatus(addr,
keystore.FullSync{})
if err != nil {
log.Errorf("Error marking address %v "+
"fully synced: %v", addr, err)
}
}
w.KeyStore.MarkDirty()
err := w.KeyStore.WriteIfDirty()
if err != nil {
log.Errorf("Could not write finished rescan "+
"progress to keystore: %v", err)
}
case <-w.quit:
break out
}
}
w.wg.Done()
}
// rescanRPCHandler reads batch jobs sent by rescanBatchHandler and sends the
// RPC requests to perform a rescan. New jobs are not read until a rescan
// finishes.
func (w *Wallet) rescanRPCHandler() {
for batch := range w.rescanBatch {
// Log the newly-started rescan.
numAddrs := len(batch.addrs)
noun := pickNoun(numAddrs, "address", "addresses")
log.Infof("Started rescan from block %v (height %d) for %d %s",
batch.bs.Hash, batch.bs.Height, numAddrs, noun)
err := w.chainSvr.Rescan(batch.bs.Hash, batch.addrs,
batch.outpoints)
if err != nil {
log.Errorf("Rescan for %d %s failed: %v", numAddrs,
noun, err)
}
batch.done(err)
}
w.wg.Done()
}
// RescanActiveAddresses begins a rescan for all active addresses of a
// wallet. This is intended to be used to sync a wallet back up to the
// current best block in the main chain, and is considered an intial sync
// rescan.
func (w *Wallet) RescanActiveAddresses() (err error) {
// Determine the block necesary to start the rescan for all active
// addresses.
hash, height := w.KeyStore.SyncedTo()
if hash == nil {
// TODO: fix our "synced to block" handling (either in
// keystore or txstore, or elsewhere) so this *always*
// returns the block hash. Looking it up by height is
// asking for problems.
hash, err = w.chainSvr.GetBlockHash(int64(height))
if err != nil {
m.MarkFinished(rescanFinished{err})
return
}
err = client.Rescan(job.StartHeight, addrs, job.OutPoints)
}
actives := w.KeyStore.SortedActiveAddresses()
addrs := make([]btcutil.Address, len(actives))
for i, addr := range actives {
addrs[i] = addr.Address()
}
unspents, err := w.TxStore.UnspentOutputs()
if err != nil {
m.MarkFinished(rescanFinished{err})
return
}
outpoints := make([]*btcwire.OutPoint, len(unspents))
for i, output := range unspents {
outpoints[i] = output.OutPoint()
}
m.wg.Done()
}
// RescanJob is a job to be processed by the RescanManager. The job includes
// a set of account's addresses, a starting height to begin the rescan, and
// outpoints spendable by the addresses thought to be unspent.
type RescanJob struct {
Addresses map[*Account][]btcutil.Address
OutPoints []*btcwire.OutPoint
StartHeight int32
}
// Merge merges the work from k into j, setting the starting height to
// the minimum of the two jobs. This method does not check for
// duplicate addresses or outpoints.
func (j *RescanJob) Merge(k *RescanJob) {
for acct, addrs := range k.Addresses {
j.Addresses[acct] = append(j.Addresses[acct], addrs...)
job := &RescanJob{
InitialSync: true,
Addrs: addrs,
OutPoints: outpoints,
BlockStamp: keystore.BlockStamp{Hash: hash, Height: height},
}
for _, op := range k.OutPoints {
j.OutPoints = append(j.OutPoints, op)
}
if k.StartHeight < j.StartHeight {
j.StartHeight = k.StartHeight
}
}
// SubmitJob submits a RescanJob to the RescanManager. A channel is returned
// that is closed once the rescan request for the job completes.
func (m *RescanManager) SubmitJob(job *RescanJob) <-chan struct{} {
m.addJob <- job
return <-m.jobCompleteChan
}
// MarkProgress messages the RescanManager with the height of the block
// last processed by a running rescan.
func (m *RescanManager) MarkProgress(height rescanProgress) {
m.status <- height
}
// MarkFinished messages the RescanManager that the currently running rescan
// finished, or errored prematurely.
func (m *RescanManager) MarkFinished(finished rescanFinished) {
m.status <- finished
// Submit merged job and block until rescan completes.
return <-w.SubmitRescan(job)
}

565
rpcclient.go
View file

@ -1,565 +0,0 @@
/*
* Copyright (c) 2013, 2014 Conformal Systems LLC <info@conformal.com>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
package main
import (
"errors"
"fmt"
"sync"
"time"
"github.com/conformal/btcrpcclient"
"github.com/conformal/btcscript"
"github.com/conformal/btcutil"
"github.com/conformal/btcwallet/keystore"
"github.com/conformal/btcwallet/txstore"
"github.com/conformal/btcwire"
"github.com/conformal/btcws"
)
// InvalidNotificationError describes an error due to an invalid chain server
// notification and should be warned by wallet, but does not indicate an
// problem with the current wallet state.
type InvalidNotificationError struct {
error
}
var (
// MismatchingNetworks represents an error where a client connection
// to btcd cannot succeed due to btcwallet and btcd operating on
// different bitcoin networks.
ErrMismatchedNets = errors.New("mismatched networks")
)
const (
// maxConcurrentClientRequests is the maximum number of
// unhandled/running requests that the server will run for a websocket
// client at a time. Beyond this limit, additional request reads will
// block until a running request handler finishes. This limit exists to
// prevent a single connection from causing a denial of service attack
// with an unnecessarily large number of requests.
maxConcurrentClientRequests = 20
)
type blockSummary struct {
hash *btcwire.ShaHash
height int32
}
type acceptedTx struct {
tx *btcutil.Tx
block *btcws.BlockDetails // nil if unmined
}
// Notification types. These are defined here and processed from from reading
// a notificationChan to avoid handling these notifications directly in
// btcrpcclient callbacks, which isn't very go-like and doesn't allow
// blocking client calls.
type (
// Container type for any notification.
notification interface {
handleNotification(*rpcClient) error
}
blockConnected blockSummary
blockDisconnected blockSummary
recvTx acceptedTx
redeemingTx acceptedTx
rescanFinished struct {
error
}
rescanProgress int32
)
func (n blockConnected) handleNotification(c *rpcClient) error {
// Update the blockstamp for the newly-connected block.
bs := keystore.BlockStamp{
Height: n.height,
Hash: *n.hash,
}
c.mtx.Lock()
c.blockStamp = bs
c.mtx.Unlock()
AcctMgr.Grab()
AcctMgr.BlockNotify(&bs)
AcctMgr.Release()
// Pass notification to wallet clients too.
if server != nil {
// TODO: marshaling should be perfomred by the server, and
// sent only to client that have requested the notification.
marshaled, err := n.MarshalJSON()
// The parsed notification is expected to be marshalable.
if err != nil {
panic(err)
}
server.broadcasts <- marshaled
}
return nil
}
// MarshalJSON creates the JSON encoding of the chain notification to pass
// to any connected wallet clients. This should never error.
func (n blockConnected) MarshalJSON() ([]byte, error) {
nn := btcws.NewBlockConnectedNtfn(n.hash.String(), n.height)
return nn.MarshalJSON()
}
func (n blockDisconnected) handleNotification(c *rpcClient) error {
AcctMgr.Grab()
defer AcctMgr.Release()
// Rollback Utxo and Tx data stores.
if err := AcctMgr.Rollback(n.height, n.hash); err != nil {
return err
}
// Pass notification to wallet clients too.
if server != nil {
// TODO: marshaling should be perfomred by the server, and
// sent only to client that have requested the notification.
marshaled, err := n.MarshalJSON()
// A btcws.BlockDisconnectedNtfn is expected to marshal without error.
// If it does, it indicates that one of its struct fields is of a
// non-marshalable type.
if err != nil {
panic(err)
}
server.broadcasts <- marshaled
}
return nil
}
// MarshalJSON creates the JSON encoding of the chain notification to pass
// to any connected wallet clients. This should never error.
func (n blockDisconnected) MarshalJSON() ([]byte, error) {
nn := btcws.NewBlockDisconnectedNtfn(n.hash.String(), n.height)
return nn.MarshalJSON()
}
func parseBlock(block *btcws.BlockDetails) (*txstore.Block, int, error) {
if block == nil {
return nil, btcutil.TxIndexUnknown, nil
}
blksha, err := btcwire.NewShaHashFromStr(block.Hash)
if err != nil {
return nil, btcutil.TxIndexUnknown, err
}
b := &txstore.Block{
Height: block.Height,
Hash: *blksha,
Time: time.Unix(block.Time, 0),
}
return b, block.Index, nil
}
func (n recvTx) handleNotification(c *rpcClient) error {
block, txIdx, err := parseBlock(n.block)
if err != nil {
return InvalidNotificationError{err}
}
n.tx.SetIndex(txIdx)
bs, err := c.BlockStamp()
if err != nil {
return fmt.Errorf("cannot get current block: %v", err)
}
AcctMgr.Grab()
defer AcctMgr.Release()
// For every output, if it pays to a wallet address, insert the
// transaction into the store (possibly moving it from unconfirmed to
// confirmed), and add a credit record if one does not already exist.
var txr *txstore.TxRecord
txInserted := false
for i, txout := range n.tx.MsgTx().TxOut {
// Errors don't matter here. If addrs is nil, the range below
// does nothing.
_, addrs, _, _ := btcscript.ExtractPkScriptAddrs(txout.PkScript,
activeNet.Params)
for _, addr := range addrs {
a, err := AcctMgr.AccountByAddress(addr)
if err != nil {
continue // try next address, if any
}
if !txInserted {
txr, err = a.TxStore.InsertTx(n.tx, block)
if err != nil {
return err
}
txInserted = true
}
// Insert and notify websocket clients of the credit if it is
// not a duplicate, otherwise, check the next txout if the
// credit has already been inserted.
if txr.HasCredit(i) {
break
}
cred, err := txr.AddCredit(uint32(i), false)
if err != nil {
return err
}
AcctMgr.ds.ScheduleTxStoreWrite(a)
ltr, err := cred.ToJSON(a.KeyStore.Name(), bs.Height,
a.KeyStore.Net())
if err != nil {
return err
}
server.NotifyNewTxDetails(a.KeyStore.Name(), ltr)
break // check whether next txout is a wallet txout
}
}
server.NotifyBalances()
return nil
}
func (n redeemingTx) handleNotification(c *rpcClient) error {
block, txIdx, err := parseBlock(n.block)
if err != nil {
return InvalidNotificationError{err}
}
n.tx.SetIndex(txIdx)
AcctMgr.Grab()
err = AcctMgr.RecordSpendingTx(n.tx, block)
AcctMgr.Release()
return err
}
func (n rescanFinished) handleNotification(c *rpcClient) error {
AcctMgr.rm.MarkFinished(n)
return nil
}
func (n rescanProgress) handleNotification(c *rpcClient) error {
AcctMgr.rm.MarkProgress(n)
return nil
}
type rpcClient struct {
*btcrpcclient.Client // client to btcd
mtx sync.Mutex
blockStamp keystore.BlockStamp
enqueueNotification chan notification
dequeueNotification chan notification
quit chan struct{}
wg sync.WaitGroup
}
func newRPCClient(certs []byte) (*rpcClient, error) {
client := rpcClient{
blockStamp: keystore.BlockStamp{
Height: int32(btcutil.BlockHeightUnknown),
},
enqueueNotification: make(chan notification),
dequeueNotification: make(chan notification),
quit: make(chan struct{}),
}
initializedClient := make(chan struct{})
ntfnCallbacks := btcrpcclient.NotificationHandlers{
OnClientConnected: func() {
log.Info("Established connection to btcd")
<-initializedClient
// nil client to broadcast to all connected clients
server.NotifyConnectionStatus(nil)
err := client.Handshake()
if err != nil {
log.Errorf("Cannot complete handshake: %v", err)
client.Stop()
}
},
OnBlockConnected: client.onBlockConnected,
OnBlockDisconnected: client.onBlockDisconnected,
OnRecvTx: client.onRecvTx,
OnRedeemingTx: client.onRedeemingTx,
OnRescanFinished: client.onRescanFinished,
OnRescanProgress: client.onRescanProgress,
}
conf := btcrpcclient.ConnConfig{
Host: cfg.RPCConnect,
Endpoint: "ws",
User: cfg.BtcdUsername,
Pass: cfg.BtcdPassword,
Certificates: certs,
}
c, err := btcrpcclient.New(&conf, &ntfnCallbacks)
if err != nil {
return nil, err
}
client.Client = c
close(initializedClient)
return &client, nil
}
func (c *rpcClient) Start() {
c.wg.Add(2)
go c.notificationQueue()
go c.handleNotifications()
}
func (c *rpcClient) Stop() {
log.Warn("Disconnecting chain server client connection")
c.Client.Shutdown()
select {
case <-c.quit:
default:
close(c.quit)
close(c.enqueueNotification)
}
}
func (c *rpcClient) WaitForShutdown() {
c.Client.WaitForShutdown()
c.wg.Wait()
}
func (c *rpcClient) onBlockConnected(hash *btcwire.ShaHash, height int32) {
c.enqueueNotification <- (blockConnected)(blockSummary{hash, height})
}
func (c *rpcClient) onBlockDisconnected(hash *btcwire.ShaHash, height int32) {
c.enqueueNotification <- (blockDisconnected)(blockSummary{hash, height})
}
func (c *rpcClient) onRecvTx(tx *btcutil.Tx, block *btcws.BlockDetails) {
c.enqueueNotification <- recvTx{tx, block}
}
func (c *rpcClient) onRedeemingTx(tx *btcutil.Tx, block *btcws.BlockDetails) {
c.enqueueNotification <- redeemingTx{tx, block}
}
func (c *rpcClient) onRescanProgress(height int32) {
c.enqueueNotification <- rescanProgress(height)
}
func (c *rpcClient) onRescanFinished(height int32) {
c.enqueueNotification <- rescanFinished{error: nil}
}
func (c *rpcClient) notificationQueue() {
// TODO: Rather than leaving this as an unbounded queue for all types of
// notifications, try dropping ones where a later enqueued notification
// can fully invalidate one waiting to be processed. For example,
// blockconnected notifications for greater block heights can remove the
// need to process earlier blockconnected notifications still waiting
// here.
var q []notification
enqueue := c.enqueueNotification
var dequeue chan notification
var next notification
out:
for {
select {
case n, ok := <-enqueue:
if !ok {
// If no notifications are queued for handling,
// the queue is finished.
if len(q) == 0 {
break out
}
// nil channel so no more reads can occur.
enqueue = nil
continue
}
if len(q) == 0 {
next = n
dequeue = c.dequeueNotification
}
q = append(q, n)
case dequeue <- next:
q[0] = nil
q = q[1:]
if len(q) != 0 {
next = q[0]
} else {
// If no more notifications can be enqueued, the
// queue is finished.
if enqueue == nil {
break out
}
dequeue = nil
}
}
}
close(c.dequeueNotification)
c.wg.Done()
}
func (c *rpcClient) handleNotifications() {
for n := range c.dequeueNotification {
err := n.handleNotification(c)
if err != nil {
switch e := err.(type) {
case InvalidNotificationError:
log.Warnf("Ignoring invalid notification: %v", e)
default:
log.Errorf("Cannot handle notification: %v", e)
}
}
}
c.wg.Done()
}
// BlockStamp returns (as a blockstamp) the height and hash of the last seen
// block from the RPC client. If no blocks have been seen (the height is -1),
// the chain server is queried for the block and the result is saved for future
// calls, or an error is returned if the RPC is unsuccessful.
func (c *rpcClient) BlockStamp() (keystore.BlockStamp, error) {
c.mtx.Lock()
defer c.mtx.Unlock()
if c.blockStamp.Height != int32(btcutil.BlockHeightUnknown) {
return c.blockStamp, nil
}
hash, height, err := c.GetBestBlock()
if err != nil {
return keystore.BlockStamp{}, err
}
bs := keystore.BlockStamp{
Hash: *hash,
Height: height,
}
c.blockStamp = bs
return bs, nil
}
// Handshake first checks that the websocket connection between btcwallet and
// btcd is valid, that is, that there are no mismatching settings between
// the two processes (such as running on different Bitcoin networks). If the
// sanity checks pass, all wallets are set to be tracked against chain
// notifications from this btcd connection.
//
// TODO(jrick): Track and Rescan commands should be replaced with a
// single TrackSince function (or similar) which requests address
// notifications and performs the rescan since some block height.
func (c *rpcClient) Handshake() error {
net, err := c.GetCurrentNet()
if err != nil {
return err
}
if net != activeNet.Net {
return ErrMismatchedNets
}
// Request notifications for connected and disconnected blocks.
if err := c.NotifyBlocks(); err != nil {
return err
}
// Get current best block. If this is before than the oldest
// saved block hash, assume that this btcd instance is not yet
// synced up to a previous btcd that was last used with this
// wallet.
bs, err := c.BlockStamp()
if err != nil {
return err
}
if server != nil {
server.NotifyNewBlockChainHeight(&bs)
server.NotifyBalances()
}
// Get default account. Only the default account is used to
// track recently-seen blocks.
a, err := AcctMgr.Account("")
if err != nil {
// No account yet is not a handshake error, but means our
// handshake is done.
return nil
}
// TODO(jrick): if height is less than the earliest-saved block
// height, should probably wait for btcd to catch up.
// Check that there was not any reorgs done since last connection.
// If so, rollback and rescan to catch up.
it := a.KeyStore.NewIterateRecentBlocks()
for cont := it != nil; cont; cont = it.Prev() {
bs := it.BlockStamp()
log.Debugf("Checking for previous saved block with height %v hash %v",
bs.Height, bs.Hash)
if _, err := c.GetBlock(&bs.Hash); err != nil {
continue
}
log.Debug("Found matching block.")
// If we had to go back to any previous blocks (it.Next
// returns true), then rollback the next and all child blocks.
// This rollback is done here instead of in the blockMissing
// check above for each removed block because Rollback will
// try to write new tx and utxo files on each rollback.
if it.Next() {
bs := it.BlockStamp()
err := AcctMgr.Rollback(bs.Height, &bs.Hash)
if err != nil {
return err
}
}
// Set default account to be marked in sync with the current
// blockstamp. This invalidates the iterator.
a.KeyStore.SetSyncedWith(bs)
// Begin tracking wallets against this btcd instance.
AcctMgr.Track()
if err := AcctMgr.RescanActiveAddresses(nil); err != nil {
return err
}
// TODO: Only begin tracking new unspent outputs as a result
// of the rescan. This is also pretty racy, as a new block
// could arrive between rescan and by the time the new outpoint
// is added to btcd's websocket's unspent output set.
AcctMgr.Track()
// (Re)send any unmined transactions to btcd in case of a btcd restart.
AcctMgr.ResendUnminedTxs()
// Get current blockchain height and best block hash.
return nil
}
// Iterator was invalid (wallet has never been synced) or there was a
// huge chain fork + reorg (more than 20 blocks).
AcctMgr.Track()
if err := AcctMgr.RescanActiveAddresses(&bs); err != nil {
return err
}
// TODO: only begin tracking new unspent outputs as a result of the
// rescan. This is also racy (see comment for second Track above).
AcctMgr.Track()
AcctMgr.ResendUnminedTxs()
return nil
}

2345
rpcserver.go
View file

File diff suppressed because it is too large Load diff

48
txstore/json.go
View file

@ -29,16 +29,19 @@ import (
func (t *TxRecord) ToJSON(account string, chainHeight int32,
net *btcnet.Params) ([]btcjson.ListTransactionsResult, error) {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
results := []btcjson.ListTransactionsResult{}
if d, err := t.Debits(); err == nil {
r, err := d.ToJSON(account, chainHeight, net)
r, err := d.toJSON(account, chainHeight, net)
if err != nil {
return nil, err
}
results = r
}
for _, c := range t.Credits() {
r, err := c.ToJSON(account, chainHeight, net)
r, err := c.toJSON(account, chainHeight, net)
if err != nil {
return nil, err
}
@ -49,7 +52,16 @@ func (t *TxRecord) ToJSON(account string, chainHeight int32,
// ToJSON returns a slice of objects that may be marshaled as a JSON array
// of JSON objects for a listtransactions RPC reply.
func (d *Debits) ToJSON(account string, chainHeight int32,
func (d Debits) ToJSON(account string, chainHeight int32,
net *btcnet.Params) ([]btcjson.ListTransactionsResult, error) {
d.s.mtx.RLock()
defer d.s.mtx.RUnlock()
return d.toJSON(account, chainHeight, net)
}
func (d Debits) toJSON(account string, chainHeight int32,
net *btcnet.Params) ([]btcjson.ListTransactionsResult, error) {
msgTx := d.Tx().MsgTx()
@ -82,7 +94,7 @@ func (d *Debits) ToJSON(account string, chainHeight int32,
result.BlockHash = b.Hash.String()
result.BlockIndex = int64(d.Tx().Index())
result.BlockTime = b.Time.Unix()
result.Confirmations = int64(d.Confirmations(chainHeight))
result.Confirmations = int64(confirms(d.BlockHeight, chainHeight))
}
reply = append(reply, result)
}
@ -102,11 +114,18 @@ const (
CreditImmature
)
// Category returns the category of the credit. The passed block chain height is
// category returns the category of the credit. The passed block chain height is
// used to distinguish immature from mature coinbase outputs.
func (c *Credit) Category(chainHeight int32) CreditCategory {
if c.IsCoinbase() {
if c.Confirmed(btcchain.CoinbaseMaturity, chainHeight) {
c.s.mtx.RLock()
defer c.s.mtx.RUnlock()
return c.category(chainHeight)
}
func (c *Credit) category(chainHeight int32) CreditCategory {
if c.isCoinbase() {
if confirmed(btcchain.CoinbaseMaturity, c.BlockHeight, chainHeight) {
return CreditGenerate
}
return CreditImmature
@ -132,7 +151,16 @@ func (c CreditCategory) String() string {
// ToJSON returns a slice of objects that may be marshaled as a JSON array
// of JSON objects for a listtransactions RPC reply.
func (c *Credit) ToJSON(account string, chainHeight int32,
func (c Credit) ToJSON(account string, chainHeight int32,
net *btcnet.Params) (btcjson.ListTransactionsResult, error) {
c.s.mtx.RLock()
defer c.s.mtx.RUnlock()
return c.toJSON(account, chainHeight, net)
}
func (c Credit) toJSON(account string, chainHeight int32,
net *btcnet.Params) (btcjson.ListTransactionsResult, error) {
msgTx := c.Tx().MsgTx()
@ -146,7 +174,7 @@ func (c *Credit) ToJSON(account string, chainHeight int32,
result := btcjson.ListTransactionsResult{
Account: account,
Category: c.Category(chainHeight).String(),
Category: c.category(chainHeight).String(),
Address: address,
Amount: btcutil.Amount(txout.Value).ToUnit(btcutil.AmountBTC),
TxID: c.Tx().Sha().String(),
@ -163,7 +191,7 @@ func (c *Credit) ToJSON(account string, chainHeight int32,
result.BlockHash = b.Hash.String()
result.BlockIndex = int64(c.Tx().Index())
result.BlockTime = b.Time.Unix()
result.Confirmations = int64(c.Confirmations(chainHeight))
result.Confirmations = int64(confirms(c.BlockHeight, chainHeight))
}
return result, nil

145
txstore/notifications.go Normal file
View file

@ -0,0 +1,145 @@
/*
* Copyright (c) 2014 Conformal Systems LLC <info@conformal.com>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
package txstore
import (
"errors"
)
// ErrDuplicateListen is returned for any attempts to listen for the same
// notification more than once. If callers must pass along a notifiation to
// multiple places, they must broadcast it themself.
var ErrDuplicateListen = errors.New("duplicate listen")
type noopLocker struct{}
func (noopLocker) Lock() {}
func (noopLocker) Unlock() {}
func (s *Store) updateNotificationLock() {
switch {
case s.newCredit == nil:
fallthrough
case s.newDebits == nil:
fallthrough
case s.minedCredit == nil:
fallthrough
case s.minedDebits == nil:
return
}
s.notificationLock = noopLocker{}
}
// ListenNewCredits returns a channel that passes all Credits that are newly
// added to the transaction store. The channel must be read, or other
// transaction store methods will block.
//
// If this is called twice, ErrDuplicateListen is returned.
func (s *Store) ListenNewCredits() (<-chan Credit, error) {
s.notificationLock.Lock()
defer s.notificationLock.Unlock()
if s.newCredit != nil {
return nil, ErrDuplicateListen
}
s.newCredit = make(chan Credit)
s.updateNotificationLock()
return s.newCredit, nil
}
// ListenNewDebits returns a channel that passes all Debits that are newly
// added to the transaction store. The channel must be read, or other
// transaction store methods will block.
//
// If this is called twice, ErrDuplicateListen is returned.
func (s *Store) ListenNewDebits() (<-chan Debits, error) {
s.notificationLock.Lock()
defer s.notificationLock.Unlock()
if s.newDebits != nil {
return nil, ErrDuplicateListen
}
s.newDebits = make(chan Debits)
s.updateNotificationLock()
return s.newDebits, nil
}
// ListenMinedCredits returns a channel that passes all that are moved
// from unconfirmed to a newly attached block. The channel must be read, or
// other transaction store methods will block.
//
// If this is called twice, ErrDuplicateListen is returned.
func (s *Store) ListenMinedCredits() (<-chan Credit, error) {
s.notificationLock.Lock()
defer s.notificationLock.Unlock()
if s.minedCredit != nil {
return nil, ErrDuplicateListen
}
s.minedCredit = make(chan Credit)
s.updateNotificationLock()
return s.minedCredit, nil
}
// ListenMinedDebits returns a channel that passes all Debits that are moved
// from unconfirmed to a newly attached block. The channel must be read, or
// other transaction store methods will block.
//
// If this is called twice, ErrDuplicateListen is returned.
func (s *Store) ListenMinedDebits() (<-chan Debits, error) {
s.notificationLock.Lock()
defer s.notificationLock.Unlock()
if s.minedDebits != nil {
return nil, ErrDuplicateListen
}
s.minedDebits = make(chan Debits)
s.updateNotificationLock()
return s.minedDebits, nil
}
func (s *Store) notifyNewCredit(c Credit) {
s.notificationLock.Lock()
if s.newCredit != nil {
s.newCredit <- c
}
s.notificationLock.Unlock()
}
func (s *Store) notifyNewDebits(d Debits) {
s.notificationLock.Lock()
if s.newDebits != nil {
s.newDebits <- d
}
s.notificationLock.Unlock()
}
func (s *Store) notifyMinedCredit(c Credit) {
s.notificationLock.Lock()
if s.minedCredit != nil {
s.minedCredit <- c
}
s.notificationLock.Unlock()
}
func (s *Store) notifyMinedDebits(d Debits) {
s.notificationLock.Lock()
if s.minedDebits != nil {
s.minedDebits <- d
}
s.notificationLock.Unlock()
}

88
txstore/serialization.go
View file

@ -22,12 +22,20 @@ import (
"errors"
"fmt"
"io"
"io/ioutil"
"os"
"path/filepath"
"time"
"github.com/conformal/btcutil"
"github.com/conformal/btcwallet/rename"
"github.com/conformal/btcwire"
)
// filename is the name of the file typically used to save a transaction
// store on disk.
const filename = "tx.bin"
// All Store versions (both old and current).
const (
versFirst uint32 = iota
@ -59,6 +67,8 @@ var byteOrder = binary.LittleEndian
// ReadFrom satisifies the io.ReaderFrom interface by deserializing a
// transaction store from an io.Reader.
func (s *Store) ReadFrom(r io.Reader) (int64, error) {
// Don't bother locking this. The mutex gets overwritten anyways.
var buf [4]byte
uint32Bytes := buf[:4]
@ -76,7 +86,7 @@ func (s *Store) ReadFrom(r io.Reader) (int64, error) {
}
// Reset store.
*s = *New()
*s = *New(s.path)
// Read block structures. Begin by reading the total number of block
// structures to be read, and then iterate that many times to read
@ -144,6 +154,13 @@ func (s *Store) ReadFrom(r io.Reader) (int64, error) {
// WriteTo satisifies the io.WriterTo interface by serializing a transaction
// store to an io.Writer.
func (s *Store) WriteTo(w io.Writer) (int64, error) {
s.mtx.RLock()
defer s.mtx.RUnlock()
return s.writeTo(w)
}
func (s *Store) writeTo(w io.Writer) (int64, error) {
var buf [4]byte
uint32Bytes := buf[:4]
@ -1128,3 +1145,72 @@ func (u *unconfirmedStore) WriteTo(w io.Writer) (int64, error) {
return n64, nil
}
// TODO: set this automatically.
func (s *Store) MarkDirty() {
s.mtx.Lock()
defer s.mtx.Unlock()
s.dirty = true
}
func (s *Store) WriteIfDirty() error {
s.mtx.RLock()
if !s.dirty {
s.mtx.RUnlock()
return nil
}
// TempFile creates the file 0600, so no need to chmod it.
fi, err := ioutil.TempFile(s.dir, s.file)
if err != nil {
s.mtx.RUnlock()
return err
}
fiPath := fi.Name()
_, err = s.writeTo(fi)
if err != nil {
s.mtx.RUnlock()
fi.Close()
return err
}
err = fi.Sync()
if err != nil {
s.mtx.RUnlock()
fi.Close()
return err
}
fi.Close()
err = rename.Atomic(fiPath, s.path)
s.mtx.RUnlock()
if err == nil {
s.mtx.Lock()
s.dirty = false
s.mtx.Unlock()
}
return err
}
// OpenDir opens a new transaction store from the specified directory.
// If the file does not exist, the error from the os package will be
// returned, and can be checked with os.IsNotExist to differentiate missing
// file errors from others (including deserialization).
func OpenDir(dir string) (*Store, error) {
path := filepath.Join(dir, filename)
fi, err := os.OpenFile(path, os.O_RDONLY, 0)
if err != nil {
return nil, err
}
defer fi.Close()
store := new(Store)
_, err = store.ReadFrom(fi)
if err != nil {
return nil, err
}
store.path = path
store.dir = dir
return store, nil
}

204
txstore/tx.go
View file

@ -19,7 +19,9 @@ package txstore
import (
"errors"
"fmt"
"path/filepath"
"sort"
"sync"
"time"
"github.com/conformal/btcchain"
@ -155,6 +157,15 @@ type blockAmounts struct {
// Store implements a transaction store for storing and managing wallet
// transactions.
type Store struct {
// TODO: Use atomic operations for dirty so the reader lock
// doesn't need to be grabbed.
dirty bool
path string
dir string
file string
mtx sync.RWMutex
// blocks holds wallet transaction records for each block they appear
// in. This is sorted by block height in increasing order. A separate
// map is included to lookup indexes for blocks at some height.
@ -166,6 +177,15 @@ type Store struct {
// unconfirmed holds a collection of wallet transactions that have not
// been mined into a block yet.
unconfirmed unconfirmedStore
// Channels to notify callers of changes to the transaction store.
// These are only created when a caller calls the appropiate
// registration method.
newCredit chan Credit
newDebits chan Debits
minedCredit chan Credit
minedDebits chan Debits
notificationLock sync.Locker
}
// blockTxCollection holds a collection of wallet transactions from exactly one
@ -254,8 +274,11 @@ type credit struct {
}
// New allocates and initializes a new transaction store.
func New() *Store {
func New(dir string) *Store {
return &Store{
path: filepath.Join(dir, filename),
dir: dir,
file: filename,
blockIndexes: map[int32]uint32{},
unspent: map[btcwire.OutPoint]BlockTxKey{},
unconfirmed: unconfirmedStore{
@ -265,6 +288,7 @@ func New() *Store {
spentUnconfirmed: map[btcwire.OutPoint]*txRecord{},
previousOutpoints: map[btcwire.OutPoint]*txRecord{},
},
notificationLock: new(sync.Mutex),
}
}
@ -443,6 +467,10 @@ func (s *Store) moveMinedTx(r *txRecord, block *Block) error {
// debits should already be non-nil
r.debits.spends = append(r.debits.spends, prev)
}
if r.debits != nil {
d := Debits{&TxRecord{key, r, s}}
s.notifyMinedDebits(d)
}
// For each credit in r, if the credit is spent by another unconfirmed
// transaction, move the spending transaction from spentUnconfirmed
@ -475,6 +503,9 @@ func (s *Store) moveMinedTx(r *txRecord, block *Block) error {
value := r.Tx().MsgTx().TxOut[i].Value
b.amountDeltas.Spendable += btcutil.Amount(value)
}
c := Credit{&TxRecord{key, r, s}, op.Index}
s.notifyMinedCredit(c)
}
// If this moved transaction debits from any previous credits, decrement
@ -496,6 +527,9 @@ func (s *Store) moveMinedTx(r *txRecord, block *Block) error {
// The transaction record is returned. Credits and debits may be added to the
// transaction by calling methods on the TxRecord.
func (s *Store) InsertTx(tx *btcutil.Tx, block *Block) (*TxRecord, error) {
s.mtx.Lock()
defer s.mtx.Unlock()
// The receive time will be the earlier of now and the block time
// (if any).
received := time.Now()
@ -565,12 +599,18 @@ func (s *Store) InsertTx(tx *btcutil.Tx, block *Block) (*TxRecord, error) {
// Received returns the earliest known time the transaction was received by.
func (t *TxRecord) Received() time.Time {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
return t.received
}
// Block returns the block details for a transaction. If the transaction is
// unmined, both the block and returned error are nil.
func (t *TxRecord) Block() (*Block, error) {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
coll, err := t.s.lookupBlock(t.BlockHeight)
if err != nil {
if err == MissingBlockError(-1) {
@ -581,56 +621,37 @@ func (t *TxRecord) Block() (*Block, error) {
return &coll.Block, nil
}
// AddDebits marks a transaction record as having debited from all them transaction
// credits in the spent slice. If spent is nil, the previous debits will be found,
// however this is an expensive lookup and should be avoided if possible.
func (t *TxRecord) AddDebits(spent []Credit) (Debits, error) {
// AddDebits marks a transaction record as having debited from previous wallet
// credits.
func (t *TxRecord) AddDebits() (Debits, error) {
t.s.mtx.Lock()
defer t.s.mtx.Unlock()
if t.debits == nil {
// Find now-spent credits if no debits have been previously set
// and none were passed in by the caller.
if len(spent) == 0 {
foundSpent, err := t.s.findPreviousCredits(t.Tx())
spent, err := t.s.findPreviousCredits(t.Tx())
if err != nil {
return Debits{}, err
}
spent = foundSpent
}
debitAmount, err := t.s.markOutputsSpent(spent, t)
if err != nil {
return Debits{}, err
}
t.debits = &debits{amount: debitAmount}
prevOutputKeys := make([]BlockOutputKey, len(spent))
for i, c := range spent {
prevOutputKeys[i] = c.outputKey()
}
t.debits = &debits{amount: debitAmount, spends: prevOutputKeys}
log.Debugf("Transaction %v spends %d previously-unspent "+
"%s totaling %v", t.tx.Sha(), len(spent),
pickNoun(len(spent), "output", "outputs"), debitAmount)
}
switch t.BlockHeight {
case -1: // unconfimred
for _, c := range spent {
op := c.OutPoint()
switch c.BlockHeight {
case -1: // unconfirmed
t.s.unconfirmed.spentUnconfirmed[*op] = t.txRecord
default:
key := c.outputKey()
t.s.unconfirmed.spentBlockOutPointKeys[*op] = key
t.s.unconfirmed.spentBlockOutPoints[key] = t.txRecord
}
}
default:
if t.debits.spends == nil {
prevOutputKeys := make([]BlockOutputKey, len(spent))
for i, c := range spent {
prevOutputKeys[i] = c.outputKey()
}
t.txRecord.debits.spends = prevOutputKeys
}
}
return Debits{t}, nil
d := Debits{t}
t.s.notifyNewDebits(d)
return d, nil
}
// findPreviousCredits searches for all unspent credits that make up the inputs
@ -699,10 +720,18 @@ func (s *Store) findPreviousCredits(tx *btcutil.Tx) ([]Credit, error) {
func (s *Store) markOutputsSpent(spent []Credit, t *TxRecord) (btcutil.Amount, error) {
var a btcutil.Amount
for _, prev := range spent {
op := prev.OutPoint()
op := prev.outPoint()
switch prev.BlockHeight {
case -1: // unconfirmed
s.unconfirmed.spentUnconfirmed[*op] = t.txRecord
op := prev.outPoint()
switch prev.BlockHeight {
case -1: // unconfirmed
t.s.unconfirmed.spentUnconfirmed[*op] = t.txRecord
default:
key := prev.outputKey()
t.s.unconfirmed.spentBlockOutPointKeys[*op] = key
t.s.unconfirmed.spentBlockOutPoints[key] = t.txRecord
}
default:
// Update spent info.
@ -722,7 +751,7 @@ func (s *Store) markOutputsSpent(spent []Credit, t *TxRecord) (btcutil.Amount, e
}
// Increment total debited amount.
a += prev.Amount()
a += prev.amount()
}
}
@ -760,6 +789,9 @@ func (r *txRecord) setCredit(index uint32, change bool, tx *btcutil.Tx) error {
// spendable by wallet. The output is added unspent, and is marked spent
// when a new transaction spending the output is inserted into the store.
func (t *TxRecord) AddCredit(index uint32, change bool) (Credit, error) {
t.s.mtx.Lock()
defer t.s.mtx.Unlock()
if len(t.tx.MsgTx().TxOut) <= int(index) {
return Credit{}, errors.New("transaction output does not exist")
}
@ -794,12 +826,17 @@ func (t *TxRecord) AddCredit(index uint32, change bool) (Credit, error) {
}
}
return Credit{t, index}, nil
c := Credit{t, index}
t.s.notifyNewCredit(c)
return c, nil
}
// Rollback removes all blocks at height onwards, moving any transactions within
// each block to the unconfirmed pool.
func (s *Store) Rollback(height int32) error {
s.mtx.Lock()
defer s.mtx.Unlock()
i := len(s.blocks)
for i != 0 && s.blocks[i-1].Height >= height {
i--
@ -944,6 +981,9 @@ func (r *txRecord) swapDebits(previous, current BlockOutputKey) error {
// transactions which debit from previous outputs and are not known to have
// been mined in a block.
func (s *Store) UnminedDebitTxs() []*btcutil.Tx {
s.mtx.RLock()
defer s.mtx.RUnlock()
unmined := make([]*btcutil.Tx, 0, len(s.unconfirmed.txs))
for _, r := range s.unconfirmed.spentBlockOutPoints {
unmined = append(unmined, r.Tx())
@ -1042,6 +1082,13 @@ func (s *Store) removeConflict(r *txRecord) error {
// UnspentOutputs returns all unspent received transaction outputs.
// The order is undefined.
func (s *Store) UnspentOutputs() ([]Credit, error) {
s.mtx.RLock()
defer s.mtx.RUnlock()
return s.unspentOutputs()
}
func (s *Store) unspentOutputs() ([]Credit, error) {
type createdCredit struct {
credit Credit
err error
@ -1139,7 +1186,10 @@ func (s creditSlice) Swap(i, j int) {
// index in increasing order. Credits (outputs) from the same transaction
// are sorted by output index in increasing order.
func (s *Store) SortedUnspentOutputs() ([]Credit, error) {
unspent, err := s.UnspentOutputs()
s.mtx.RLock()
defer s.mtx.RUnlock()
unspent, err := s.unspentOutputs()
if err != nil {
return []Credit{}, err
}
@ -1170,6 +1220,13 @@ func confirms(txHeight, curHeight int32) int32 {
// at a current chain height of curHeight. Coinbase outputs are only included
// in the balance if maturity has been reached.
func (s *Store) Balance(minConf int, chainHeight int32) (btcutil.Amount, error) {
s.mtx.RLock()
defer s.mtx.RUnlock()
return s.balance(minConf, chainHeight)
}
func (s *Store) balance(minConf int, chainHeight int32) (btcutil.Amount, error) {
var bal btcutil.Amount
// Shadow these functions to avoid repeating arguments unnecesarily.
@ -1244,6 +1301,9 @@ func (s *Store) Balance(minConf int, chainHeight int32) (btcutil.Amount, error)
// saved by the store. This is sorted first by block height in increasing
// order, and then by transaction index for each tx in a block.
func (s *Store) Records() (records []*TxRecord) {
s.mtx.RLock()
defer s.mtx.RUnlock()
for _, b := range s.blocks {
for _, r := range b.txs {
key := BlockTxKey{r.tx.Index(), b.Block.Height}
@ -1275,6 +1335,9 @@ func (r byReceiveDate) Swap(i, j int) { r[i], r[j] = r[j], r[i] }
// Debits returns the debit record for the transaction, or a non-nil error if
// the transaction does not debit from any previous transaction credits.
func (t *TxRecord) Debits() (Debits, error) {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
if t.debits == nil {
return Debits{}, errors.New("no debits")
}
@ -1284,6 +1347,9 @@ func (t *TxRecord) Debits() (Debits, error) {
// Credits returns all credit records for this transaction's outputs that are or
// were spendable by wallet.
func (t *TxRecord) Credits() []Credit {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
credits := make([]Credit, 0, len(t.credits))
for i, c := range t.credits {
if c != nil {
@ -1296,6 +1362,9 @@ func (t *TxRecord) Credits() []Credit {
// HasCredit returns whether the transaction output at the passed index is
// a wallet credit.
func (t *TxRecord) HasCredit(i int) bool {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
if len(t.credits) <= i {
return false
}
@ -1304,11 +1373,21 @@ func (t *TxRecord) HasCredit(i int) bool {
// InputAmount returns the total amount debited from previous credits.
func (d Debits) InputAmount() btcutil.Amount {
d.s.mtx.RLock()
defer d.s.mtx.RUnlock()
return d.txRecord.debits.amount
}
// OutputAmount returns the total amount of all outputs for a transaction.
func (t *TxRecord) OutputAmount(ignoreChange bool) btcutil.Amount {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
return t.outputAmount(ignoreChange)
}
func (t *TxRecord) outputAmount(ignoreChange bool) btcutil.Amount {
a := btcutil.Amount(0)
for i, txOut := range t.Tx().MsgTx().TxOut {
if ignoreChange {
@ -1325,7 +1404,7 @@ func (t *TxRecord) OutputAmount(ignoreChange bool) btcutil.Amount {
// Fee returns the difference between the debited amount and the total
// transaction output.
func (d Debits) Fee() btcutil.Amount {
return d.InputAmount() - d.OutputAmount(false)
return d.txRecord.debits.amount - d.outputAmount(false)
}
// Addresses parses the pubkey script, extracting all addresses for a
@ -1333,6 +1412,9 @@ func (d Debits) Fee() btcutil.Amount {
func (c Credit) Addresses(net *btcnet.Params) (btcscript.ScriptClass,
[]btcutil.Address, int, error) {
c.s.mtx.RLock()
defer c.s.mtx.RUnlock()
msgTx := c.Tx().MsgTx()
pkScript := msgTx.TxOut[c.OutputIndex].PkScript
return btcscript.ExtractPkScriptAddrs(pkScript, net)
@ -1340,28 +1422,51 @@ func (c Credit) Addresses(net *btcnet.Params) (btcscript.ScriptClass,
// Change returns whether the credit is the result of a change output.
func (c Credit) Change() bool {
c.s.mtx.RLock()
defer c.s.mtx.RUnlock()
return c.txRecord.credits[c.OutputIndex].change
}
// Confirmed returns whether a transaction has reached some target number of
// confirmations, given the current best chain height.
func (t *TxRecord) Confirmed(target int, chainHeight int32) bool {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
return confirmed(target, t.BlockHeight, chainHeight)
}
// Confirmations returns the total number of confirmations a transaction has
// reached, given the current best chain height.
func (t *TxRecord) Confirmations(chainHeight int32) int32 {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
return confirms(t.BlockHeight, chainHeight)
}
// IsCoinbase returns whether the transaction is a coinbase.
func (t *TxRecord) IsCoinbase() bool {
t.s.mtx.RLock()
defer t.s.mtx.RUnlock()
return t.isCoinbase()
}
func (t *TxRecord) isCoinbase() bool {
return t.BlockHeight != -1 && t.BlockIndex == 0
}
// Amount returns the amount credited to the account from a transaction output.
func (c Credit) Amount() btcutil.Amount {
c.s.mtx.RLock()
defer c.s.mtx.RUnlock()
return c.amount()
}
func (c Credit) amount() btcutil.Amount {
msgTx := c.Tx().MsgTx()
return btcutil.Amount(msgTx.TxOut[c.OutputIndex].Value)
}
@ -1369,6 +1474,13 @@ func (c Credit) Amount() btcutil.Amount {
// OutPoint returns the outpoint needed to include in a transaction input
// to spend this output.
func (c Credit) OutPoint() *btcwire.OutPoint {
c.s.mtx.RLock()
defer c.s.mtx.RUnlock()
return c.outPoint()
}
func (c Credit) outPoint() *btcwire.OutPoint {
return btcwire.NewOutPoint(c.Tx().Sha(), c.OutputIndex)
}
@ -1382,11 +1494,17 @@ func (c Credit) outputKey() BlockOutputKey {
// Spent returns whether the transaction output is currently spent or not.
func (c Credit) Spent() bool {
c.s.mtx.RLock()
defer c.s.mtx.RUnlock()
return c.txRecord.credits[c.OutputIndex].spentBy != nil
}
// TxOut returns the transaction output which this credit references.
func (c Credit) TxOut() *btcwire.TxOut {
c.s.mtx.RLock()
defer c.s.mtx.RUnlock()
return c.Tx().MsgTx().TxOut[c.OutputIndex]
}

14
txstore/tx_test.go
View file

@ -88,7 +88,7 @@ func TestInsertsCreditsDebitsRollbacks(t *testing.T) {
{
name: "new store",
f: func(_ *Store) (*Store, error) {
return New(), nil
return New("/tmp/tx.bin"), nil
},
bal: 0,
unc: 0,
@ -249,7 +249,7 @@ func TestInsertsCreditsDebitsRollbacks(t *testing.T) {
{
name: "insert unconfirmed debit",
f: func(s *Store) (*Store, error) {
prev, err := s.InsertTx(TstDoubleSpendTx, TstRecvTxBlockDetails)
_, err := s.InsertTx(TstDoubleSpendTx, TstRecvTxBlockDetails)
if err != nil {
return nil, err
}
@ -259,7 +259,7 @@ func TestInsertsCreditsDebitsRollbacks(t *testing.T) {
return nil, err
}
_, err = r.AddDebits(prev.Credits())
_, err = r.AddDebits()
if err != nil {
return nil, err
}
@ -280,7 +280,7 @@ func TestInsertsCreditsDebitsRollbacks(t *testing.T) {
{
name: "insert unconfirmed debit again",
f: func(s *Store) (*Store, error) {
prev, err := s.InsertTx(TstDoubleSpendTx, TstRecvTxBlockDetails)
_, err := s.InsertTx(TstDoubleSpendTx, TstRecvTxBlockDetails)
if err != nil {
return nil, err
}
@ -290,7 +290,7 @@ func TestInsertsCreditsDebitsRollbacks(t *testing.T) {
return nil, err
}
_, err = r.AddDebits(prev.Credits())
_, err = r.AddDebits()
if err != nil {
return nil, err
}
@ -548,7 +548,7 @@ func TestInsertsCreditsDebitsRollbacks(t *testing.T) {
}
func TestFindingSpentCredits(t *testing.T) {
s := New()
s := New("/tmp/tx.bin")
// Insert transaction and credit which will be spent.
r, err := s.InsertTx(TstRecvTx, TstRecvTxBlockDetails)
@ -570,7 +570,7 @@ func TestFindingSpentCredits(t *testing.T) {
if err != nil {
t.Fatal(err)
}
_, err = r2.AddDebits(nil)
_, err = r2.AddDebits()
if err != nil {
t.Fatal(err)
}

166
updates.go
View file

@ -1,166 +0,0 @@
/*
* Copyright (c) 2013, 2014 Conformal Systems LLC <info@conformal.com>
*
* Permission to use, copy, modify, and distribute this software for any
* purpose with or without fee is hereby granted, provided that the above
* copyright notice and this permission notice appear in all copies.
*
* THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
* WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
* MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
* ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
* WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
* ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
* OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
*/
package main
import (
"errors"
"os"
"path/filepath"
"strings"
)
// ErrNotAccountDir describes an error where a directory in the btcwallet
// data directory cannot be parsed as a directory holding account files.
var ErrNotAccountDir = errors.New("directory is not an account directory")
// updateOldFileLocations moves files for wallets, transactions, and
// recorded unspent transaction outputs to more recent locations.
func updateOldFileLocations() error {
// Before version 0.1.1, accounts were saved with the following
// format:
//
// ~/.btcwallet/
// - btcwallet/
// - wallet.bin
// - tx.bin
// - utxo.bin
// - btcwallet-AccountA/
// - wallet.bin
// - tx.bin
// - utxo.bin
//
// This format does not scale well (see Github issue #16), and
// since version 0.1.1, the above directory format has changed
// to the following:
//
// ~/.btcwallet/
// - testnet/
// - wallet.bin
// - tx.bin
// - utxo.bin
// - AccountA-wallet.bin
// - AccountA-tx.bin
// - AccountA-utxo.bin
//
// Previous account files are placed in the testnet directory
// as 0.1.0 and earlier only ran on testnet.
//
// UTXOs and transaction history are intentionally not moved over, as
// the UTXO file is no longer used (it was combined with txstore), and
// the tx history is now written in an incompatible format and would
// be ignored on first read.
datafi, err := os.Open(cfg.DataDir)
if err != nil {
return nil
}
defer func() {
if err := datafi.Close(); err != nil {
log.Warnf("Cannot close data directory: %v", err)
}
}()
// Get info on all files in the data directory.
fi, err := datafi.Readdir(0)
if err != nil {
log.Errorf("Cannot read files in data directory: %v", err)
return err
}
acctsExist := false
for i := range fi {
// Ignore non-directories.
if !fi[i].IsDir() {
continue
}
if strings.HasPrefix(fi[i].Name(), "btcwallet") {
acctsExist = true
break
}
}
if !acctsExist {
return nil
}
// Create testnet directory, if it doesn't already exist.
netdir := filepath.Join(cfg.DataDir, "testnet")
if err := checkCreateDir(netdir); err != nil {
log.Errorf("Cannot continue without a testnet directory: %v", err)
return err
}
// Check all files in the datadir for old accounts to update.
for i := range fi {
// Ignore non-directories.
if !fi[i].IsDir() {
continue
}
account, err := parseOldAccountDir(cfg.DataDir, fi[i].Name())
switch err {
case nil:
break
case ErrNotAccountDir:
continue
default: // all other non-nil errors
log.Errorf("Cannot open old account directory: %v", err)
return err
}
log.Infof("Updating old file locations for account %v", account)
// Move old wallet.bin, if any.
old := filepath.Join(cfg.DataDir, fi[i].Name(), "wallet.bin")
if fileExists(old) {
new := accountFilename("wallet.bin", account, netdir)
if err := Rename(old, new); err != nil {
log.Errorf("Cannot move old %v for account %v to new location: %v",
"wallet.bin", account, err)
return err
}
}
// Cleanup old account directory.
if err := os.RemoveAll(filepath.Join(cfg.DataDir, fi[i].Name())); err != nil {
log.Warnf("Could not remove pre 0.1.1 account directory: %v", err)
}
}
return nil
}
type oldAccountDir struct {
account string
dir *os.File
}
func parseOldAccountDir(dir, base string) (string, error) {
if base == "btcwallet" {
return "", nil
}
const accountPrefix = "btcwallet-"
if strings.HasPrefix(base, accountPrefix) {
account := strings.TrimPrefix(base, accountPrefix)
return account, nil
}
return "", ErrNotAccountDir
}

1367
wallet.go Normal file
View file

File diff suppressed because it is too large Load diff