Merge pull request #1719 from cfromknecht/duplicate-payments

Control Tower: Sender-side checks for duplicate payments #2

channeldb/db.go

@@ -67,6 +67,13 @@ var (
 			number:    4,
 			migration: migrateEdgePolicies,
 		},
+		{
+			// The DB version where we persist each attempt to send
+			// an HTLC to a payment hash, and track whether the
+			// payment is in-flight, succeeded, or failed.
+			number:    5,
+			migration: paymentStatusesMigration,
+		},
 	}
 
 	// Big endian is the preferred byte order, due to cursor scans over

channeldb/meta_test.go

@@ -9,6 +9,59 @@ import (
 	"github.com/go-errors/errors"
 )
 
+// applyMigration is a helper test function that encapsulates the general steps
+// which are needed to properly check the result of applying migration function.
+func applyMigration(t *testing.T, beforeMigration, afterMigration func(d *DB),
+	migrationFunc migration, shouldFail bool) {
+
+	cdb, cleanUp, err := makeTestDB()
+	defer cleanUp()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// beforeMigration usually used for populating the database
+	// with test data.
+	beforeMigration(cdb)
+
+	// Create test meta info with zero database version and put it on disk.
+	// Than creating the version list pretending that new version was added.
+	meta := &Meta{DbVersionNumber: 0}
+	if err := cdb.PutMeta(meta); err != nil {
+		t.Fatalf("unable to store meta data: %v", err)
+	}
+
+	versions := []version{
+		{
+			number:    0,
+			migration: nil,
+		},
+		{
+			number:    1,
+			migration: migrationFunc,
+		},
+	}
+
+	defer func() {
+		if r := recover(); r != nil {
+			err = errors.New(r)
+		}
+
+		if err == nil && shouldFail {
+			t.Fatal("error wasn't received on migration stage")
+		} else if err != nil && !shouldFail {
+			t.Fatal("error was received on migration stage")
+		}
+
+		// afterMigration usually used for checking the database state and
+		// throwing the error if something went wrong.
+		afterMigration(cdb)
+	}()
+
+	// Sync with the latest version - applying migration function.
+	err = cdb.syncVersions(versions)
+}
+
 // TestVersionFetchPut checks the propernces of fetch/put methods
 // and also initialization of meta data in case if don't have any in
 // database.
@@ -118,59 +171,8 @@ func TestGlobalVersionList(t *testing.T) {
 	}
 }
 
-// applyMigration is a helper test function that encapsulates the general steps
-// which are needed to properly check the result of applying migration function.
-func applyMigration(t *testing.T, beforeMigration, afterMigration func(d *DB),
-	migrationFunc migration, shouldFail bool) {
-
-	cdb, cleanUp, err := makeTestDB()
-	defer cleanUp()
-	if err != nil {
-		t.Fatal(err)
-	}
-
-	// beforeMigration usually used for populating the database
-	// with test data.
-	beforeMigration(cdb)
-
-	// Create test meta info with zero database version and put it on disk.
-	// Than creating the version list pretending that new version was added.
-	meta := &Meta{DbVersionNumber: 0}
-	if err := cdb.PutMeta(meta); err != nil {
-		t.Fatalf("unable to store meta data: %v", err)
-	}
-
-	versions := []version{
-		{
-			number:    0,
-			migration: nil,
-		},
-		{
-			number:    1,
-			migration: migrationFunc,
-		},
-	}
-
-	defer func() {
-		if r := recover(); r != nil {
-			err = errors.New(r)
-		}
-
-		if err == nil && shouldFail {
-			t.Fatal("error wasn't received on migration stage")
-		} else if err != nil && !shouldFail {
-			t.Fatal("error was received on migration stage")
-		}
-
-		// afterMigration usually used for checking the database state and
-		// throwing the error if something went wrong.
-		afterMigration(cdb)
-	}()
-
-	// Sync with the latest version - applying migration function.
-	err = cdb.syncVersions(versions)
-}
-
+// TestMigrationWithPanic asserts that if migration logic panics, we will return
+// to the original state unaltered.
 func TestMigrationWithPanic(t *testing.T) {
 	t.Parallel()
 
@@ -242,6 +244,8 @@ func TestMigrationWithPanic(t *testing.T) {
 		true)
 }
 
