channeldb: re-design OpenChannel schema, update tests

The state of OpenChannel on disk has now been partitioned into several
buckets+keys within the db. At the top level, a set of prefixed keys
stored common data updated frequently (with every channel update).
These fields are stored at the top level in order to facilities prefix
scans, and to avoid read/write amplification due to
serialization/deserialization with each read/write.

Within the active channel bucket, a nested bucket keyed on the node’s
ID stores the remainder of the channel.

Additionally OpenChannel now uses elkrem rather than shachain, delivery
scripts instead of addresses, stores the total net fees, and splits the
csv delay into the remote vs local node’s.

Several TODO’s have been left lingering, to be visited in the near
future.
This commit is contained in:
Olaoluwa Osuntokun 2016-03-23 22:39:52 -07:00
parent 631e76519e
commit 87603e780f
4 changed files with 686 additions and 330 deletions

@ -2,47 +2,92 @@ package channeldb
import ( import (
"bytes" "bytes"
"encoding/binary" "encoding/hex"
"fmt" "fmt"
"io" "sync"
"time" "time"
"github.com/LightningNetwork/lnd/elkrem"
"github.com/boltdb/bolt" "github.com/boltdb/bolt"
"github.com/btcsuite/btcd/btcec" "github.com/btcsuite/btcd/btcec"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcd/wire" "github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btcutil" "github.com/btcsuite/btcutil"
"github.com/btcsuite/btcwallet/waddrmgr" "github.com/btcsuite/btcwallet/walletdb"
"github.com/lightningnetwork/lnd/shachain"
) )
var ( var (
openChannelBucket = []byte("o") // openChanBucket stores all the currently open channels. This bucket
closedChannelBucket = []byte("c") // has a second, nested bucket which is keyed by a node's ID. Additionally,
activeChanKey = []byte("a") // at the base level of this bucket several prefixed keys are stored which
// house channel meta-data such as total satoshis sent, number of updates
// etc. These fields are stored at this top level rather than within a
// node's channel bucket in orer to facilitate sequential prefix scans
// to gather stats such as total satoshis received.
openChannelBucket = []byte("ocb")
identityKey = []byte("idkey") // closedChannelBucket stores summarization information concerning
// previously open, but now closed channels.
closedChannelBucket = []byte("ccb")
// TODO(roasbeef): replace w/ tesnet-L also revisit dependancy... // channelLogBucket is dedicated for storing the necessary delta state
ActiveNetParams = &chaincfg.TestNet3Params // between channel updates required to re-construct a past state in
// order to punish a counter party attempting a non-cooperative channel
// closure.
channelLogBucket = []byte("clb")
// identityKey is the key for storing this node's current LD identity key.
identityKey = []byte("idk")
// The following prefixes are stored at the base level within the
// openChannelBucket. In order to retrieve a particular field for an
// active, or historic channel, append the channels ID to the prefix:
// key = prefix || chanID. Storing certain fields at the top level
// using a prefix scheme serves two purposes: first to facilitate
// sequential prefix scans, and second to eliminate write amplification
// caused by serializing/deserializing the *entire* struct with each
// update.
chanCapacityPrefix = []byte("ccp")
selfBalancePrefix = []byte("sbp")
theirBalancePrefix = []byte("tbp")
minFeePerKbPrefix = []byte("mfp")
updatePrefix = []byte("uup")
satSentPrefix = []byte("ssp")
satRecievedPrefix = []byte("srp")
netFeesPrefix = []byte("ntp")
// chanIDKey stores the node, and channelID for an active channel.
chanIDKey = []byte("cik")
// commitKeys stores both commitment keys (ours, and theirs) for an
// active channel. Our private key is stored in an encrypted format
// using channeldb's currently registered cryptoSystem.
commitKeys = []byte("ckk")
// commitTxnsKey stores the full version of both current, non-revoked
// commitment transactions in addition to the csvDelay for both.
commitTxnsKey = []byte("ctk")
// fundingTxnKey stroes the funding tx, our encrypted multi-sig key,
// and finally 2-of-2 multisig redeem script.
fundingTxnKey = []byte("fsk")
// elkremStateKey stores their current revocation hash, and our elkrem
// sender, and their elkrem reciever.
elkremStateKey = []byte("esk")
// deliveryScriptsKey stores the scripts for the final delivery in the
// case of a cooperative closure.
deliveryScriptsKey = []byte("dsk")
) )
// Payment...
type Payment struct {
// r [32]byte
// path *Route
}
// ClosedChannel...
type ClosedChannel struct {
}
// OpenChannel... // OpenChannel...
// TODO(roasbeef): store only the essentials? optimize space... // TODO(roasbeef): Copy/Clone method, so CoW on writes?
// TODO(roasbeef): switch to "column store" // * CoW method would allow for intelligent partial writes for updates
// TODO(roasbeef): UpdateState(func (newChan *OpenChannel) error)
// * need mutex, invarient that all reads/writes grab the mutex
// * needs to also return two slices of added then removed HTLC's
type OpenChannel struct { type OpenChannel struct {
// Hash? or Their current pubKey? // Hash? or Their current pubKey?
// TODO(roasbeef): switch to Tadge's LNId
TheirLNID [wire.HashSize]byte TheirLNID [wire.HashSize]byte
// The ID of a channel is the txid of the funding transaction. // The ID of a channel is the txid of the funding transaction.
@ -67,332 +112,621 @@ type OpenChannel struct {
// commitment transaction includes a valid sigScript, and is ready for // commitment transaction includes a valid sigScript, and is ready for
// broadcast. // broadcast.
TheirCommitTx *wire.MsgTx TheirCommitTx *wire.MsgTx
OurCommitTx *wire.MsgTx // TODO(roasbeef): store hash instead? OurCommitTx *wire.MsgTx
// The final funding transaction. Kept wallet-related records. // The final funding transaction. Kept for wallet-related records.
FundingTx *wire.MsgTx FundingTx *wire.MsgTx
MultiSigKey *btcec.PrivateKey MultiSigKey *btcec.PrivateKey
FundingRedeemScript []byte FundingRedeemScript []byte
// In blocks
LocalCsvDelay uint32
RemoteCsvDelay uint32
// Current revocation for their commitment transaction. However, since // Current revocation for their commitment transaction. However, since
// this is the hash, and not the pre-image, we can't yet verify that // this is the hash, and not the pre-image, we can't yet verify that
// it's actually in the chain. // it's actually in the chain.
TheirCurrentRevocation [20]byte TheirCurrentRevocation [20]byte
TheirShaChain *shachain.HyperShaChain LocalElkrem *elkrem.ElkremSender
OurShaChain *shachain.HyperShaChain RemoteElkrem *elkrem.ElkremReceiver
// Final delivery address // The pkScript for both sides to be used for final delivery in the case
// TODO(roasbeef): should just be output scripts // of a cooperative close.
OurDeliveryAddress btcutil.Address OurDeliveryScript []byte
TheirDeliveryAddress btcutil.Address TheirDeliveryScript []byte
// In blocks
CsvDelay uint32
// TODO(roasbeef): track fees, other stats?
NumUpdates uint64 NumUpdates uint64
TotalSatoshisSent uint64 TotalSatoshisSent uint64
TotalSatoshisReceived uint64 TotalSatoshisReceived uint64
CreationTime time.Time TotalNetFees uint64 // TODO(roasbeef): total fees paid too?
CreationTime time.Time // TODO(roasbeef): last update time?
// isPrevState denotes if this instane of an OpenChannel is a previous,
// revoked channel state. If so, then the FullSynv, and UpdateState
// methods are disabled in order to prevent overiding the latest channel
// state.
// TODO(roasbeef): scrap? already have snapshots now?
isPrevState bool
db *DB
sync.RWMutex
} }
// PutOpenChannel... // ChannelSnapshot....
func (d *DB) PutOpenChannel(channel *OpenChannel) error { // TODO(roasbeef): methods to roll forwards/backwards in state etc
return d.db.Update(func(tx *bolt.Tx) error { // * use botldb cursor?
// Get the bucket dedicated to storing the meta-data for open type ChannelSnapshot struct {
// channels. OpenChannel
openChanBucket, err := tx.CreateBucketIfNotExists(openChannelBucket)
if err != nil {
return err
}
return putOpenChannel(openChanBucket, channel, d.addrmgr) // TODO(roasbeef): active HTLC's + their direction
}) updateNum uint64
deltaNamespace walletdb.Namespace
} }
// GetOpenChannel... // FindPreviousState...
// TODO(roasbeef): assumes only 1 active channel per-node // TODO(roasbeef): method to retrieve both old commitment txns given update #
func (d *DB) FetchOpenChannel(nodeID [32]byte) (*OpenChannel, error) { func (c *OpenChannel) FindPreviousState(updateNum uint64) (*ChannelSnapshot, error) {
var channel *OpenChannel return nil, nil
}
err := d.db.View(func(tx *bolt.Tx) error { // Snapshot....
// Get the bucket dedicated to storing the meta-data for open // read-only snapshot
// channels. func (c *OpenChannel) Snapshot() (*ChannelSnapshot, error) {
openChanBucket := tx.Bucket(openChannelBucket) return nil, nil
if openChannelBucket == nil { }
return fmt.Errorf("open channel bucket does not exist")
}
oChannel, err := fetchOpenChannel(openChanBucket, nodeID, // ChannelDelta...
d.addrmgr) // TODO(roasbeef): binlog like entry?
if err != nil { type ChannelDelta struct {
return err // change in allocations
} // added + removed htlcs
channel = oChannel // index
}
// RecordChannelDelta
// TODO(roasbeef): only need their commit?
// * or as internal helper func to UpdateState func?
func (c OpenChannel) RecordChannelDelta(theirRevokedCommit *wire.MsgTx, updateNum uint64) error {
// TODO(roasbeef): record all HTLCs, pass those instead?
// *
return nil return nil
})
return channel, err
} }
// putChannel... // FullSync serializes, and writes to disk the *full* channel state, using
func putOpenChannel(activeChanBucket *bolt.Bucket, channel *OpenChannel, // both the active channel bucket to store the prefixed column fields, and the
addrmgr *waddrmgr.Manager) error { // remote node's ID to store the remainder of the channel state.
//
// NOTE: This method requires an active EncryptorDecryptor to be registered in
// order to encrypt sensitive information.
func (c *OpenChannel) FullSync() error {
return c.db.store.Update(func(tx *bolt.Tx) error {
// First fetch the top level bucket which stores all data related to
// current, active channels.
chanBucket := tx.Bucket(openChannelBucket)
// Generate a serialized version of the open channel. The addrmgr is // WIthin this top level bucket, fetch the bucket dedicated to storing
// required in order to encrypt densitive data. // open channel data specific to the remote node.
var b bytes.Buffer nodeChanBucket, err := chanBucket.CreateBucketIfNotExists(c.TheirLNID[:])
if err := channel.Encode(&b, addrmgr); err != nil {
return err
}
// Grab the bucket dedicated to storing data related to this particular
// node.
nodeBucket, err := activeChanBucket.CreateBucketIfNotExists(channel.TheirLNID[:])
if err != nil { if err != nil {
return err return err
} }
return nodeBucket.Put(activeChanKey, b.Bytes()) return putOpenChannel(chanBucket, nodeChanBucket, c, c.db.cryptoSystem)
})
} }
// fetchOpenChannel // putChannel serializes, and stores the current state of the channel in its
func fetchOpenChannel(bucket *bolt.Bucket, nodeID [32]byte, // entirety.
addrmgr *waddrmgr.Manager) (*OpenChannel, error) { func putOpenChannel(openChanBucket *bolt.Bucket, nodeChanBucket *bolt.Bucket,
channel *OpenChannel, encryptor EncryptorDecryptor) error {
// Grab the bucket dedicated to storing data related to this particular // First write out all the "common" fields using the field's prefix
// node. // appened with the channel's ID. These fields go into a top-level bucket
nodeBucket := bucket.Bucket(nodeID[:]) // to allow for ease of metric aggregation via efficient prefix scans.
if nodeBucket == nil { if err := putChanCapacity(openChanBucket, channel); err != nil {
return nil, fmt.Errorf("channel bucket for node does not exist") return err
}
if err := putChanMinFeePerKb(openChanBucket, channel); err != nil {
return err
}
if err := putChanNumUpdates(openChanBucket, channel); err != nil {
return err
}
if err := putChanTotalFlow(openChanBucket, channel); err != nil {
return err
}
if err := putChanNetFee(openChanBucket, channel); err != nil {
return err
} }
serializedChannel := nodeBucket.Get(activeChanKey) // Next, write out the fields of the channel update less frequently.
if serializedChannel == nil { if err := putChannelIDs(nodeChanBucket, channel); err != nil {
// TODO(roasbeef): make proper in error.go return err
return nil, fmt.Errorf("node has no open channels") }
if err := putChanCommitKeys(nodeChanBucket, channel, encryptor); err != nil {
return err
}
if err := putChanCommitTxns(nodeChanBucket, channel); err != nil {
return err
}
if err := putChanFundingInfo(nodeChanBucket, channel, encryptor); err != nil {
return err
}
if err := putChanEklremState(nodeChanBucket, channel); err != nil {
return err
}
if err := putChanDeliveryScripts(nodeChanBucket, channel); err != nil {
return err
}
return nil
}
// fetchOpenChannel retrieves, and deserializes (including decrypting
// sensitive) the complete channel currently active with the passed nodeID.
// An EncryptorDecryptor is required to decrypt sensitive information stored
// within the database.
func fetchOpenChannel(openChanBucket *bolt.Bucket, nodeID [32]byte,
decryptor EncryptorDecryptor) (*OpenChannel, error) {
// WIthin this top level bucket, fetch the bucket dedicated to storing
// open channel data specific to the remote node.
nodeChanBucket := openChanBucket.Bucket(nodeID[:])
if nodeChanBucket == nil {
return nil, fmt.Errorf("node chan bucket for node %v does not exist",
hex.EncodeToString(nodeID[:]))
} }
// Decode the serialized channel state, using the addrmgr to decrypt
// sensitive information.
channel := &OpenChannel{} channel := &OpenChannel{}
reader := bytes.NewReader(serializedChannel)
if err := channel.Decode(reader, addrmgr); err != nil { // First, read out the fields of the channel update less frequently.
if err := fetchChannelIDs(nodeChanBucket, channel); err != nil {
return nil, err
}
if err := fetchChanCommitKeys(nodeChanBucket, channel, decryptor); err != nil {
return nil, err
}
if err := fetchChanCommitTxns(nodeChanBucket, channel); err != nil {
return nil, err
}
if err := fetchChanFundingInfo(nodeChanBucket, channel, decryptor); err != nil {
return nil, err
}
if err := fetchChanEklremState(nodeChanBucket, channel); err != nil {
return nil, err
}
if err := fetchChanDeliveryScripts(nodeChanBucket, channel); err != nil {
return nil, err
}
// With the existence of an open channel bucket with this node verified,
// perform a full read of the entire struct. Starting with the prefixed
// fields residing in the parent bucket.
if err := fetchChanCapacity(openChanBucket, channel); err != nil {
return nil, err
}
if err := fetchChanMinFeePerKb(openChanBucket, channel); err != nil {
return nil, err
}
if err := fetchChanNumUpdates(openChanBucket, channel); err != nil {
return nil, err
}
if err := fetchChanTotalFlow(openChanBucket, channel); err != nil {
return nil, err
}
if err := fetchChanNetFee(openChanBucket, channel); err != nil {
return nil, err return nil, err
} }
return channel, nil return channel, nil
} }
// Encode... func putChanCapacity(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
// TODO(roasbeef): checksum // Some scratch bytes re-used for serializing each of the uint64's.
func (o *OpenChannel) Encode(b io.Writer, addrManager *waddrmgr.Manager) error { scratch1 := make([]byte, 8)
if _, err := b.Write(o.TheirLNID[:]); err != nil { scratch2 := make([]byte, 8)
return err scratch3 := make([]byte, 8)
}
if _, err := b.Write(o.ChanID[:]); err != nil { keyPrefix := make([]byte, 3+32)
copy(keyPrefix[3:], channel.ChanID[:])
copy(keyPrefix[:3], chanCapacityPrefix)
byteOrder.PutUint64(scratch1, uint64(channel.Capacity))
if err := openChanBucket.Put(keyPrefix, scratch1); err != nil {
return err return err
} }
if err := binary.Write(b, endian, uint64(o.MinFeePerKb)); err != nil { copy(keyPrefix[:3], selfBalancePrefix)
byteOrder.PutUint64(scratch2, uint64(channel.OurBalance))
if err := openChanBucket.Put(keyPrefix, scratch2); err != nil {
return err return err
} }
encryptedPriv, err := addrManager.Encrypt(waddrmgr.CKTPrivate, copy(keyPrefix[:3], theirBalancePrefix)
o.OurCommitKey.Serialize()) byteOrder.PutUint64(scratch3, uint64(channel.TheirBalance))
return openChanBucket.Put(keyPrefix, scratch3)
}
func fetchChanCapacity(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
// A byte slice re-used to compute each key prefix eblow.
keyPrefix := make([]byte, 3+32)
copy(keyPrefix[3:], channel.ChanID[:])
copy(keyPrefix[:3], chanCapacityPrefix)
capacityBytes := openChanBucket.Get(keyPrefix)
channel.Capacity = btcutil.Amount(byteOrder.Uint64(capacityBytes))
copy(keyPrefix[:3], selfBalancePrefix)
selfBalanceBytes := openChanBucket.Get(keyPrefix)
channel.OurBalance = btcutil.Amount(byteOrder.Uint64(selfBalanceBytes))
copy(keyPrefix[:3], theirBalancePrefix)
theirBalanceBytes := openChanBucket.Get(keyPrefix)
channel.TheirBalance = btcutil.Amount(byteOrder.Uint64(theirBalanceBytes))
return nil
}
func putChanMinFeePerKb(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
scratch := make([]byte, 8)
byteOrder.PutUint64(scratch, uint64(channel.MinFeePerKb))
keyPrefix := make([]byte, 3+32)
copy(keyPrefix, minFeePerKbPrefix)
copy(keyPrefix[3:], channel.ChanID[:])
return openChanBucket.Put(keyPrefix, scratch)
}
func fetchChanMinFeePerKb(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
keyPrefix := make([]byte, 3+32)
copy(keyPrefix, minFeePerKbPrefix)
copy(keyPrefix[3:], channel.ChanID[:])
feeBytes := openChanBucket.Get(keyPrefix)
channel.MinFeePerKb = btcutil.Amount(byteOrder.Uint64(feeBytes))
return nil
}
func putChanNumUpdates(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
scratch := make([]byte, 8)
byteOrder.PutUint64(scratch, channel.NumUpdates)
keyPrefix := make([]byte, 3+32)
copy(keyPrefix, updatePrefix)
copy(keyPrefix[3:], channel.ChanID[:])
return openChanBucket.Put(keyPrefix, scratch)
}
func fetchChanNumUpdates(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
keyPrefix := make([]byte, 3+32)
copy(keyPrefix, updatePrefix)
copy(keyPrefix[3:], channel.ChanID[:])
updateBytes := openChanBucket.Get(keyPrefix)
channel.NumUpdates = byteOrder.Uint64(updateBytes)
return nil
}
func putChanTotalFlow(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
scratch1 := make([]byte, 8)
scratch2 := make([]byte, 8)
keyPrefix := make([]byte, 3+32)
copy(keyPrefix[3:], channel.ChanID[:])
copy(keyPrefix[:3], satSentPrefix)
byteOrder.PutUint64(scratch1, uint64(channel.TotalSatoshisSent))
if err := openChanBucket.Put(keyPrefix, scratch1); err != nil {
return err
}
copy(keyPrefix[:3], satRecievedPrefix)
byteOrder.PutUint64(scratch2, uint64(channel.TotalSatoshisReceived))
return openChanBucket.Put(keyPrefix, scratch2)
}
func fetchChanTotalFlow(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
keyPrefix := make([]byte, 3+32)
copy(keyPrefix[3:], channel.ChanID[:])
copy(keyPrefix[:3], satSentPrefix)
totalSentBytes := openChanBucket.Get(keyPrefix)
channel.TotalSatoshisSent = byteOrder.Uint64(totalSentBytes)
copy(keyPrefix[:3], satRecievedPrefix)
totalReceivedBytes := openChanBucket.Get(keyPrefix)
channel.TotalSatoshisReceived = byteOrder.Uint64(totalReceivedBytes)
return nil
}
func putChanNetFee(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
scratch := make([]byte, 8)
keyPrefix := make([]byte, 3+32)
copy(keyPrefix, netFeesPrefix)
copy(keyPrefix[3:], channel.ChanID[:])
byteOrder.PutUint64(scratch, uint64(channel.TotalNetFees))
return openChanBucket.Put(keyPrefix, scratch)
}
func fetchChanNetFee(openChanBucket *bolt.Bucket, channel *OpenChannel) error {
keyPrefix := make([]byte, 3+32)
copy(keyPrefix, netFeesPrefix)
copy(keyPrefix[3:], channel.ChanID[:])
feeBytes := openChanBucket.Get(keyPrefix)
channel.TotalNetFees = byteOrder.Uint64(feeBytes)
return nil
}
func putChannelIDs(nodeChanBucket *bolt.Bucket, channel *OpenChannel) error {
idSlice := make([]byte, wire.HashSize*2)
copy(idSlice, channel.TheirLNID[:])
copy(idSlice[32:], channel.ChanID[:])
return nodeChanBucket.Put(chanIDKey, idSlice)
}
func fetchChannelIDs(nodeChanBucket *bolt.Bucket, channel *OpenChannel) error {
idBytes := nodeChanBucket.Get(chanIDKey)
copy(channel.TheirLNID[:], idBytes[:32])
copy(channel.ChanID[:], idBytes[32:])
return nil
}
func putChanCommitKeys(nodeChanBucket *bolt.Bucket, channel *OpenChannel,
ed EncryptorDecryptor) error {
var b bytes.Buffer
if _, err := b.Write(channel.TheirCommitKey.SerializeCompressed()); err != nil {
return err
}
encryptedPriv, err := ed.Encrypt(channel.OurCommitKey.Serialize())
if err != nil { if err != nil {
return err return err
} }
if _, err := b.Write(encryptedPriv); err != nil { if _, err := b.Write(encryptedPriv); err != nil {
return err return err
} }
if _, err := b.Write(o.TheirCommitKey.SerializeCompressed()); err != nil {
return err
}
if err := binary.Write(b, endian, uint64(o.Capacity)); err != nil { return nodeChanBucket.Put(commitKeys, b.Bytes())
return err }
}
if err := binary.Write(b, endian, uint64(o.OurBalance)); err != nil {
return err
}
if err := binary.Write(b, endian, uint64(o.TheirBalance)); err != nil {
return err
}
if err := o.TheirCommitTx.Serialize(b); err != nil { func fetchChanCommitKeys(nodeChanBucket *bolt.Bucket, channel *OpenChannel,
return err ed EncryptorDecryptor) error {
}
if err := o.OurCommitTx.Serialize(b); err != nil {
return err
}
if err := o.FundingTx.Serialize(b); err != nil { var err error
return err keyBytes := nodeChanBucket.Get(commitKeys)
}
encryptedPriv, err = addrManager.Encrypt(waddrmgr.CKTPrivate, channel.TheirCommitKey, err = btcec.ParsePubKey(keyBytes[:33], btcec.S256())
o.MultiSigKey.Serialize())
if err != nil { if err != nil {
return err return err
} }
if _, err := b.Write(encryptedPriv); err != nil {
return err decryptedPriv, err := ed.Decrypt(keyBytes[33:])
} if err != nil {
if _, err := b.Write(o.FundingRedeemScript); err != nil {
return err return err
} }
if _, err := b.Write(o.TheirCurrentRevocation[:]); err != nil { channel.OurCommitKey, _ = btcec.PrivKeyFromBytes(btcec.S256(), decryptedPriv)
return err if err != nil {
}
// TODO(roasbeef): serialize shachains
if _, err := b.Write([]byte(o.OurDeliveryAddress.EncodeAddress())); err != nil {
return err
}
if _, err := b.Write([]byte(o.TheirDeliveryAddress.EncodeAddress())); err != nil {
return err
}
if err := binary.Write(b, endian, o.CsvDelay); err != nil {
return err
}
if err := binary.Write(b, endian, o.NumUpdates); err != nil {
return err
}
if err := binary.Write(b, endian, o.TotalSatoshisSent); err != nil {
return err
}
if err := binary.Write(b, endian, o.TotalSatoshisReceived); err != nil {
return err
}
if err := binary.Write(b, endian, o.CreationTime.Unix()); err != nil {
return err return err
} }
return nil return nil
} }
// Decode... func putChanCommitTxns(nodeChanBucket *bolt.Bucket, channel *OpenChannel) error {
func (o *OpenChannel) Decode(b io.Reader, addrManager *waddrmgr.Manager) error { var b bytes.Buffer
var scratch [8]byte
if _, err := b.Read(o.TheirLNID[:]); err != nil { if err := channel.TheirCommitTx.Serialize(&b); err != nil {
return err
}
if _, err := b.Read(o.ChanID[:]); err != nil {
return err return err
} }
if _, err := b.Read(scratch[:]); err != nil { if err := channel.OurCommitTx.Serialize(&b); err != nil {
return err
}
o.MinFeePerKb = btcutil.Amount(endian.Uint64(scratch[:]))
// nonce + serPrivKey + mac
var encryptedPriv [24 + 32 + 16]byte
if _, err := b.Read(encryptedPriv[:]); err != nil {
return err
}
decryptedPriv, err := addrManager.Decrypt(waddrmgr.CKTPrivate, encryptedPriv[:])
if err != nil {
return err
}
o.OurCommitKey, _ = btcec.PrivKeyFromBytes(btcec.S256(), decryptedPriv)
var serPubKey [33]byte
if _, err := b.Read(serPubKey[:]); err != nil {
return err
}
o.TheirCommitKey, err = btcec.ParsePubKey(serPubKey[:], btcec.S256())
if err != nil {
return err return err
} }
if _, err := b.Read(scratch[:]); err != nil { scratch := make([]byte, 4)
byteOrder.PutUint32(scratch, channel.LocalCsvDelay)
if _, err := b.Write(scratch); err != nil {
return err return err
} }
o.Capacity = btcutil.Amount(endian.Uint64(scratch[:])) byteOrder.PutUint32(scratch, channel.RemoteCsvDelay)
if _, err := b.Read(scratch[:]); err != nil { if _, err := b.Write(scratch); err != nil {
return err
}
o.OurBalance = btcutil.Amount(endian.Uint64(scratch[:]))
if _, err := b.Read(scratch[:]); err != nil {
return err
}
o.TheirBalance = btcutil.Amount(endian.Uint64(scratch[:]))
o.TheirCommitTx = wire.NewMsgTx()
if err := o.TheirCommitTx.Deserialize(b); err != nil {
return err
}
o.OurCommitTx = wire.NewMsgTx()
if err := o.OurCommitTx.Deserialize(b); err != nil {
return err return err
} }
o.FundingTx = wire.NewMsgTx() return nodeChanBucket.Put(commitTxnsKey, b.Bytes())
if err := o.FundingTx.Deserialize(b); err != nil { }
func fetchChanCommitTxns(nodeChanBucket *bolt.Bucket, channel *OpenChannel) error {
txnBytes := bytes.NewReader(nodeChanBucket.Get(commitTxnsKey))
channel.TheirCommitTx = wire.NewMsgTx()
if err := channel.TheirCommitTx.Deserialize(txnBytes); err != nil {
return err return err
} }
if _, err := b.Read(encryptedPriv[:]); err != nil { channel.OurCommitTx = wire.NewMsgTx()
return err if err := channel.OurCommitTx.Deserialize(txnBytes); err != nil {
}
decryptedPriv, err = addrManager.Decrypt(waddrmgr.CKTPrivate, encryptedPriv[:])
if err != nil {
return err
}
o.MultiSigKey, _ = btcec.PrivKeyFromBytes(btcec.S256(), decryptedPriv)
var redeemScript [71]byte
if _, err := b.Read(redeemScript[:]); err != nil {
return err
}
o.FundingRedeemScript = redeemScript[:]
if _, err := b.Read(o.TheirCurrentRevocation[:]); err != nil {
return err return err
} }
var addr [34]byte scratch := make([]byte, 4)
if _, err := b.Read(addr[:]); err != nil {
return err
}
o.OurDeliveryAddress, err = btcutil.DecodeAddress(string(addr[:]), ActiveNetParams)
if err != nil {
return err
}
if _, err := b.Read(addr[:]); err != nil { if _, err := txnBytes.Read(scratch); err != nil {
return err
}
o.TheirDeliveryAddress, err = btcutil.DecodeAddress(string(addr[:]), ActiveNetParams)
if err != nil {
return err return err
} }
channel.LocalCsvDelay = byteOrder.Uint32(scratch)
if err := binary.Read(b, endian, &o.CsvDelay); err != nil { if _, err := txnBytes.Read(scratch); err != nil {
return err
}
channel.RemoteCsvDelay = byteOrder.Uint32(scratch)
return nil
}
func putChanFundingInfo(nodeChanBucket *bolt.Bucket, channel *OpenChannel,
ed EncryptorDecryptor) error {
var b bytes.Buffer
if err := channel.FundingTx.Serialize(&b); err != nil {
return err
}
encryptedPriv, err := ed.Encrypt(channel.MultiSigKey.Serialize())
if err != nil {
return err
}
if err := wire.WriteVarBytes(&b, 0, encryptedPriv); err != nil {
return err
}
if err := wire.WriteVarBytes(&b, 0, channel.FundingRedeemScript[:]); err != nil {
return err
}
scratch := make([]byte, 8)
byteOrder.PutUint64(scratch, uint64(channel.CreationTime.Unix()))
if _, err := b.Write(scratch); err != nil {
return err
}
return nodeChanBucket.Put(fundingTxnKey, b.Bytes())
}
func fetchChanFundingInfo(nodeChanBucket *bolt.Bucket, channel *OpenChannel,
ed EncryptorDecryptor) error {
infoBytes := bytes.NewReader(nodeChanBucket.Get(fundingTxnKey))
channel.FundingTx = wire.NewMsgTx()
if err := channel.FundingTx.Deserialize(infoBytes); err != nil {
return err
}
encryptedPrivBytes, err := wire.ReadVarBytes(infoBytes, 0, 100, "")
if err != nil {
return err
}
decryptedPriv, err := ed.Decrypt(encryptedPrivBytes)
if err != nil {
return err
}
channel.MultiSigKey, _ = btcec.PrivKeyFromBytes(btcec.S256(), decryptedPriv)
channel.FundingRedeemScript, err = wire.ReadVarBytes(infoBytes, 0, 520, "")
if err != nil {
return err
}
scratch := make([]byte, 8)
if _, err := infoBytes.Read(scratch); err != nil {
return err
}
unixSecs := byteOrder.Uint64(scratch)
channel.CreationTime = time.Unix(int64(unixSecs), 0)
return nil
}
func putChanEklremState(nodeChanBucket *bolt.Bucket, channel *OpenChannel) error {
var b bytes.Buffer
if _, err := b.Write(channel.TheirCurrentRevocation[:]); err != nil {
return err
}
senderBytes, err := channel.LocalElkrem.ToBytes()
if err != nil {
return err
}
if err := wire.WriteVarBytes(&b, 0, senderBytes); err != nil {
return err
}
reciverBytes, err := channel.RemoteElkrem.ToBytes()
if err != nil {
return err
}
if err := wire.WriteVarBytes(&b, 0, reciverBytes); err != nil {
return err
}
return nodeChanBucket.Put(elkremStateKey, b.Bytes())
}
func fetchChanEklremState(nodeChanBucket *bolt.Bucket, channel *OpenChannel) error {
elkremStateBytes := bytes.NewReader(nodeChanBucket.Get(elkremStateKey))
if _, err := elkremStateBytes.Read(channel.TheirCurrentRevocation[:]); err != nil {
return err
}
senderBytes, err := wire.ReadVarBytes(elkremStateBytes, 0, 1000, "")
if err != nil {
return err
}
localE, err := elkrem.ElkremSenderFromBytes(senderBytes)
if err != nil {
return err
}
channel.LocalElkrem = &localE
reciverBytes, err := wire.ReadVarBytes(elkremStateBytes, 0, 1000, "")
if err != nil {
return err
}
remoteE, err := elkrem.ElkremReceiverFromBytes(reciverBytes)
if err != nil {
return err
}
channel.RemoteElkrem = &remoteE
return nil
}
func putChanDeliveryScripts(nodeChanBucket *bolt.Bucket, channel *OpenChannel) error {
var b bytes.Buffer
if err := wire.WriteVarBytes(&b, 0, channel.OurDeliveryScript); err != nil {
return err
}
if err := wire.WriteVarBytes(&b, 0, channel.TheirDeliveryScript); err != nil {
return err
}
return nodeChanBucket.Put(deliveryScriptsKey, b.Bytes())
}
func fetchChanDeliveryScripts(nodeChanBucket *bolt.Bucket, channel *OpenChannel) error {
var err error
deliveryBytes := bytes.NewReader(nodeChanBucket.Get(deliveryScriptsKey))
channel.OurDeliveryScript, err = wire.ReadVarBytes(deliveryBytes, 0, 520, "")
if err != nil {
return err
}
channel.TheirDeliveryScript, err = wire.ReadVarBytes(deliveryBytes, 0, 520, "")
if err != nil {
return err return err
} }
if err := binary.Read(b, endian, &o.NumUpdates); err != nil {
return err
}
if err := binary.Read(b, endian, &o.TotalSatoshisSent); err != nil {
return err
}
if err := binary.Read(b, endian, &o.TotalSatoshisReceived); err != nil {
return err
}
var unix int64
if err := binary.Read(b, endian, &unix); err != nil {
return err
}
o.CreationTime = time.Unix(unix, 0)
return nil return nil
} }

