diff --git a/channeldb/db.go b/channeldb/db.go index 10a6454ed..aa4bbc3a5 100644 --- a/channeldb/db.go +++ b/channeldb/db.go @@ -124,6 +124,10 @@ func (d *DB) Wipe() error { if err != nil && err != bolt.ErrBucketNotFound { return err } + err = tx.DeleteBucket(graphMetaBucket) + if err != nil && err != bolt.ErrBucketNotFound { + return err + } return nil }) @@ -169,6 +173,9 @@ func createChannelDB(dbPath string) error { if _, err := tx.CreateBucket(edgeBucket); err != nil { return err } + if _, err := tx.CreateBucket(graphMetaBucket); err != nil { + return err + } if _, err := tx.CreateBucket(metaBucket); err != nil { return err diff --git a/channeldb/error.go b/channeldb/error.go index 91422dd22..835f81f7a 100644 --- a/channeldb/error.go +++ b/channeldb/error.go @@ -23,6 +23,7 @@ var ( ErrGraphNodesNotFound = fmt.Errorf("no graph nodes exist") ErrGraphNoEdgesFound = fmt.Errorf("no graph edges exist") ErrGraphNodeNotFound = fmt.Errorf("unable to find node") + ErrGraphNeverPruned = fmt.Errorf("graph never pruned") ErrEdgeNotFound = fmt.Errorf("edge for chanID not found") diff --git a/channeldb/graph.go b/channeldb/graph.go index 29ab3c4be..ebe332857 100644 --- a/channeldb/graph.go +++ b/channeldb/graph.go @@ -75,6 +75,19 @@ var ( // maps: outPoint -> chanID channelPointBucket = []byte("chan-index") + // graphMetaBucket is a top-level bucket which stores various meta-deta + // related to the on-disk channel graph. Data strored in this bucket + // includes the block to which the graph has been synced to, the total + // number of channels, etc. + graphMetaBucket = []byte("graph-meta") + + // pruneTipKey is a key within the above graphMetaBucket that stores + // the best known blockhash+height that the channel graph has been + // known to be pruned to. Once a new block is discovered, any channels + // that have been closed (by spending the outpoint) can safely be + // removed from the graph. + pruneTipKey = []byte("prune-tip") + edgeBloomKey = []byte("edge-bloom") nodeBloomKey = []byte("node-bloom") ) @@ -281,6 +294,7 @@ func addLightningNode(tx *bolt.Tx, node *LightningNode) error { } // LookupAlias attempts to return the alias as advertised by the target node. +// TODO(roasbeef): currently assumes that aliases are unique... func (r *ChannelGraph) LookupAlias(pub *btcec.PublicKey) (string, error) { var alias string @@ -439,14 +453,27 @@ func (r *ChannelGraph) HasChannelEdge(chanID uint64) (bool, error) { return b, err } -// DeleteChannelEdge removes an edge from the database as identified by it's -// funding outpoint. If the edge does not exist within the database, then this -func (r *ChannelGraph) DeleteChannelEdge(chanPoint *wire.OutPoint) error { - // TODO(roasbeef): possibly delete from node bucket if node has no more - // channels - // TODO(roasbeef): don't delete both edges? +const ( + // pruneTipBytes is the total size of the value which stores the + // current prune tip of the graph. The prune tip indicates if the + // channel graph is in sync with the current UTXO state. The structure + // is: blockHash || blockHeight, taking 36 bytes total. + pruneTipBytes = 32 + 4 +) - return r.db.Update(func(tx *bolt.Tx) error { +// PruneGraph prunes newly closed channels from the channel graph in response +// to a new block being solved on the network. Any transactions which spend the +// funding output of any known channels withint he graph will be deleted. +// Additionally, the "prune tip", or the last block which has been used to +// prune the graph is stored so callers can ensure the graph is fully in sync +// with the current UTXO state. An integer is returned which reflects the +// number of channels pruned due to the new incoming block. +func (r *ChannelGraph) PruneGraph(spentOutputs []*wire.OutPoint, + blockHash *wire.ShaHash, blockHeight uint32) (uint32, error) { + + var numChans uint32 + + err := r.db.Update(func(tx *bolt.Tx) error { // First grab the edges bucket which houses the information // we'd like to delete edges, err := tx.CreateBucketIfNotExists(edgeBucket) @@ -464,52 +491,159 @@ func (r *ChannelGraph) DeleteChannelEdge(chanPoint *wire.OutPoint) error { return err } - var b bytes.Buffer - if err := writeOutpoint(&b, chanPoint); err != nil { - return err - } + // For each of the outpoints that've been spent within the + // block, we attempt to delete them from the graph as if that + // outpoint was a channel, then it has now been closed. + for _, chanPoint := range spentOutputs { + // TODO(roasbeef): load channel bloom filter, continue + // if NOT if filter - // If the channel's outpoint doesn't exist within the outpoint - // index, then the edge does not exist. - chanID := chanIndex.Get(b.Bytes()) - if chanID == nil { - return ErrEdgeNotFound - } - - // Otherwise we obtain the two public keys from the mapping: - // chanID -> pubKey1 || pubKey2. With this, we can construct - // the keys which house both of the directed edges for this - // channel. - nodeKeys := edgeIndex.Get(chanID) - - // The edge key is of the format pubKey || chanID. First we - // construct the latter half, populating the channel ID. - var edgeKey [33 + 8]byte - copy(edgeKey[33:], chanID) - - // With the latter half constructed, copy over the first public - // key to delete the edge in this direction, then the second to - // delete the edge in the opposite direction. - copy(edgeKey[:33], nodeKeys[:33]) - if edges.Get(edgeKey[:]) != nil { - if err := edges.Delete(edgeKey[:]); err != nil { - return err - } - } - copy(edgeKey[:33], nodeKeys[33:]) - if edges.Get(edgeKey[:]) != nil { - if err := edges.Delete(edgeKey[:]); err != nil { + // Attempt to delete the channel, and ErrEdgeNotFound + // will be returned if that outpoint isn't known to be + // a channel. If no error is returned, then a channel + // was successfully pruned. + err := delChannelByEdge(edges, edgeIndex, chanIndex, + chanPoint) + if err != nil && err != ErrEdgeNotFound { return err + } else if err == nil { + numChans += 1 } } - // Finally, with the edge data deleted, we can purge the - // information from the two edge indexes. - if err := edgeIndex.Delete(chanID); err != nil { + metaBucket, err := tx.CreateBucketIfNotExists(graphMetaBucket) + if err != nil { return err } - return chanIndex.Delete(b.Bytes()) + + // With the graph pruned, update the current "prune tip" which + // can eb used to check if the graph is fully synced with the + // current UTXO state. + var newTip [pruneTipBytes]byte + copy(newTip[:], blockHash[:]) + byteOrder.PutUint32(newTip[32:], uint32(blockHeight)) + + return metaBucket.Put(pruneTipKey, newTip[:]) }) + if err != nil { + return 0, err + } + + return numChans, nil +} + +// PruneTip returns the block height and hash of the latest block that has been +// used to prune channels in the graph. Knowing the "prune tip" allows callers +// to tell if the graph is currently in sync with the current best known UTXO +// state. +func (r *ChannelGraph) PruneTip() (*wire.ShaHash, uint32, error) { + var ( + currentTip [pruneTipBytes]byte + tipHash wire.ShaHash + tipHeight uint32 + ) + + err := r.db.View(func(tx *bolt.Tx) error { + graphMeta := tx.Bucket(graphMetaBucket) + if graphMeta == nil { + return ErrGraphNotFound + } + + tipBytes := graphMeta.Get(pruneTipKey) + if tipBytes == nil { + return ErrGraphNeverPruned + } + copy(currentTip[:], tipBytes) + + return nil + }) + if err != nil { + return nil, 0, err + } + + // Once we have the prune tip, the first 32 bytes are the block hash, + // with the latter 4 bytes being the block height. + copy(tipHash[:], currentTip[:32]) + tipHeight = byteOrder.Uint32(currentTip[32:]) + + return &tipHash, tipHeight, nil +} + +// DeleteChannelEdge removes an edge from the database as identified by it's +// funding outpoint. If the edge does not exist within the database, then this +func (r *ChannelGraph) DeleteChannelEdge(chanPoint *wire.OutPoint) error { + // TODO(roasbeef): possibly delete from node bucket if node has no more + // channels + // TODO(roasbeef): don't delete both edges? + + return r.db.Update(func(tx *bolt.Tx) error { + // First grab the edges bucket which houses the information + // we'd like to delete + edges, err := tx.CreateBucketIfNotExists(edgeBucket) + if err != nil { + return err + } + // Next grab the two edge indexes which will also need to be updated. + edgeIndex, err := edges.CreateBucketIfNotExists(edgeIndexBucket) + if err != nil { + return err + } + chanIndex, err := edges.CreateBucketIfNotExists(channelPointBucket) + if err != nil { + return err + } + + return delChannelByEdge(edges, edgeIndex, chanIndex, chanPoint) + }) +} + +func delChannelByEdge(edges *bolt.Bucket, edgeIndex *bolt.Bucket, + chanIndex *bolt.Bucket, chanPoint *wire.OutPoint) error { + var b bytes.Buffer + if err := writeOutpoint(&b, chanPoint); err != nil { + return err + } + + // If the channel's outpoint doesn't exist within the outpoint + // index, then the edge does not exist. + chanID := chanIndex.Get(b.Bytes()) + if chanID == nil { + return ErrEdgeNotFound + } + + // Otherwise we obtain the two public keys from the mapping: + // chanID -> pubKey1 || pubKey2. With this, we can construct + // the keys which house both of the directed edges for this + // channel. + nodeKeys := edgeIndex.Get(chanID) + + // The edge key is of the format pubKey || chanID. First we + // construct the latter half, populating the channel ID. + var edgeKey [33 + 8]byte + copy(edgeKey[33:], chanID) + + // With the latter half constructed, copy over the first public + // key to delete the edge in this direction, then the second to + // delete the edge in the opposite direction. + copy(edgeKey[:33], nodeKeys[:33]) + if edges.Get(edgeKey[:]) != nil { + if err := edges.Delete(edgeKey[:]); err != nil { + return err + } + } + copy(edgeKey[:33], nodeKeys[33:]) + if edges.Get(edgeKey[:]) != nil { + if err := edges.Delete(edgeKey[:]); err != nil { + return err + } + } + + // Finally, with the edge data deleted, we can purge the + // information from the two edge indexes. + if err := edgeIndex.Delete(chanID); err != nil { + return err + } + return chanIndex.Delete(b.Bytes()) } // UpdateEdgeInfo updates the edge information for a single directed edge diff --git a/channeldb/graph_test.go b/channeldb/graph_test.go index 6c2380767..ff83d5b01 100644 --- a/channeldb/graph_test.go +++ b/channeldb/graph_test.go @@ -7,6 +7,7 @@ import ( prand "math/rand" "net" "reflect" + "runtime" "testing" "time" @@ -505,3 +506,174 @@ func TestGraphTraversal(t *testing.T) { t.Fatalf("all edges for node reached within ForEach") } } + +func assertPruneTip(t *testing.T, graph *ChannelGraph, blockHash *wire.ShaHash, + blockHeight uint32) { + + pruneHash, pruneHeight, err := graph.PruneTip() + if err != nil { + _, _, line, _ := runtime.Caller(1) + t.Fatalf("line %v: unable to fetch prune tip: %v", line, err) + } + if !bytes.Equal(blockHash[:], pruneHash[:]) { + _, _, line, _ := runtime.Caller(1) + t.Fatalf("line: %v, prune tips don't match, expected %x got %x", + line, blockHash, pruneHash) + } + if pruneHeight != blockHeight { + _, _, line, _ := runtime.Caller(1) + t.Fatalf("line %v: prune heights don't match, expected %v "+ + "got %v", line, blockHeight, pruneHeight) + } +} + +func asserNumChans(t *testing.T, graph *ChannelGraph, n int) { + numChans := 0 + if err := graph.ForEachChannel(func(*ChannelEdge, *ChannelEdge) error { + numChans += 1 + return nil + }); err != nil { + _, _, line, _ := runtime.Caller(1) + t.Fatalf("line %v:unable to scan channels: %v", line, err) + } + if numChans != n { + _, _, line, _ := runtime.Caller(1) + t.Fatalf("line %v: expected %v chans instead have %v", line, n, numChans) + } +} + +func TestGraphPruning(t *testing.T) { + db, cleanUp, err := makeTestDB() + if err != nil { + t.Fatalf("unable to make test database: %v", err) + } + defer cleanUp() + + graph := db.ChannelGraph() + + // As initial set up for the test, we'll create a graph with 5 vertexes + // and enough edges to create a fully connected graph. The graph will + // be rather simple, representing a straight line. + const numNodes = 5 + graphNodes := make([]*LightningNode, numNodes) + for i := 0; i < numNodes; i++ { + node, err := createTestVertex(db) + if err != nil { + t.Fatalf("unable to create node: %v", err) + } + + if err := graph.AddLightningNode(node); err != nil { + t.Fatalf("unable to add node: %v", err) + } + + graphNodes[i] = node + } + + // With the vertexes created, we'll next create a series of channels + // between them. + channelPoints := make([]*wire.OutPoint, 0, numNodes-1) + for i := 0; i < numNodes-1; i++ { + txHash := fastsha256.Sum256([]byte{byte(i)}) + chanID := uint64(i + 1) + op := wire.OutPoint{ + Hash: txHash, + Index: 0, + } + + channelPoints = append(channelPoints, &op) + + err := graph.AddChannelEdge(graphNodes[i].PubKey, + graphNodes[i+1].PubKey, &op, chanID) + if err != nil { + t.Fatalf("unable to add node: %v", err) + } + + // Create and add an edge with random data that points from + // node_i -> node_i+1 + edge := randEdge(chanID, op, db) + edge.Flags = 0 + edge.Node = graphNodes[i] + if err := graph.UpdateEdgeInfo(edge); err != nil { + t.Fatalf("unable to update edge: %v", err) + } + + // Create another random edge that points from node_i+1 -> + // node_i this time. + edge = randEdge(chanID, op, db) + edge.Flags = 1 + edge.Node = graphNodes[i] + if err := graph.UpdateEdgeInfo(edge); err != nil { + t.Fatalf("unable to update edge: %v", err) + } + } + + // Now with our test graph created, we can test the pruning + // capabilities of the channel graph. + + // First we create a mock block that ends up closing the first two + // channels. + var blockHash wire.ShaHash + copy(blockHash[:], bytes.Repeat([]byte{1}, 32)) + blockHeight := uint32(1) + block := channelPoints[:2] + numPruned, err := graph.PruneGraph(block, &blockHash, blockHeight) + if err != nil { + t.Fatalf("unable to prune graph: %v", err) + } + if numPruned != 2 { + t.Fatalf("incorrect number of channels pruned: expected %v, got %v", + 2, numPruned) + } + + // Now ensure that the prune tip has been updated. + assertPruneTip(t, graph, &blockHash, blockHeight) + + // Count up the number of channels known within the graph, only 2 + // should be remaining. + asserNumChans(t, graph, 2) + + // Next we'll create a block that doesn't close any channels within the + // graph to test the negative error case. + fakeHash := fastsha256.Sum256([]byte("test prune")) + nonChannel := &wire.OutPoint{ + Hash: fakeHash, + Index: 9, + } + blockHash = fastsha256.Sum256(blockHash[:]) + blockHeight = 2 + numPruned, err = graph.PruneGraph([]*wire.OutPoint{nonChannel}, + &blockHash, blockHeight) + if err != nil { + t.Fatalf("unable to prune graph: %v", err) + } + + // No channels should've been detected as pruned. + if numPruned != 0 { + t.Fatalf("channels were pruned but shouldn't have been") + } + + // Once again, the prune tip should've been updated. + assertPruneTip(t, graph, &blockHash, blockHeight) + asserNumChans(t, graph, 2) + + // Finally, create a block that prunes the remainder of the channels + // from the graph. + blockHash = fastsha256.Sum256(blockHash[:]) + blockHeight = 3 + numPruned, err = graph.PruneGraph(channelPoints[2:], &blockHash, + blockHeight) + if err != nil { + t.Fatalf("unable to prune graph: %v", err) + } + + // The remainder of the channels should've been pruned from the graph. + if numPruned != 2 { + t.Fatalf("incorrect number of channels pruned: expected %v, got %v", + 2, numPruned) + } + + // The prune tip should be updated, and no channels should be found + // within the current graph. + assertPruneTip(t, graph, &blockHash, blockHeight) + asserNumChans(t, graph, 0) +}