+// TestMigrationWithFatal asserts that migrations which fail do not modify the
+// database.
 func TestMigrationWithFatal(t *testing.T) {
 	t.Parallel()
 
@@ -312,6 +316,8 @@ func TestMigrationWithFatal(t *testing.T) {
 		true)
 }
 
+// TestMigrationWithoutErrors asserts that a successful migration has its
+// changes applied to the database.
 func TestMigrationWithoutErrors(t *testing.T) {
 	t.Parallel()
 

channeldb/migrations.go

@@ -2,6 +2,8 @@ package channeldb
 
 import (
 	"bytes"
+	"crypto/sha256"
+	"encoding/binary"
 	"fmt"
 
 	"github.com/coreos/bbolt"
@@ -373,3 +375,86 @@ func migrateEdgePolicies(tx *bolt.Tx) error {
 
 	return nil
 }
+
+// paymentStatusesMigration is a database migration intended for adding payment
+// statuses for each existing payment entity in bucket to be able control
+// transitions of statuses and prevent cases such as double payment
+func paymentStatusesMigration(tx *bolt.Tx) error {
+	// Get the bucket dedicated to storing statuses of payments,
+	// where a key is payment hash, value is payment status.
+	paymentStatuses, err := tx.CreateBucketIfNotExists(paymentStatusBucket)
+	if err != nil {
+		return err
+	}
+
+	log.Infof("Migrating database to support payment statuses")
+
+	circuitAddKey := []byte("circuit-adds")
+	circuits := tx.Bucket(circuitAddKey)
+	if circuits != nil {
+		log.Infof("Marking all known circuits with status InFlight")
+
+		err = circuits.ForEach(func(k, v []byte) error {
+			// Parse the first 8 bytes as the short chan ID for the
+			// circuit. We'll skip all short chan IDs are not
+			// locally initiated, which includes all non-zero short
+			// chan ids.
+			chanID := binary.BigEndian.Uint64(k[:8])
+			if chanID != 0 {
+				return nil
+			}
+
+			// The payment hash is the third item in the serialized
+			// payment circuit. The first two items are an AddRef
+			// (10 bytes) and the incoming circuit key (16 bytes).
+			const payHashOffset = 10 + 16
+
+			paymentHash := v[payHashOffset : payHashOffset+32]
+
+			return paymentStatuses.Put(
+				paymentHash[:], StatusInFlight.Bytes(),
+			)
+		})
+		if err != nil {
+			return err
+		}
+	}
+
+	log.Infof("Marking all existing payments with status Completed")
+
+	// Get the bucket dedicated to storing payments
+	bucket := tx.Bucket(paymentBucket)
+	if bucket == nil {
+		return nil
+	}
+
+	// For each payment in the bucket, deserialize the payment and mark it
+	// as completed.
+	err = bucket.ForEach(func(k, v []byte) error {
+		// Ignores if it is sub-bucket.
+		if v == nil {
+			return nil
+		}
+
+		r := bytes.NewReader(v)
+		payment, err := deserializeOutgoingPayment(r)
+		if err != nil {
+			return err
+		}
+
+		// Calculate payment hash for current payment.
+		paymentHash := sha256.Sum256(payment.PaymentPreimage[:])
+
+		// Update status for current payment to completed. If it fails,
+		// the migration is aborted and the payment bucket is returned
+		// to its previous state.
+		return paymentStatuses.Put(paymentHash[:], StatusCompleted.Bytes())
+	})
+	if err != nil {
+		return err
+	}
+
+	log.Infof("Migration of payment statuses complete!")
+
+	return nil
+}

channeldb/migrations_test.go

@@ -0,0 +1,191 @@
+package channeldb
+
+import (
+	"crypto/sha256"
+	"encoding/binary"
+	"testing"
+
+	"github.com/coreos/bbolt"
+)
+
+// TestPaymentStatusesMigration checks that already completed payments will have
+// their payment statuses set to Completed after the migration.
+func TestPaymentStatusesMigration(t *testing.T) {
+	t.Parallel()
+
+	fakePayment := makeFakePayment()
+	paymentHash := sha256.Sum256(fakePayment.PaymentPreimage[:])
+
+	// Add fake payment to test database, verifying that it was created,
+	// that we have only one payment, and its status is not "Completed".
+	beforeMigrationFunc := func(d *DB) {
+		if err := d.AddPayment(fakePayment); err != nil {
+			t.Fatalf("unable to add payment: %v", err)
+		}
+
+		payments, err := d.FetchAllPayments()
+		if err != nil {
+			t.Fatalf("unable to fetch payments: %v", err)
+		}
+
+		if len(payments) != 1 {
+			t.Fatalf("wrong qty of paymets: expected 1, got %v",
+				len(payments))
+		}
+
+		paymentStatus, err := d.FetchPaymentStatus(paymentHash)
+		if err != nil {
+			t.Fatalf("unable to fetch payment status: %v", err)
+		}
+
+		// We should receive default status if we have any in database.
+		if paymentStatus != StatusGrounded {
+			t.Fatalf("wrong payment status: expected %v, got %v",
+				StatusGrounded.String(), paymentStatus.String())
+		}
+
+		// Lastly, we'll add a locally-sourced circuit and
+		// non-locally-sourced circuit to the circuit map. The
+		// locally-sourced payment should end up with an InFlight
+		// status, while the other should remain unchanged, which
+		// defaults to Grounded.
+		err = d.Update(func(tx *bolt.Tx) error {
+			circuits, err := tx.CreateBucketIfNotExists(
+				[]byte("circuit-adds"),
+			)
+			if err != nil {
+				return err
+			}
+
+			groundedKey := make([]byte, 16)
+			binary.BigEndian.PutUint64(groundedKey[:8], 1)
+			binary.BigEndian.PutUint64(groundedKey[8:], 1)
+
+			// Generated using TestHalfCircuitSerialization with nil
+			// ErrorEncrypter, which is the case for locally-sourced
+			// payments. No payment status should end up being set
+			// for this circuit, since the short channel id of the
+			// key is non-zero (e.g., a forwarded circuit). This
+			// will default it to Grounded.
+			groundedCircuit := []byte{
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x01,
+				// start payment hash
+				0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				// end payment hash
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x0f,
+				0x42, 0x40, 0x00,
+			}
+
+			err = circuits.Put(groundedKey, groundedCircuit)
+			if err != nil {
+				return err
+			}
+
+			inFlightKey := make([]byte, 16)
+			binary.BigEndian.PutUint64(inFlightKey[:8], 0)
+			binary.BigEndian.PutUint64(inFlightKey[8:], 1)
+
+			// Generated using TestHalfCircuitSerialization with nil
+			// ErrorEncrypter, which is not the case for forwarded
+			// payments, but should have no impact on the
+			// correctness of the test. The payment status for this
+			// circuit should be set to InFlight, since the short
+			// channel id in the key is 0 (sourceHop).
+			inFlightCircuit := []byte{
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x01,
+				// start payment hash
+				0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				// end payment hash
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+				0x00, 0x00, 0x00, 0x00, 0x00, 0x0f,
+				0x42, 0x40, 0x00,
+			}
+
+			return circuits.Put(inFlightKey, inFlightCircuit)
+		})
+		if err != nil {
+			t.Fatalf("unable to add circuit map entry: %v", err)
+		}
+	}
+
+	// Verify that the created payment status is "Completed" for our one
+	// fake payment.
+	afterMigrationFunc := func(d *DB) {
+		meta, err := d.FetchMeta(nil)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if meta.DbVersionNumber != 1 {
+			t.Fatal("migration 'paymentStatusesMigration' wasn't applied")
+		}
+
+		// Check that our completed payments were migrated.
+		paymentStatus, err := d.FetchPaymentStatus(paymentHash)
+		if err != nil {
+			t.Fatalf("unable to fetch payment status: %v", err)
+		}
+
+		if paymentStatus != StatusCompleted {
+			t.Fatalf("wrong payment status: expected %v, got %v",
+				StatusCompleted.String(), paymentStatus.String())
+		}
+
+		inFlightHash := [32]byte{
+			0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+		}
+
+		// Check that the locally sourced payment was transitioned to
+		// InFlight.
+		paymentStatus, err = d.FetchPaymentStatus(inFlightHash)
+		if err != nil {
+			t.Fatalf("unable to fetch payment status: %v", err)
+		}
+
+		if paymentStatus != StatusInFlight {
+			t.Fatalf("wrong payment status: expected %v, got %v",
+				StatusInFlight.String(), paymentStatus.String())
+		}
+
+		groundedHash := [32]byte{
+			0x01, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+			0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+		}
+
+		// Check that non-locally sourced payments remain in the default
+		// Grounded state.
+		paymentStatus, err = d.FetchPaymentStatus(groundedHash)
+		if err != nil {
+			t.Fatalf("unable to fetch payment status: %v", err)
+		}
+
+		if paymentStatus != StatusGrounded {
+			t.Fatalf("wrong payment status: expected %v, got %v",
+				StatusGrounded.String(), paymentStatus.String())
+		}
+	}
+
+	applyMigration(t,
+		beforeMigrationFunc,
+		afterMigrationFunc,
+		paymentStatusesMigration,
+		false)
+}

channeldb/payments.go

@@ -3,6 +3,7 @@ package channeldb
 import (
 	"bytes"
 	"encoding/binary"
+	"errors"
 	"io"
 
 	"github.com/coreos/bbolt"
@@ -17,8 +18,65 @@ var (
 	// which is a monotonically increasing uint64.  BoltDB's sequence
 	// feature is used for generating monotonically increasing id.
 	paymentBucket = []byte("payments")
+
+	// paymentStatusBucket is the name of the bucket within the database that
+	// stores the status of a payment indexed by the payment's preimage.
+	paymentStatusBucket = []byte("payment-status")
 )
 
+// PaymentStatus represent current status of payment
+type PaymentStatus byte
+
+const (
+	// StatusGrounded is the status where a payment has never been
+	// initiated, or has been initiated and received an intermittent
+	// failure.
+	StatusGrounded PaymentStatus = 0
+
+	// StatusInFlight is the status where a payment has been initiated, but
+	// a response has not been received.
+	StatusInFlight PaymentStatus = 1
+
+	// StatusCompleted is the status where a payment has been initiated and
+	// the payment was completed successfully.
+	StatusCompleted PaymentStatus = 2
+)
+
+// Bytes returns status as slice of bytes.
+func (ps PaymentStatus) Bytes() []byte {
+	return []byte{byte(ps)}
+}
+
+// FromBytes sets status from slice of bytes.
+func (ps *PaymentStatus) FromBytes(status []byte) error {
+	if len(status) != 1 {
+		return errors.New("payment status is empty")
+	}
+
+	switch PaymentStatus(status[0]) {
+	case StatusGrounded, StatusInFlight, StatusCompleted:
+		*ps = PaymentStatus(status[0])
+	default:
+		return errors.New("unknown payment status")
+	}
+
+	return nil
+}
+
+// String returns readable representation of payment status.
+func (ps PaymentStatus) String() string {
+	switch ps {
+	case StatusGrounded:
+		return "Grounded"
+	case StatusInFlight:
+		return "In Flight"
+	case StatusCompleted:
+		return "Completed"
+	default:
+		return "Unknown"
+	}
+}
+
 // OutgoingPayment represents a successful payment between the daemon and a
 // remote node. Details such as the total fee paid, and the time of the payment
 // are stored.
@@ -129,6 +187,68 @@ func (db *DB) DeleteAllPayments() error {
 	})
 }
 
+// UpdatePaymentStatus sets the payment status for outgoing/finished payments in
+// local database.
+func (db *DB) UpdatePaymentStatus(paymentHash [32]byte, status PaymentStatus) error {
+	return db.Batch(func(tx *bolt.Tx) error {
+		return UpdatePaymentStatusTx(tx, paymentHash, status)
+	})
+}
+
+// UpdatePaymentStatusTx is a helper method that sets the payment status for
+// outgoing/finished payments in the local database. This method accepts a
+// boltdb transaction such that the operation can be composed into other
+// database transactions.
+func UpdatePaymentStatusTx(tx *bolt.Tx,
+	paymentHash [32]byte, status PaymentStatus) error {
+
+	paymentStatuses, err := tx.CreateBucketIfNotExists(paymentStatusBucket)
+	if err != nil {
+		return err
+	}
+
+	return paymentStatuses.Put(paymentHash[:], status.Bytes())
+}
+
+// FetchPaymentStatus returns the payment status for outgoing payment.
+// If status of the payment isn't found, it will default to "StatusGrounded".
+func (db *DB) FetchPaymentStatus(paymentHash [32]byte) (PaymentStatus, error) {
+	var paymentStatus = StatusGrounded
+	err := db.View(func(tx *bolt.Tx) error {
+		var err error
+		paymentStatus, err = FetchPaymentStatusTx(tx, paymentHash)
+		return err
+	})
+	if err != nil {
+		return StatusGrounded, err
+	}
+
+	return paymentStatus, nil
+}
+
+// FetchPaymentStatusTx is a helper method that returns the payment status for
+// outgoing payment.  If status of the payment isn't found, it will default to
+// "StatusGrounded". It accepts the boltdb transactions such that this method
+// can be composed into other atomic operations.
+func FetchPaymentStatusTx(tx *bolt.Tx, paymentHash [32]byte) (PaymentStatus, error) {
+	// The default status for all payments that aren't recorded in database.
+	var paymentStatus = StatusGrounded
+
+	bucket := tx.Bucket(paymentStatusBucket)
+	if bucket == nil {
+		return paymentStatus, nil
+	}
+
+	paymentStatusBytes := bucket.Get(paymentHash[:])
+	if paymentStatusBytes == nil {
+		return paymentStatus, nil
+	}
+
+	paymentStatus.FromBytes(paymentStatusBytes)
+
+	return paymentStatus, nil
+}
+
 func serializeOutgoingPayment(w io.Writer, p *OutgoingPayment) error {
 	var scratch [8]byte
 

channeldb/payments_test.go

@@ -40,6 +40,14 @@ func makeFakePayment() *OutgoingPayment {
 	return fakePayment
 }
 
+func makeFakePaymentHash() [32]byte {
+	var paymentHash [32]byte
+	rBytes, _ := randomBytes(0, 32)
+	copy(paymentHash[:], rBytes)
+
+	return paymentHash
+}
+
 // randomBytes creates random []byte with length in range [minLen, maxLen)
 func randomBytes(minLen, maxLen int) ([]byte, error) {
 	randBuf := make([]byte, minLen+rand.Intn(maxLen-minLen))
@@ -195,3 +203,51 @@ func TestOutgoingPaymentWorkflow(t *testing.T) {
 			len(paymentsAfterDeletion), 0)
 	}
 }
+
+func TestPaymentStatusWorkflow(t *testing.T) {
+	t.Parallel()
+
+	db, cleanUp, err := makeTestDB()
+	defer cleanUp()
+	if err != nil {
+		t.Fatalf("unable to make test db: %v", err)
+	}
+
+	testCases := []struct {
+		paymentHash [32]byte
+		status      PaymentStatus
+	}{
+		{
+			paymentHash: makeFakePaymentHash(),
+			status:      StatusGrounded,
+		},
+		{
+			paymentHash: makeFakePaymentHash(),
+			status:      StatusInFlight,
+		},
+		{
+			paymentHash: makeFakePaymentHash(),
+			status:      StatusCompleted,
+		},
+	}
+
+	for _, testCase := range testCases {
+		err := db.UpdatePaymentStatus(testCase.paymentHash, testCase.status)
+		if err != nil {
+			t.Fatalf("unable to put payment in DB: %v", err)
+		}
+
+		status, err := db.FetchPaymentStatus(testCase.paymentHash)
+		if err != nil {
+			t.Fatalf("unable to fetch payments from DB: %v", err)
+		}
+
+		if status != testCase.status {
+			t.Fatalf("Wrong payments status after reading from DB."+
+				"Got %v, want %v",
+				spew.Sdump(status),
+				spew.Sdump(testCase.status),
+			)
+		}
+	}
+}