diff --git a/channeldb/db.go b/channeldb/db.go index 8292febbc..076b1c2c1 100644 --- a/channeldb/db.go +++ b/channeldb/db.go @@ -40,6 +40,13 @@ var ( number: 0, migration: nil, }, + { + // The version of the database where two new indexes + // for the update time of node and channel updates were + // added. + number: 1, + migration: migrateNodeAndEdgeUpdateIndex, + }, } // Big endian is the preferred byte order, due to cursor scans over @@ -523,8 +530,9 @@ func (d *DB) FetchClosedChannel(chanID *wire.OutPoint) (*ChannelCloseSummary, er // MarkChanFullyClosed marks a channel as fully closed within the database. A // channel should be marked as fully closed if the channel was initially -// cooperatively closed and it's reached a single confirmation, or after all the -// pending funds in a channel that has been forcibly closed have been swept. +// cooperatively closed and it's reached a single confirmation, or after all +// the pending funds in a channel that has been forcibly closed have been +// swept. func (d *DB) MarkChanFullyClosed(chanPoint *wire.OutPoint) error { return d.Update(func(tx *bolt.Tx) error { var b bytes.Buffer @@ -594,8 +602,9 @@ func (d *DB) syncVersions(versions []version) error { // Otherwise, we fetch the migrations which need to applied, and // execute them serially within a single database transaction to ensure // the migration is atomic. - migrations, migrationVersions := getMigrationsToApply(versions, - meta.DbVersionNumber) + migrations, migrationVersions := getMigrationsToApply( + versions, meta.DbVersionNumber, + ) return d.Update(func(tx *bolt.Tx) error { for i, migration := range migrations { if migration == nil { diff --git a/channeldb/migrations.go b/channeldb/migrations.go index d03b34062..50ceec790 100644 --- a/channeldb/migrations.go +++ b/channeldb/migrations.go @@ -1 +1,114 @@ package channeldb + +import ( + "bytes" + "fmt" + + "github.com/coreos/bbolt" +) + +// migrateNodeAndEdgeUpdateIndex is a migration function that will update the +// database from version 0 to version 1. In version 1, we add two new indexes +// (one for nodes and one for edges) to keep track of the last time a node or +// edge was updated on the network. These new indexes allow us to implement the +// new graph sync protocol added. +func migrateNodeAndEdgeUpdateIndex(tx *bolt.Tx) error { + // First, we'll populating the node portion of the new index. Before we + // can add new values to the index, we'll first create the new bucket + // where these items will be housed. + nodes, err := tx.CreateBucketIfNotExists(nodeBucket) + if err != nil { + return fmt.Errorf("unable to create node bucket: %v", err) + } + nodeUpdateIndex, err := nodes.CreateBucketIfNotExists( + nodeUpdateIndexBucket, + ) + if err != nil { + return fmt.Errorf("unable to create node update index: %v", err) + } + + log.Infof("Populating new node update index bucket") + + // Now that we know the bucket has been created, we'll iterate over the + // entire node bucket so we can add the (updateTime || nodePub) key + // into the node update index. + err = nodes.ForEach(func(nodePub, nodeInfo []byte) error { + if len(nodePub) != 33 { + return nil + } + + log.Tracef("Adding %x to node update index", nodePub) + + // The first 8 bytes of a node's serialize data is the update + // time, so we can extract that without decoding the entire + // structure. + updateTime := nodeInfo[:8] + + // Now that we have the update time, we can construct the key + // to insert into the index. + var indexKey [8 + 33]byte + copy(indexKey[:8], updateTime) + copy(indexKey[8:], nodePub) + + return nodeUpdateIndex.Put(indexKey[:], nil) + }) + if err != nil { + return fmt.Errorf("unable to update node indexes: %v", err) + } + + log.Infof("Populating new edge update index bucket") + + // With the set of nodes updated, we'll now update all edges to have a + // corresponding entry in the edge update index. + edges, err := tx.CreateBucketIfNotExists(edgeBucket) + if err != nil { + return fmt.Errorf("unable to create edge bucket: %v", err) + } + edgeUpdateIndex, err := edges.CreateBucketIfNotExists( + edgeUpdateIndexBucket, + ) + if err != nil { + return fmt.Errorf("unable to create edge update index: %v", err) + } + + // We'll now run through each edge policy in the database, and update + // the index to ensure each edge has the proper record. + err = edges.ForEach(func(edgeKey, edgePolicyBytes []byte) error { + if len(edgeKey) != 41 { + return nil + } + + // Now that we know this is the proper record, we'll grab the + // channel ID (last 8 bytes of the key), and then decode the + // edge policy so we can access the update time. + chanID := edgeKey[33:] + edgePolicyReader := bytes.NewReader(edgePolicyBytes) + + edgePolicy, err := deserializeChanEdgePolicy( + edgePolicyReader, nodes, + ) + if err != nil { + return err + } + + log.Tracef("Adding chan_id=%v to edge update index", + edgePolicy.ChannelID) + + // We'll now construct the index key using the channel ID, and + // the last time it was updated: (updateTime || chanID). + var indexKey [8 + 8]byte + byteOrder.PutUint64( + indexKey[:], uint64(edgePolicy.LastUpdate.Unix()), + ) + copy(indexKey[8:], chanID) + + return edgeUpdateIndex.Put(indexKey[:], nil) + }) + if err != nil { + return fmt.Errorf("unable to update edge indexes: %v", err) + } + + log.Infof("Migration to node and edge update indexes complete!") + + return nil +}