@ -2,19 +2,16 @@ package channeldb
import ( import (
"bytes" "bytes"
"fmt"
"io/ioutil" "io/ioutil"
"os" "os"
"path/filepath"
"testing" "testing"
"time" "time"
"github.com/LightningNetwork/lnd/elkrem"
"github.com/Roasbeef/btcd/txscript"
"github.com/btcsuite/btcd/btcec" "github.com/btcsuite/btcd/btcec"
"github.com/btcsuite/btcd/txscript"
"github.com/btcsuite/btcd/wire" "github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btcutil" "github.com/btcsuite/btcutil"
"github.com/btcsuite/btcwallet/waddrmgr"
"github.com/btcsuite/btcwallet/walletdb"
_ "github.com/btcsuite/btcwallet/walletdb/bdb" _ "github.com/btcsuite/btcwallet/walletdb/bdb"
) )
@ -70,61 +67,40 @@ var (
} }
) )
// createDbNamespace creates a new wallet database at the provided path and type MockEncryptorDecryptor struct {
// returns it along with the address manager namespace.
func createDbNamespace(dbPath string) (walletdb.DB, walletdb.Namespace, error) {
db, err := walletdb.Create("bdb", dbPath)
if err != nil {
fmt.Println("fuk")
return nil, nil, err
}
namespace, err := db.Namespace([]byte("waddr"))
if err != nil {
db.Close()
return nil, nil, err
}
return db, namespace, nil
} }
// setupManager creates a new address manager and returns a teardown function func (m *MockEncryptorDecryptor) Encrypt(n []byte) ([]byte, error) {
// that should be invoked to ensure it is closed and removed upon completion. return n, nil
func createTestManager(t *testing.T) (tearDownFunc func(), mgr *waddrmgr.Manager) {
t.Parallel()
// Create a new manager in a temp directory.
dirName, err := ioutil.TempDir("", "mgrtest")
if err != nil {
t.Fatalf("Failed to create db temp dir: %v", err)
}
dbPath := filepath.Join(dirName, "mgrtest.db")
db, namespace, err := createDbNamespace(dbPath)
if err != nil {
_ = os.RemoveAll(dirName)
t.Fatalf("createDbNamespace: unexpected error: %v", err)
}
mgr, err = waddrmgr.Create(namespace, key[:], []byte("test"),
[]byte("test"), ActiveNetParams, nil)
if err != nil {
db.Close()
_ = os.RemoveAll(dirName)
t.Fatalf("Failed to create Manager: %v", err)
}
tearDownFunc = func() {
mgr.Close()
db.Close()
_ = os.RemoveAll(dirName)
}
if err := mgr.Unlock([]byte("test")); err != nil {
t.Fatalf("unable to unlock mgr: %v", err)
}
return tearDownFunc, mgr
} }
func (m *MockEncryptorDecryptor) Decrypt(n []byte) ([]byte, error) {
return n, nil
}
func (m *MockEncryptorDecryptor) OverheadSize() uint32 {
return 0
}
var _ EncryptorDecryptor = (*MockEncryptorDecryptor)(nil)
func TestOpenChannelEncodeDecode(t *testing.T) { func TestOpenChannelEncodeDecode(t *testing.T) {
teardown, manager := createTestManager(t) // First, create a temporary directory to be used for the duration of
defer teardown() // this test.
tempDirName, err := ioutil.TempDir("", "channeldb")
if err != nil {
t.Fatalf("unable to create temp dir: %v")
}
defer os.RemoveAll(tempDirName)
// Next, create channeldb for the first time, also setting a mock
// EncryptorDecryptor implementation for testing purposes.
cdb, err := Create(tempDirName)
if err != nil {
t.Fatalf("unable to create channeldb: %v", err)
}
cdb.RegisterCryptoSystem(&MockEncryptorDecryptor{})
defer cdb.Close()
privKey, pubKey := btcec.PrivKeyFromBytes(btcec.S256(), key[:]) privKey, pubKey := btcec.PrivKeyFromBytes(btcec.S256(), key[:])
addr, err := btcutil.NewAddressPubKey(pubKey.SerializeCompressed(), ActiveNetParams) addr, err := btcutil.NewAddressPubKey(pubKey.SerializeCompressed(), ActiveNetParams)
@ -137,6 +113,21 @@ func TestOpenChannelEncodeDecode(t *testing.T) {
t.Fatalf("unable to create redeemScript") t.Fatalf("unable to create redeemScript")
} }
// Simulate 1000 channel updates via progression of the elkrem
// revocation trees.
sender := elkrem.NewElkremSender(32, key)
receiver := elkrem.NewElkremReceiver(32)
for i := 0; i < 1000; i++ {
preImage, err := sender.AtIndex(uint64(i))
if err != nil {
t.Fatalf("unable to progress elkrem sender: %v", err)
}
if receiver.AddNext(preImage); err != nil {
t.Fatalf("unable to progress elkrem receiver: %v", err)
}
}
state := OpenChannel{ state := OpenChannel{
TheirLNID: id, TheirLNID: id,
ChanID: id, ChanID: id,
@ -145,31 +136,34 @@ func TestOpenChannelEncodeDecode(t *testing.T) {
TheirCommitKey: pubKey, TheirCommitKey: pubKey,
Capacity: btcutil.Amount(10000), Capacity: btcutil.Amount(10000),
OurBalance: btcutil.Amount(3000), OurBalance: btcutil.Amount(3000),
TheirBalance: btcutil.Amount(7000), TheirBalance: btcutil.Amount(9000),
TheirCommitTx: testTx, TheirCommitTx: testTx,
OurCommitTx: testTx, OurCommitTx: testTx,
LocalElkrem: &sender,
RemoteElkrem: &receiver,
FundingTx: testTx, FundingTx: testTx,
MultiSigKey: privKey, MultiSigKey: privKey,
FundingRedeemScript: script, FundingRedeemScript: script,
TheirCurrentRevocation: rev, TheirCurrentRevocation: rev,
OurDeliveryAddress: addr, OurDeliveryScript: script,
TheirDeliveryAddress: addr, TheirDeliveryScript: script,
CsvDelay: 5, LocalCsvDelay: 5,
RemoteCsvDelay: 9,
NumUpdates: 1, NumUpdates: 1,
TotalSatoshisSent: 1, TotalSatoshisSent: 8,
TotalSatoshisReceived: 2, TotalSatoshisReceived: 2,
TotalNetFees: 9,
CreationTime: time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC), CreationTime: time.Date(2009, time.November, 10, 23, 0, 0, 0, time.UTC),
db: cdb,
} }
var b bytes.Buffer if err := state.FullSync(); err != nil {
if err := state.Encode(&b, manager); err != nil { t.Fatalf("unable to save and serialize channel state: %v", err)
t.Fatalf("unable to encode channel state: %v", err)
} }
reader := bytes.NewReader(b.Bytes()) newState, err := cdb.FetchOpenChannel(id)
newState := &OpenChannel{} if err != nil {
if err := newState.Decode(reader, manager); err != nil { t.Fatalf("unable to fetch open channel: %v", err)
t.Fatalf("unable to decode channel state: %v", err)
} }
// The decoded channel state should be identical to what we stored // The decoded channel state should be identical to what we stored
@ -194,7 +188,8 @@ func TestOpenChannelEncodeDecode(t *testing.T) {
} }
if state.Capacity != newState.Capacity { if state.Capacity != newState.Capacity {
t.Fatalf("capacity doesn't match") t.Fatalf("capacity doesn't match: %v vs %v", state.Capacity,
newState.Capacity)
} }
if state.OurBalance != newState.OurBalance { if state.OurBalance != newState.OurBalance {
t.Fatalf("our balance doesn't match") t.Fatalf("our balance doesn't match")
@ -248,10 +243,10 @@ func TestOpenChannelEncodeDecode(t *testing.T) {
t.Fatalf("redeem script doesn't match") t.Fatalf("redeem script doesn't match")
} }
if state.OurDeliveryAddress.EncodeAddress() != newState.OurDeliveryAddress.EncodeAddress() { if !bytes.Equal(state.OurDeliveryScript, newState.OurDeliveryScript) {
t.Fatalf("our delivery address doesn't match") t.Fatalf("our delivery address doesn't match")
} }
if state.TheirDeliveryAddress.EncodeAddress() != newState.TheirDeliveryAddress.EncodeAddress() { if !bytes.Equal(state.TheirDeliveryScript, newState.TheirDeliveryScript) {
t.Fatalf("their delivery address doesn't match") t.Fatalf("their delivery address doesn't match")
} }
@ -259,9 +254,13 @@ func TestOpenChannelEncodeDecode(t *testing.T) {
t.Fatalf("num updates doesn't match: %v vs %v", t.Fatalf("num updates doesn't match: %v vs %v",
state.NumUpdates, newState.NumUpdates) state.NumUpdates, newState.NumUpdates)
} }
if state.CsvDelay != newState.CsvDelay { if state.RemoteCsvDelay != newState.RemoteCsvDelay {
t.Fatalf("csv delay doesn't match: %v vs %v", t.Fatalf("csv delay doesn't match: %v vs %v",
state.CsvDelay, newState.CsvDelay) state.RemoteCsvDelay, newState.RemoteCsvDelay)
}
if state.LocalCsvDelay != newState.LocalCsvDelay {
t.Fatalf("csv delay doesn't match: %v vs %v",
state.LocalCsvDelay, newState.LocalCsvDelay)
} }
if state.TotalSatoshisSent != newState.TotalSatoshisSent { if state.TotalSatoshisSent != newState.TotalSatoshisSent {
t.Fatalf("satoshis sent doesn't match: %v vs %v", t.Fatalf("satoshis sent doesn't match: %v vs %v",

@ -132,3 +132,24 @@ func fileExists(path string) bool {
return true return true
} }
// FetchOpenChannel...
func (d *DB) FetchOpenChannel(nodeID [32]byte) (*OpenChannel, error) {
var channel *OpenChannel
err := d.store.View(func(tx *bolt.Tx) error {
// Get the bucket dedicated to storing the meta-data for open
// channels.
openChanBucket := tx.Bucket(openChannelBucket)
if openChannelBucket == nil {
return fmt.Errorf("open channel bucket does not exist")
}
oChannel, err := fetchOpenChannel(openChanBucket, nodeID, d.cryptoSystem)
if err != nil {
return err
}
channel = oChannel
return nil
})
return channel, err
}

@ -6,17 +6,19 @@ import (
"golang.org/x/crypto/ripemd160" "golang.org/x/crypto/ripemd160"
"github.com/boltdb/bolt" "github.com/boltdb/bolt"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcutil" "github.com/btcsuite/btcutil"
) )
var ( var (
idBucket = []byte("i") idBucket = []byte("i")
ActiveNetParams = &chaincfg.TestNet3Params
) )
// PutIdKey saves the hash160 of the public key used for our identity within // PutIdKey saves the hash160 of the public key used for our identity within
// the Lightning Network. // the Lightning Network.
func (d *DB) PutIdKey(pkh []byte) error { func (d *DB) PutIdKey(pkh []byte) error {
return d.db.Update(func(tx *bolt.Tx) error { return d.store.Update(func(tx *bolt.Tx) error {
// Get the bucket dedicated to storing the meta-data for open // Get the bucket dedicated to storing the meta-data for open
// channels. // channels.
bucket, err := tx.CreateBucketIfNotExists(idBucket) bucket, err := tx.CreateBucketIfNotExists(idBucket)
@ -32,7 +34,7 @@ func (d *DB) PutIdKey(pkh []byte) error {
// the Lightning Network as a p2pkh bitcoin address. // the Lightning Network as a p2pkh bitcoin address.
func (d *DB) GetIdAdr() (*btcutil.AddressPubKeyHash, error) { func (d *DB) GetIdAdr() (*btcutil.AddressPubKeyHash, error) {
pkh := make([]byte, ripemd160.Size) pkh := make([]byte, ripemd160.Size)
err := d.db.View(func(tx *bolt.Tx) error { err := d.store.View(func(tx *bolt.Tx) error {
// Get the bucket dedicated to storing the meta-data for open // Get the bucket dedicated to storing the meta-data for open
// channels. // channels.
bucket := tx.Bucket(idBucket) bucket := tx.Bucket(idBucket)