diff --git a/channeldb/channel.go b/channeldb/channel.go index 48a52648d..41f513374 100644 --- a/channeldb/channel.go +++ b/channeldb/channel.go @@ -21,6 +21,7 @@ import ( "github.com/lightningnetwork/lnd/keychain" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/shachain" + "github.com/lightningnetwork/lnd/tlv" ) const ( @@ -45,6 +46,13 @@ var ( // TODO(roasbeef): flesh out comment openChannelBucket = []byte("open-chan-bucket") + // outpointBucket stores all of our channel outpoints and a tlv + // stream containing channel data. + // + // outpoint -> tlv stream + // + outpointBucket = []byte("outpoint-bucket") + // historicalChannelBucket stores all channels that have seen their // commitment tx confirm. All information from their previous open state // is retained. @@ -167,12 +175,35 @@ var ( // the height requested in the revocation log. ErrLogEntryNotFound = fmt.Errorf("log entry not found") + // ErrMissingIndexEntry is returned when a caller attempts to close a + // channel and the outpoint is missing from the index. + ErrMissingIndexEntry = fmt.Errorf("missing outpoint from index") + // errHeightNotFound is returned when a query for channel balances at // a height that we have not reached yet is made. errHeightNotReached = fmt.Errorf("height requested greater than " + "current commit height") ) +const ( + // A tlv type definition used to serialize an outpoint's indexStatus + // for use in the outpoint index. + indexStatusType tlv.Type = 0 +) + +// indexStatus is an enum-like type that describes what state the +// outpoint is in. Currently only two possible values. +type indexStatus uint8 + +const ( + // outpointOpen represents an outpoint that is open in the outpoint index. + outpointOpen indexStatus = 0 + + // outpointClosed represents an outpoint that is closed in the outpoint + // index. + outpointClosed indexStatus = 1 +) + // ChannelType is an enum-like type that describes one of several possible // channel types. Each open channel is associated with a particular type as the // channel type may determine how higher level operations are conducted such as @@ -827,6 +858,39 @@ func fetchChanBucketRw(tx kvdb.RwTx, nodeKey *btcec.PublicKey, // nolint:interfa // fullSync syncs the contents of an OpenChannel while re-using an existing // database transaction. func (c *OpenChannel) fullSync(tx kvdb.RwTx) error { + // Fetch the outpoint bucket and check if the outpoint already exists. + opBucket := tx.ReadWriteBucket(outpointBucket) + + var chanPointBuf bytes.Buffer + if err := writeOutpoint(&chanPointBuf, &c.FundingOutpoint); err != nil { + return err + } + + // Now, check if the outpoint exists in our index. + if opBucket.Get(chanPointBuf.Bytes()) != nil { + return ErrChanAlreadyExists + } + + status := uint8(outpointOpen) + + // Write the status of this outpoint as the first entry in a tlv + // stream. + statusRecord := tlv.MakePrimitiveRecord(indexStatusType, &status) + opStream, err := tlv.NewStream(statusRecord) + if err != nil { + return err + } + + var b bytes.Buffer + if err := opStream.Encode(&b); err != nil { + return err + } + + // Add the outpoint to our outpoint index with the tlv stream. + if err := opBucket.Put(chanPointBuf.Bytes(), b.Bytes()); err != nil { + return err + } + // First fetch the top level bucket which stores all data related to // current, active channels. openChanBucket, err := tx.CreateTopLevelBucket(openChannelBucket) @@ -851,10 +915,6 @@ func (c *OpenChannel) fullSync(tx kvdb.RwTx) error { // With the bucket for the node fetched, we can now go down another // level, creating the bucket for this channel itself. - var chanPointBuf bytes.Buffer - if err := writeOutpoint(&chanPointBuf, &c.FundingOutpoint); err != nil { - return err - } chanBucket, err := chainBucket.CreateBucket( chanPointBuf.Bytes(), ) @@ -1258,7 +1318,7 @@ func (c *OpenChannel) clearChanStatus(status ChannelStatus) error { return nil } -// putChannel serializes, and stores the current state of the channel in its +// putOpenChannel serializes, and stores the current state of the channel in its // entirety. func putOpenChannel(chanBucket kvdb.RwBucket, channel *OpenChannel) error { // First, we'll write out all the relatively static fields, that are @@ -2772,6 +2832,36 @@ func (c *OpenChannel) CloseChannel(summary *ChannelCloseSummary, return err } + // Fetch the outpoint bucket to see if the outpoint exists or + // not. + opBucket := tx.ReadWriteBucket(outpointBucket) + + // Add the closed outpoint to our outpoint index. This should + // replace an open outpoint in the index. + if opBucket.Get(chanPointBuf.Bytes()) == nil { + return ErrMissingIndexEntry + } + + status := uint8(outpointClosed) + + // Write the IndexStatus of this outpoint as the first entry in a tlv + // stream. + statusRecord := tlv.MakePrimitiveRecord(indexStatusType, &status) + opStream, err := tlv.NewStream(statusRecord) + if err != nil { + return err + } + + var b bytes.Buffer + if err := opStream.Encode(&b); err != nil { + return err + } + + // Finally add the closed outpoint and tlv stream to the index. + if err := opBucket.Put(chanPointBuf.Bytes(), b.Bytes()); err != nil { + return err + } + // Add channel state to the historical channel bucket. historicalBucket, err := tx.CreateTopLevelBucket( historicalChannelBucket, diff --git a/channeldb/db.go b/channeldb/db.go index 6cdebd5f4..d2618e10f 100644 --- a/channeldb/db.go +++ b/channeldb/db.go @@ -17,6 +17,7 @@ import ( "github.com/lightningnetwork/lnd/channeldb/migration12" "github.com/lightningnetwork/lnd/channeldb/migration13" "github.com/lightningnetwork/lnd/channeldb/migration16" + "github.com/lightningnetwork/lnd/channeldb/migration20" "github.com/lightningnetwork/lnd/channeldb/migration_01_to_11" "github.com/lightningnetwork/lnd/clock" "github.com/lightningnetwork/lnd/lnwire" @@ -170,6 +171,17 @@ var ( number: 18, migration: mig.CreateTLB(peersBucket), }, + { + // Create a top level bucket which holds outpoint + // information. + number: 19, + migration: mig.CreateTLB(outpointBucket), + }, + { + // Migrate some data to the outpoint index. + number: 20, + migration: migration20.MigrateOutpointIndex, + }, } // Big endian is the preferred byte order, due to cursor scans over @@ -309,13 +321,14 @@ var topLevelBuckets = [][]byte{ graphMetaBucket, metaBucket, closeSummaryBucket, + outpointBucket, } // Wipe completely deletes all saved state within all used buckets within the // database. The deletion is done in a single transaction, therefore this // operation is fully atomic. func (d *DB) Wipe() error { - return kvdb.Update(d, func(tx kvdb.RwTx) error { + err := kvdb.Update(d, func(tx kvdb.RwTx) error { for _, tlb := range topLevelBuckets { err := tx.DeleteTopLevelBucket(tlb) if err != nil && err != kvdb.ErrBucketNotFound { @@ -324,6 +337,11 @@ func (d *DB) Wipe() error { } return nil }, func() {}) + if err != nil { + return err + } + + return initChannelDB(d.Backend) } // createChannelDB creates and initializes a fresh version of channeldb. In diff --git a/channeldb/db_test.go b/channeldb/db_test.go index a86eee695..9b75fd39c 100644 --- a/channeldb/db_test.go +++ b/channeldb/db_test.go @@ -19,6 +19,7 @@ import ( "github.com/lightningnetwork/lnd/keychain" "github.com/lightningnetwork/lnd/lnwire" "github.com/lightningnetwork/lnd/shachain" + "github.com/stretchr/testify/require" ) func TestOpenWithCreate(t *testing.T) { @@ -96,16 +97,13 @@ func TestWipe(t *testing.T) { t.Fatalf("unable to wipe channeldb: %v", err) } // Check correct errors are returned - _, err = cdb.FetchAllOpenChannels() - if err != ErrNoActiveChannels { - t.Fatalf("fetching open channels: expected '%v' instead got '%v'", - ErrNoActiveChannels, err) - } - _, err = cdb.FetchClosedChannels(false) - if err != ErrNoClosedChannels { - t.Fatalf("fetching closed channels: expected '%v' instead got '%v'", - ErrNoClosedChannels, err) - } + openChannels, err := cdb.FetchAllOpenChannels() + require.NoError(t, err, "fetching open channels") + require.Equal(t, 0, len(openChannels)) + + closedChannels, err := cdb.FetchClosedChannels(false) + require.NoError(t, err, "fetching closed channels") + require.Equal(t, 0, len(closedChannels)) } // TestFetchClosedChannelForID tests that we are able to properly retrieve a diff --git a/channeldb/migration20/codec.go b/channeldb/migration20/codec.go new file mode 100644 index 000000000..37481c997 --- /dev/null +++ b/channeldb/migration20/codec.go @@ -0,0 +1,36 @@ +package migration20 + +import ( + "encoding/binary" + "io" + + "github.com/btcsuite/btcd/wire" +) + +var ( + byteOrder = binary.BigEndian +) + +// writeOutpoint writes an outpoint from the passed writer. +func writeOutpoint(w io.Writer, o *wire.OutPoint) error { + if _, err := w.Write(o.Hash[:]); err != nil { + return err + } + if err := binary.Write(w, byteOrder, o.Index); err != nil { + return err + } + + return nil +} + +// readOutpoint reads an outpoint from the passed reader. +func readOutpoint(r io.Reader, o *wire.OutPoint) error { + if _, err := io.ReadFull(r, o.Hash[:]); err != nil { + return err + } + if err := binary.Read(r, byteOrder, &o.Index); err != nil { + return err + } + + return nil +} diff --git a/channeldb/migration20/log.go b/channeldb/migration20/log.go new file mode 100644 index 000000000..da75760d0 --- /dev/null +++ b/channeldb/migration20/log.go @@ -0,0 +1,14 @@ +package migration20 + +import ( + "github.com/btcsuite/btclog" +) + +// log is a logger that is initialized as disabled. This means the package +// will not perform any logging by default until a logger is set. +var log = btclog.Disabled + +// UseLogger uses a specified Logger to output package logging info. +func UseLogger(logger btclog.Logger) { + log = logger +} diff --git a/channeldb/migration20/migration.go b/channeldb/migration20/migration.go new file mode 100644 index 000000000..dee53be77 --- /dev/null +++ b/channeldb/migration20/migration.go @@ -0,0 +1,196 @@ +package migration20 + +import ( + "bytes" + "fmt" + + "github.com/btcsuite/btcd/wire" + "github.com/lightningnetwork/lnd/channeldb/kvdb" + "github.com/lightningnetwork/lnd/tlv" +) + +var ( + // openChanBucket stores all the open channel information. + openChanBucket = []byte("open-chan-bucket") + + // closedChannelBucket stores all the closed channel information. + closedChannelBucket = []byte("closed-chan-bucket") + + // outpointBucket is an index mapping outpoints to a tlv + // stream of channel data. + outpointBucket = []byte("outpoint-bucket") +) + +const ( + // A tlv type definition used to serialize an outpoint's indexStatus for + // use in the outpoint index. + indexStatusType tlv.Type = 0 +) + +// indexStatus is an enum-like type that describes what state the +// outpoint is in. Currently only two possible values. +type indexStatus uint8 + +const ( + // outpointOpen represents an outpoint that is open in the outpoint index. + outpointOpen indexStatus = 0 + + // outpointClosed represents an outpoint that is closed in the outpoint + // index. + outpointClosed indexStatus = 1 +) + +// MigrateOutpointIndex populates the outpoint index with outpoints that +// the node already has. This takes every outpoint in the open channel +// bucket and every outpoint in the closed channel bucket and stores them +// in this index. +func MigrateOutpointIndex(tx kvdb.RwTx) error { + log.Info("Migrating to the outpoint index") + + // First get the set of open outpoints. + openList, err := getOpenOutpoints(tx) + if err != nil { + return err + } + + // Then get the set of closed outpoints. + closedList, err := getClosedOutpoints(tx) + if err != nil { + return err + } + + // Get the outpoint bucket which was created in migration 19. + bucket := tx.ReadWriteBucket(outpointBucket) + + // Store the set of open outpoints in the outpoint bucket. + if err := putOutpoints(bucket, openList, false); err != nil { + return err + } + + // Store the set of closed outpoints in the outpoint bucket. + return putOutpoints(bucket, closedList, true) +} + +// getOpenOutpoints traverses through the openChanBucket and returns the +// list of these channels' outpoints. +func getOpenOutpoints(tx kvdb.RwTx) ([]*wire.OutPoint, error) { + var ops []*wire.OutPoint + + openBucket := tx.ReadBucket(openChanBucket) + if openBucket == nil { + return ops, nil + } + + // Iterate through every node and chain bucket to get every open + // outpoint. + // + // The bucket tree: + // openChanBucket -> nodePub -> chainHash -> chanPoint + err := openBucket.ForEach(func(k, v []byte) error { + // Ensure that the key is the same size as a pubkey and the + // value is nil. + if len(k) != 33 || v != nil { + return nil + } + + nodeBucket := openBucket.NestedReadBucket(k) + if nodeBucket == nil { + return nil + } + + return nodeBucket.ForEach(func(k, v []byte) error { + // If there's a value it's not a bucket. + if v != nil { + return nil + } + + chainBucket := nodeBucket.NestedReadBucket(k) + if chainBucket == nil { + return fmt.Errorf("unable to read "+ + "bucket for chain: %x", k) + } + + return chainBucket.ForEach(func(k, v []byte) error { + // If there's a value it's not a bucket. + if v != nil { + return nil + } + + var op wire.OutPoint + r := bytes.NewReader(k) + if err := readOutpoint(r, &op); err != nil { + return err + } + + ops = append(ops, &op) + + return nil + }) + }) + }) + if err != nil { + return nil, err + } + return ops, nil +} + +// getClosedOutpoints traverses through the closedChanBucket and returns +// a list of closed outpoints. +func getClosedOutpoints(tx kvdb.RwTx) ([]*wire.OutPoint, error) { + var ops []*wire.OutPoint + closedBucket := tx.ReadBucket(closedChannelBucket) + if closedBucket == nil { + return ops, nil + } + + // Iterate through every key-value pair to gather all outpoints. + err := closedBucket.ForEach(func(k, v []byte) error { + var op wire.OutPoint + r := bytes.NewReader(k) + if err := readOutpoint(r, &op); err != nil { + return err + } + + ops = append(ops, &op) + + return nil + }) + if err != nil { + return nil, err + } + + return ops, nil +} + +// putOutpoints puts the set of outpoints into the outpoint bucket. +func putOutpoints(bucket kvdb.RwBucket, ops []*wire.OutPoint, isClosed bool) error { + status := uint8(outpointOpen) + if isClosed { + status = uint8(outpointClosed) + } + + record := tlv.MakePrimitiveRecord(indexStatusType, &status) + stream, err := tlv.NewStream(record) + if err != nil { + return err + } + + var b bytes.Buffer + if err := stream.Encode(&b); err != nil { + return err + } + + // Store the set of outpoints with the encoded tlv stream. + for _, op := range ops { + var opBuf bytes.Buffer + if err := writeOutpoint(&opBuf, op); err != nil { + return err + } + + if err := bucket.Put(opBuf.Bytes(), b.Bytes()); err != nil { + return err + } + } + + return nil +} diff --git a/channeldb/migration20/migration_test.go b/channeldb/migration20/migration_test.go new file mode 100644 index 000000000..c5faf907e --- /dev/null +++ b/channeldb/migration20/migration_test.go @@ -0,0 +1,102 @@ +package migration20 + +import ( + "testing" + + "github.com/lightningnetwork/lnd/channeldb/kvdb" + "github.com/lightningnetwork/lnd/channeldb/migtest" +) + +var ( + hexStr = migtest.Hex + + nodeID = hexStr("031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce") + chainHash = hexStr("81b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce9") + outpoint1 = hexStr("81b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce90fb463ad") + commitKey1 = hexStr("6368616e2d636f6d6d69746d656e742d6b657900") + commitment1 = hexStr("0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000023280000000000000bb828996f0422464389000000000000138801000000010000000000000000000000000000000000000000000000000000000000000000ffffffff070431dc001b0162ffffffff0100f2052a01000000434104d64bdfd09eb1c5fe295abdeb1dca4281be988e2da0b6c1c6a59dc226c28624e18175e851c96b973d81b01cc31f047834bc06d6d6edf620d184241a6aed8b63a6ac0500000047010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010100014730450221008ce2bc69281ce27da07e6683571319d18e949ddfa2965fb6caa1bf0314f882d70220299105481d63e0f4bc2a88121167221b6700d72a0ead154c03be696a292d24ae81b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce9000000000000000a000000010000000001096f6e696f6e626c6f6200000000000000000000000000000000") + commitKey2 = hexStr("6368616e2d636f6d6d69746d656e742d6b657901") + commitment2 = hexStr("000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000bb800000000000023286342a17ec61d66b3000000000000138801000000010000000000000000000000000000000000000000000000000000000000000000ffffffff070431dc001b0162ffffffff0100f2052a01000000434104d64bdfd09eb1c5fe295abdeb1dca4281be988e2da0b6c1c6a59dc226c28624e18175e851c96b973d81b01cc31f047834bc06d6d6edf620d184241a6aed8b63a6ac0500000047010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010101010100014730450221008ce2bc69281ce27da07e6683571319d18e949ddfa2965fb6caa1bf0314f882d70220299105481d63e0f4bc2a88121167221b6700d72a0ead154c03be696a292d24ae81b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce9000000000000000a000000010000000000096f6e696f6e626c6f6200000000000000000000000000000000") + chanInfo = hexStr("1081b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce981b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce90fb463ad6fb5ae2162a7a98e01010000000064000400031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce00000000000027100000000000000008000000000000000201000000010000000000000000000000000000000000000000000000000000000000000000ffffffff070431dc001b0162ffffffff0100f2052a01000000434104d64bdfd09eb1c5fe295abdeb1dca4281be988e2da0b6c1c6a59dc226c28624e18175e851c96b973d81b01cc31f047834bc06d6d6edf620d184241a6aed8b63a6ac0500000041bc5221edd86a320d13a74909313ec348a942f50a648bb159bb2f3362169465c3a4f6d0000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce1eeeb7a89ae233c330b2e1eec7b8115c343083544c79ea954d38071b259e6703c8e5996c000000000000000901031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000010000000801031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000030000000701031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000040000000601031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000020000000501031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce") + frozenInfo = hexStr("00000064") + revState = hexStr("031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce81b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce9010000ffffffffffff537f9d8d5794fb14d5be766c16d3569466fd94186d89dbc48ce34909ef7a604d0000fffffffffffe031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce") + + // openChannel is the data in the open channel bucket in database + // version 16 format. + openChannel = map[string]interface{}{ + nodeID: map[string]interface{}{ + chainHash: map[string]interface{}{ + outpoint1: map[string]interface{}{ + commitKey1: commitment1, + commitKey2: commitment2, + "chan-info-key": chanInfo, + "frozen-chans": frozenInfo, + "revocation-state-key": revState, + }, + }, + }, + } + + outpoint2 = hexStr("81b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce952d6c6c7") + outpoint2Info = hexStr("81b637d8fcd2c6da6859e6963113a1170de793e4b725b84d1e0b4cf99ec58ce952d6c6c700000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000000000000000000000000001f40000000000002710000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce1a1e52008bd73aae77799e9e90cc5c154fbb11704005ce08717d115a04942e89aee57f81000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce000000000000000001031df11fcf521bf537579cd91c192b28343adcd04970817bcd68a3ca43648effce0102bcd89759e0688feed099ea7ed37ceb4fb808e4af814fe1986e3d30107aa3069f00") + + // closedChannel is the data in the closed channel bucket in database + // version 16 format. + closedChannel = map[string]interface{}{ + outpoint2: outpoint2Info, + } + + // outpointData is used to create the outpoint bucket from scratch since + // migration19 creates it. + outpointData = map[string]interface{}{} + + // The tlv streams in string form for both an outpointOpen and outpointClosed + // indexStatus. + tlvOutpointOpen = hexStr("000100") + tlvOutpointClosed = hexStr("000101") + + // post is the expected data in the outpoint bucket after migration. + post = map[string]interface{}{ + outpoint1: tlvOutpointOpen, + outpoint2: tlvOutpointClosed, + } +) + +// TestMigrateOutpointIndex asserts that the database is properly migrated to +// contain an outpoint index. +func TestMigrateOutpointIndex(t *testing.T) { + // Prime the database with both the openChannel and closedChannel data. + // We also create the outpointBucket since migration19 creates it and + // migration20 assumes it exists. + before := func(tx kvdb.RwTx) error { + err := migtest.RestoreDB(tx, openChanBucket, openChannel) + if err != nil { + return err + } + + err = migtest.RestoreDB(tx, closedChannelBucket, closedChannel) + if err != nil { + return err + } + + return migtest.RestoreDB(tx, outpointBucket, outpointData) + } + + // Make sure that the openChanBucket and closedChannelBuckets are untouched + // and that the outpoint index is populated correctly. + after := func(tx kvdb.RwTx) error { + err := migtest.VerifyDB(tx, openChanBucket, openChannel) + if err != nil { + return err + } + + err = migtest.VerifyDB(tx, closedChannelBucket, closedChannel) + if err != nil { + return err + } + + return migtest.VerifyDB(tx, outpointBucket, post) + } + + migtest.ApplyMigration(t, before, after, MigrateOutpointIndex, false) +}