persistlog: fix garbage collector, Get, and tests

This commit reverts some of the work of the previous commit. It
fixes the garbage collector to sweep up the CLTV by decrementing
the CLTV for evey block notification it receives. It also fixes
the Get method to return a CLTV value of math.MaxUint32 upon
retrieving no CLTV value. Tests were also cleaned up and fixed.

Author: Crypt-iQ

bench_test.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"testing"
 
-	"github.com/lightningnetwork/lightning-onion/persistlog"
 	"github.com/roasbeef/btcd/btcec"
 )
 
@@ -56,27 +55,28 @@ func BenchmarkPathPacketConstruction(b *testing.B) {
 
 func BenchmarkProcessPacket(b *testing.B) {
 	b.StopTimer()
-
-	// Create the DecayedLog object
-	d := &persistlog.DecayedLog{}
-	if err := d.Start(); err != nil {
-		b.Fatalf("unable to start channeldb")
-	}
-
-	path, _, sphinxPacket, err := newTestRoute(1, d)
+	path, _, sphinxPacket, err := newTestRoute(1)
 	if err != nil {
 		b.Fatalf("unable to create test route: %v", err)
 	}
 	b.ReportAllocs()
+	path[0].d.Start("0")
+	defer shutdown("0", path[0].d)
 	b.StartTimer()
 
 	var (
 		pkt *ProcessedPacket
 	)
+	for i := 0; i < b.N; i++ {
+		pkt, err = path[0].ProcessOnionPacket(sphinxPacket, nil)
+		if err != nil {
+			b.Fatalf("unable to process packet: %v", err)
+		}
 
-	pkt, err = path[0].ProcessOnionPacket(sphinxPacket, nil)
-	if err != nil {
-		b.Fatalf("unable to process packet: %v", err)
+		b.StopTimer()
+		shutdown("0", path[0].d)
+		path[0].d.Start("0")
+		b.StartTimer()
 	}
 
 	p = pkt

glide.lock

@@ -7,6 +7,7 @@ import (
 	"github.com/boltdb/bolt"
 	"github.com/lightningnetwork/lnd/chainntnfs"
 	"github.com/lightningnetwork/lnd/channeldb"
+	"math"
 	"sync"
 )
 
@@ -59,13 +60,14 @@ func (d *DecayedLog) garbageCollector() error {
 outer:
 	for {
 		select {
-		case epoch, ok := <-epochClient.Epochs:
+		case _, ok := <-epochClient.Epochs:
 			if !ok {
 				return fmt.Errorf("Epoch client shutting " +
 					"down")
 			}
 
 			var expiredCltv [][]byte
+			validCltv := make(map[string]uint32)
 			err := d.db.View(func(tx *bolt.Tx) error {
 				// Grab the shared hash bucket
 				sharedHashes := tx.Bucket(sharedHashBucket)
@@ -76,9 +78,13 @@ outer:
 
 				sharedHashes.ForEach(func(k, v []byte) error {
 					cltv := uint32(binary.BigEndian.Uint32(v))
-					if uint32(epoch.Height) > cltv {
+					cltv--
+					if cltv == 0 {
 						// Store expired hash in array
 						expiredCltv = append(expiredCltv, k)
+					} else {
+						// Store valid <hash, cltv> in map
+						validCltv[string(k)] = cltv
 					}
 					return nil
 				})
@@ -94,11 +100,20 @@ outer:
 			for _, hash := range expiredCltv {
 				err = d.Delete(hash)
 				if err != nil {
-					return fmt.Errorf("Unable to delete"+
+					return fmt.Errorf("Unable to delete "+
 						"expired secret: %v", err)
 				}
 			}
 
+			// Update decremented CLTV's via validCltv
+			for hash, cltv := range validCltv {
+				err = d.Put([]byte(hash), cltv)
+				if err != nil {
+					return fmt.Errorf("Unable to decrement "+
+						"cltv value: %v", err)
+				}
+			}
+
 		case <-d.quit:
 			break outer
 		}
@@ -141,9 +156,9 @@ func (d *DecayedLog) Delete(hash []byte) error {
 
 // Get retrieves the CLTV value of a processed HTLC given the first 20 bytes
 // of the Sha-256 hash of the shared secret used during sphinx processing.
-func (d *DecayedLog) Get(hash []byte) (
-	uint32, error) {
-	var value uint32
+func (d *DecayedLog) Get(hash []byte) (uint32, error) {
+	// math.MaxUint32 is returned when Get did not retrieve a value.
+	var value uint32 = math.MaxUint32
 
 	err := d.db.View(func(tx *bolt.Tx) error {
 		// Grab the shared hash bucket which stores the mapping from
@@ -172,10 +187,9 @@ func (d *DecayedLog) Get(hash []byte) (
 	return value, nil
 }
 
-// Put stores a <shared secret hash, CLTV value> key-pair into the
-// sharedHashBucket.
-func (d *DecayedLog) Put(hash []byte,
-	value uint32) error {
+// Put stores a shared secret hash as the key and a slice consisting of the
+// current blockheight and the outgoing CLTV value
+func (d *DecayedLog) Put(hash []byte, value uint32) error {
 
 	var scratch [4]byte
 
@@ -196,13 +210,20 @@ func (d *DecayedLog) Put(hash []byte,
 // Start opens the database we will be using to store hashed shared secrets.
 // It also starts the garbage collector in a goroutine to remove stale
 // database entries.
-func (d *DecayedLog) Start() error {
+func (d *DecayedLog) Start(dbDir string) error {
 	// Create the quit channel
 	d.quit = make(chan struct{})
 
+	var directory string
+	if dbDir == "" {
+		directory = defaultDbDirectory
+	} else {
+		directory = dbDir
+	}
+
 	// Open the channeldb for use.
 	var err error
-	if d.db, err = channeldb.Open(defaultDbDirectory); err != nil {
+	if d.db, err = channeldb.Open(directory); err != nil {
 		return fmt.Errorf("Could not open channeldb: %v", err)
 	}
 

persistlog/decayedlog.go

@@ -6,6 +6,8 @@ import (
 	"github.com/roasbeef/btcd/btcec"
 	"github.com/roasbeef/btcd/chaincfg/chainhash"
 	"github.com/roasbeef/btcd/wire"
+	"math"
+	"os"
 	"testing"
 	"time"
 )
@@ -61,12 +63,19 @@ func generateSharedSecret(pub *btcec.PublicKey, priv *btcec.PrivateKey) [32]byte
 	return sha256.Sum256(s.SerializeCompressed())
 }
 
+// shutdown stops the DecayedLog and deletes the folder enclosing the
+// temporary channel database.
+func shutdown(d *DecayedLog) {
+	os.RemoveAll("tempdir")
+	d.Stop()
+}
+
 // TestDecayedLogGarbageCollector tests the ability of the garbage collector
 // to delete expired cltv values every time a block is received. Expired cltv
 // values are cltv values that are <= current block height.
 func TestDecayedLogGarbageCollector(t *testing.T) {
-	// Random (EXPIRED) cltv value
-	cltv := uint32(200390)
+	// Random (TO-BE-EXPIRED) cltv value
+	cltv := uint32(2)
 
 	// Create the MockNotifier which triggers the garbage collector
 	MockNotifier := &mockNotifier{
@@ -77,11 +86,11 @@ func TestDecayedLogGarbageCollector(t *testing.T) {
 	d := DecayedLog{Notifier: MockNotifier}
 
 	// Open the channeldb (start the garbage collector)
-	err := d.Start()
+	err := d.Start("tempdir")
 	if err != nil {
 		t.Fatalf("Unable to start / open DecayedLog")
 	}
-	defer d.Stop()
+	defer shutdown(&d)
 
 	// Create a new private key on elliptic curve secp256k1
 	priv, err := btcec.NewPrivateKey(btcec.S256())
@@ -110,7 +119,10 @@ func TestDecayedLogGarbageCollector(t *testing.T) {
 	// should remove the entry we just added to sharedHashBucket as it is
 	// now expired.
 	MockNotifier.epochChan <- &chainntnfs.BlockEpoch{
-		Height: int32(cltv + 1),
+		Height: int32(101),
+	}
+	MockNotifier.epochChan <- &chainntnfs.BlockEpoch{
+		Height: int32(102),
 	}
 
 	// Wait for database write (GC is in a goroutine)
@@ -122,7 +134,7 @@ func TestDecayedLogGarbageCollector(t *testing.T) {
 		t.Fatalf("Delete failed - received an error upon Get")
 	}
 
-	if val != 0 {
+	if val != math.MaxUint32 {
 		t.Fatalf("cltv was not deleted")
 	}
 }
@@ -132,17 +144,17 @@ func TestDecayedLogGarbageCollector(t *testing.T) {
 // longer retrieve it.
 func TestDecayedLogInsertionAndDeletion(t *testing.T) {
 	// Random cltv value
-	cltv := uint32(503928)
+	cltv := uint32(5)
 
 	// Create a DecayedLog object
 	d := DecayedLog{}
 
 	// Open the channeldb
-	err := d.Start()
+	err := d.Start("tempdir")
 	if err != nil {
 		t.Fatalf("Unable to start / open DecayedLog")
 	}
-	defer d.Stop()
+	defer shutdown(&d)
 
 	// Create a new private key on elliptic curve secp256k1
 	priv, err := btcec.NewPrivateKey(btcec.S256())
@@ -179,7 +191,7 @@ func TestDecayedLogInsertionAndDeletion(t *testing.T) {
 		t.Fatalf("Delete failed - received the wrong error message")
 	}
 
-	if val != 0 {
+	if val != math.MaxUint32 {
 		t.Fatalf("cltv was not deleted")
 	}
 
@@ -192,17 +204,17 @@ func TestDecayedLogInsertionAndDeletion(t *testing.T) {
 // the cltv value and check that it persists upon startup.
 func TestDecayedLogStartAndStop(t *testing.T) {
 	// Random cltv value
-	cltv := uint32(909020)
+	cltv := uint32(6)
 
 	// Create a DecayedLog object
 	d := DecayedLog{}
 
 	// Open the channeldb
-	err := d.Start()
+	err := d.Start("tempdir")
 	if err != nil {
 		t.Fatalf("Unable to start / open DecayedLog")
 	}
-	defer d.Stop()
+	defer shutdown(&d)
 
 	// Create a new private key on elliptic curve secp256k1
 	priv, err := btcec.NewPrivateKey(btcec.S256())
@@ -231,7 +243,7 @@ func TestDecayedLogStartAndStop(t *testing.T) {
 	d.Stop()
 
 	// Startup the DecayedLog's channeldb
-	err = d.Start()
+	err = d.Start("tempdir")
 	if err != nil {
 		t.Fatalf("Unable to start / open DecayedLog")
 	}
@@ -258,18 +270,18 @@ func TestDecayedLogStartAndStop(t *testing.T) {
 	d.Stop()
 
 	// Startup the DecayedLog's channeldb
-	err = d.Start()
+	err = d.Start("tempdir")
 	if err != nil {
 		t.Fatalf("Unable to start / open DecayedLog")
 	}
 
 	// Assert that hashedSecret is not in the sharedHashBucket
 	val, err := d.Get(hashedSecret[:])
 	if err != nil {
-		t.Fatalf("Delete failed - received the wrong error message")
+		t.Fatalf("Delete failed")
 	}
 
-	if val != 0 {
+	if val != math.MaxUint32 {
 		t.Fatalf("cltv was not deleted")
 	}
 
@@ -280,17 +292,17 @@ func TestDecayedLogStartAndStop(t *testing.T) {
 // and retrieved cltv values are equal.
 func TestDecayedLogStorageAndRetrieval(t *testing.T) {
 	// Random cltv value
-	cltv := uint32(302930)
+	cltv := uint32(7)
 
 	// Create a DecayedLog object
 	d := DecayedLog{}
 
 	// Open the channeldb
-	err := d.Start()
+	err := d.Start("tempdir")
 	if err != nil {
 		t.Fatalf("Unable to start / open DecayedLog")
 	}
-	defer d.Stop()
+	defer shutdown(&d)
 
 	// Create a new private key on elliptic curve secp256k1
 	priv, err := btcec.NewPrivateKey(btcec.S256())

persistlog/decayedlog_test.go

@@ -19,7 +19,7 @@ type PersistLog interface {
 
 	// Start starts up the on-disk persistent log. It returns an error if
 	// one occurs.
-	Start() error
+	Start(string) error
 
 	// Stop safely stops the on-disk persistent log.
 	Stop()