Merge pull request #1937 from halseth/data-loss-protect-resending

Data loss protect resending

channeldb/channel.go

@@ -3,6 +3,7 @@ package channeldb
 import (
 	"bytes"
 	"encoding/binary"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -1832,6 +1833,10 @@ type ChannelCloseSummary struct {
 
 	// LocalChanCfg is the channel configuration for the local node.
 	LocalChanConfig ChannelConfig
+
+	// LastChanSyncMsg is the ChannelReestablish message for this channel
+	// for the state at the point where it was closed.
+	LastChanSyncMsg *lnwire.ChannelReestablish
 }
 
 // CloseChannel closes a previously active Lightning channel. Closing a channel
@@ -2059,7 +2064,12 @@ func serializeChannelCloseSummary(w io.Writer, cs *ChannelCloseSummary) error {
 	// If this is a close channel summary created before the addition of
 	// the new fields, then we can exit here.
 	if cs.RemoteCurrentRevocation == nil {
-		return nil
+		return WriteElements(w, false)
+	}
+
+	// If fields are present, write boolean to indicate this, and continue.
+	if err := WriteElements(w, true); err != nil {
+		return err
 	}
 
 	if err := WriteElements(w, cs.RemoteCurrentRevocation); err != nil {
@@ -2070,14 +2080,34 @@ func serializeChannelCloseSummary(w io.Writer, cs *ChannelCloseSummary) error {
 		return err
 	}
 
-	// We'll write this field last, as it's possible for a channel to be
-	// closed before we learn of the next unrevoked revocation point for
-	// the remote party.
-	if cs.RemoteNextRevocation == nil {
-		return nil
+	// The RemoteNextRevocation field is optional, as it's possible for a
+	// channel to be closed before we learn of the next unrevoked
+	// revocation point for the remote party. Write a boolen indicating
+	// whether this field is present or not.
+	if err := WriteElements(w, cs.RemoteNextRevocation != nil); err != nil {
+		return err
 	}
 
-	return WriteElements(w, cs.RemoteNextRevocation)
+	// Write the field, if present.
+	if cs.RemoteNextRevocation != nil {
+		if err = WriteElements(w, cs.RemoteNextRevocation); err != nil {
+			return err
+		}
+	}
+
+	// Write whether the channel sync message is present.
+	if err := WriteElements(w, cs.LastChanSyncMsg != nil); err != nil {
+		return err
+	}
+
+	// Write the channel sync message, if present.
+	if cs.LastChanSyncMsg != nil {
+		if err := WriteElements(w, cs.LastChanSyncMsg); err != nil {
+			return err
+		}
+	}
+
+	return nil
 }
 
 func fetchChannelCloseSummary(tx *bolt.Tx,
@@ -2111,15 +2141,19 @@ func deserializeCloseChannelSummary(r io.Reader) (*ChannelCloseSummary, error) {
 
 	// We'll now check to see if the channel close summary was encoded with
 	// any of the additional optional fields.
-	err = ReadElements(r, &c.RemoteCurrentRevocation)
-	switch {
-	case err == io.EOF:
-		return c, nil
+	var hasNewFields bool
+	err = ReadElements(r, &hasNewFields)
+	if err != nil {
+		return nil, err
+	}
 
-	// If we got a non-eof error, then we know there's an actually issue.
-	// Otherwise, it may have been the case that this summary didn't have
-	// the set of optional fields.
-	case err != nil:
+	// If fields are not present, we can return.
+	if !hasNewFields {
+		return c, nil
+	}
+
+	// Otherwise read the new fields.
+	if err := ReadElements(r, &c.RemoteCurrentRevocation); err != nil {
 		return nil, err
 	}
 
@@ -2129,17 +2163,48 @@ func deserializeCloseChannelSummary(r io.Reader) (*ChannelCloseSummary, error) {
 
 	// Finally, we'll attempt to read the next unrevoked commitment point
 	// for the remote party. If we closed the channel before receiving a
-	// funding locked message, then this can be nil. As a result, we'll use
-	// the same technique to read the field, only if there's still data
-	// left in the buffer.
-	err = ReadElements(r, &c.RemoteNextRevocation)
-	if err != nil && err != io.EOF {
-		// If we got a non-eof error, then we know there's an actually
-		// issue. Otherwise, it may have been the case that this
-		// summary didn't have the set of optional fields.
+	// funding locked message then this might not be present. A boolean
+	// indicating whether the field is present will come first.
+	var hasRemoteNextRevocation bool
+	err = ReadElements(r, &hasRemoteNextRevocation)
+	if err != nil {
 		return nil, err
 	}
 
+	// If this field was written, read it.
+	if hasRemoteNextRevocation {
+		err = ReadElements(r, &c.RemoteNextRevocation)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// Check if we have a channel sync message to read.
+	var hasChanSyncMsg bool
+	err = ReadElements(r, &hasChanSyncMsg)
+	if err == io.EOF {
+		return c, nil
+	} else if err != nil {
+		return nil, err
+	}
+
+	// If a chan sync message is present, read it.
+	if hasChanSyncMsg {
+		// We must pass in reference to a lnwire.Message for the codec
+		// to support it.
+		var msg lnwire.Message
+		if err := ReadElements(r, &msg); err != nil {
+			return nil, err
+		}
+
+		chanSync, ok := msg.(*lnwire.ChannelReestablish)
+		if !ok {
+			return nil, errors.New("unable cast db Message to " +
+				"ChannelReestablish")
+		}
+		c.LastChanSyncMsg = chanSync
+	}
+
 	return c, nil
 }
 

channeldb/channel_test.go

@@ -248,10 +248,10 @@ func TestOpenChannelPutGetDelete(t *testing.T) {
 	t.Parallel()
 
 	cdb, cleanUp, err := makeTestDB()
-	defer cleanUp()
 	if err != nil {
 		t.Fatalf("unable to make test database: %v", err)
 	}
+	defer cleanUp()
 
 	// Create the test channel state, then add an additional fake HTLC
 	// before syncing to disk.
@@ -368,10 +368,10 @@ func TestChannelStateTransition(t *testing.T) {
 	t.Parallel()
 
 	cdb, cleanUp, err := makeTestDB()
-	defer cleanUp()
 	if err != nil {
 		t.Fatalf("unable to make test database: %v", err)
 	}
+	defer cleanUp()
 
 	// First create a minimal channel, then perform a full sync in order to
 	// persist the data.

channeldb/db.go

@@ -12,6 +12,7 @@ import (
 	"github.com/btcsuite/btcd/wire"
 	"github.com/coreos/bbolt"
 	"github.com/go-errors/errors"
+	"github.com/lightningnetwork/lnd/lnwire"
 )
 
 const (
@@ -80,6 +81,13 @@ var (
 			number:    6,
 			migration: migratePruneEdgeUpdateIndex,
 		},
+		{
+			// The DB version that migrates the ChannelCloseSummary
+			// to a format where optional fields are indicated with
+			// boolean flags.
+			number:    7,
+			migration: migrateOptionalChannelCloseSummaryFields,
+		},
 	}
 
 	// Big endian is the preferred byte order, due to cursor scans over
@@ -609,6 +617,54 @@ func (d *DB) FetchClosedChannel(chanID *wire.OutPoint) (*ChannelCloseSummary, er
 	return chanSummary, nil
 }
 
+// FetchClosedChannelForID queries for a channel close summary using the
+// channel ID of the channel in question.
+func (d *DB) FetchClosedChannelForID(cid lnwire.ChannelID) (
+	*ChannelCloseSummary, error) {
+
+	var chanSummary *ChannelCloseSummary
+	if err := d.View(func(tx *bolt.Tx) error {
+		closeBucket := tx.Bucket(closedChannelBucket)
+		if closeBucket == nil {
+			return ErrClosedChannelNotFound
+		}
+
+		// The first 30 bytes of the channel ID and outpoint will be
+		// equal.
+		cursor := closeBucket.Cursor()
+		op, c := cursor.Seek(cid[:30])
+
+		// We scan over all possible candidates for this channel ID.
+		for ; op != nil && bytes.Compare(cid[:30], op[:30]) <= 0; op, c = cursor.Next() {
+			var outPoint wire.OutPoint
+			err := readOutpoint(bytes.NewReader(op), &outPoint)
+			if err != nil {
+				return err
+			}
+
+			// If the found outpoint does not correspond to this
+			// channel ID, we continue.
+			if !cid.IsChanPoint(&outPoint) {
+				continue
+			}
+
+			// Deserialize the close summary and return.
+			r := bytes.NewReader(c)
+			chanSummary, err = deserializeCloseChannelSummary(r)
+			if err != nil {
+				return err
+			}
+
+			return nil
+		}
+		return ErrClosedChannelNotFound
+	}); err != nil {
+		return nil, err
+	}
+
+	return chanSummary, nil
+}
+
 // 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

channeldb/db_test.go

@@ -5,6 +5,9 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+
+	"github.com/btcsuite/btcutil"
+	"github.com/lightningnetwork/lnd/lnwire"
 )
 
 func TestOpenWithCreate(t *testing.T) {
@@ -71,3 +74,76 @@ func TestWipe(t *testing.T) {
 			ErrNoClosedChannels, err)
 	}
 }
+
+// TestFetchClosedChannelForID tests that we are able to properly retrieve a
+// ChannelCloseSummary from the DB given a ChannelID.
+func TestFetchClosedChannelForID(t *testing.T) {
+	t.Parallel()
+
+	const numChans = 101
+
+	cdb, cleanUp, err := makeTestDB()
+	if err != nil {
+		t.Fatalf("unable to make test database: %v", err)
+	}
+	defer cleanUp()
+
+	// Create the test channel state, that we will mutate the index of the
+	// funding point.
+	state, err := createTestChannelState(cdb)
+	if err != nil {
+		t.Fatalf("unable to create channel state: %v", err)
+	}
+
+	// Now run through the number of channels, and modify the outpoint index
+	// to create new channel IDs.
+	for i := uint32(0); i < numChans; i++ {
+		// Save the open channel to disk.
+		state.FundingOutpoint.Index = i
+		if err := state.FullSync(); err != nil {
+			t.Fatalf("unable to save and serialize channel "+
+				"state: %v", err)
+		}
+
+		// Close the channel. To make sure we retrieve the correct
+		// summary later, we make them differ in the SettledBalance.
+		closeSummary := &ChannelCloseSummary{
+			ChanPoint:      state.FundingOutpoint,
+			RemotePub:      state.IdentityPub,
+			SettledBalance: btcutil.Amount(500 + i),
+		}
+		if err := state.CloseChannel(closeSummary); err != nil {
+			t.Fatalf("unable to close channel: %v", err)
+		}
+	}
+
+	// Now run though them all again and make sure we are able to retrieve
+	// summaries from the DB.
+	for i := uint32(0); i < numChans; i++ {
+		state.FundingOutpoint.Index = i
+
+		// We calculate the ChannelID and use it to fetch the summary.
+		cid := lnwire.NewChanIDFromOutPoint(&state.FundingOutpoint)
+		fetchedSummary, err := cdb.FetchClosedChannelForID(cid)
+		if err != nil {
+			t.Fatalf("unable to fetch close summary: %v", err)
+		}
+
+		// Make sure we retrieved the correct one by checking the
+		// SettledBalance.
+		if fetchedSummary.SettledBalance != btcutil.Amount(500+i) {
+			t.Fatalf("summaries don't match: expected %v got %v",
+				btcutil.Amount(500+i),
+				fetchedSummary.SettledBalance)
+		}
+	}
+
+	// As a final test we make sure that we get ErrClosedChannelNotFound
+	// for a ChannelID we didn't add to the DB.
+	state.FundingOutpoint.Index++
+	cid := lnwire.NewChanIDFromOutPoint(&state.FundingOutpoint)
+	_, err = cdb.FetchClosedChannelForID(cid)
+	if err != ErrClosedChannelNotFound {
+		t.Fatalf("expected ErrClosedChannelNotFound, instead got: %v", err)
+	}
+}

channeldb/legacy_serialization.go

@@ -0,0 +1,53 @@
+package channeldb
+
+import "io"
+
+// deserializeCloseChannelSummaryV6 reads the v6 database format for
+// ChannelCloseSummary.
+//
+// NOTE: deprecated, only for migration.
+func deserializeCloseChannelSummaryV6(r io.Reader) (*ChannelCloseSummary, error) {
+	c := &ChannelCloseSummary{}
+
+	err := ReadElements(r,
+		&c.ChanPoint, &c.ShortChanID, &c.ChainHash, &c.ClosingTXID,
+		&c.CloseHeight, &c.RemotePub, &c.Capacity, &c.SettledBalance,
+		&c.TimeLockedBalance, &c.CloseType, &c.IsPending,
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	// We'll now check to see if the channel close summary was encoded with
+	// any of the additional optional fields.
+	err = ReadElements(r, &c.RemoteCurrentRevocation)
+	switch {
+	case err == io.EOF:
+		return c, nil
+
+	// If we got a non-eof error, then we know there's an actually issue.
+	// Otherwise, it may have been the case that this summary didn't have
+	// the set of optional fields.
+	case err != nil:
+		return nil, err
+	}
+
+	if err := readChanConfig(r, &c.LocalChanConfig); err != nil {
+		return nil, err
+	}
+
+	// Finally, we'll attempt to read the next unrevoked commitment point
+	// for the remote party. If we closed the channel before receiving a
+	// funding locked message, then this can be nil. As a result, we'll use
+	// the same technique to read the field, only if there's still data
+	// left in the buffer.
+	err = ReadElements(r, &c.RemoteNextRevocation)
+	if err != nil && err != io.EOF {
+		// If we got a non-eof error, then we know there's an actually
+		// issue. Otherwise, it may have been the case that this
+		// summary didn't have the set of optional fields.
+		return nil, err
+	}
+
+	return c, nil
+}

channeldb/migrations.go

@@ -573,3 +573,40 @@ func migratePruneEdgeUpdateIndex(tx *bolt.Tx) error {
 
 	return nil
 }
+
+// migrateOptionalChannelCloseSummaryFields migrates the serialized format of
+// ChannelCloseSummary to a format where optional fields' presence is indicated
+// with boolean markers.
+func migrateOptionalChannelCloseSummaryFields(tx *bolt.Tx) error {
+	closedChanBucket := tx.Bucket(closedChannelBucket)
+	if closedChanBucket == nil {
+		return nil
+	}
+
+	log.Info("Migrating to new closed channel format...")
+	err := closedChanBucket.ForEach(func(chanID, summary []byte) error {
+		r := bytes.NewReader(summary)
+
+		// Read the old (v6) format from the database.
+		c, err := deserializeCloseChannelSummaryV6(r)
+		if err != nil {
+			return err
+		}
+
+		// Serialize using the new format, and put back into the
+		// bucket.
+		var b bytes.Buffer
+		if err := serializeChannelCloseSummary(&b, c); err != nil {
+			return err
+		}
+
+		return closedChanBucket.Put(chanID, b.Bytes())
+	})
+	if err != nil {
+		return fmt.Errorf("unable to update closed channels: %v", err)
+	}
+
+	log.Info("Migration to new closed channel format complete!")
+
+	return nil
+}

channeldb/migrations_test.go

@@ -1,11 +1,17 @@
 package channeldb
 
 import (
+	"bytes"
 	"crypto/sha256"
 	"encoding/binary"
+	"fmt"
+	"reflect"
 	"testing"
 
+	"github.com/btcsuite/btcutil"
 	"github.com/coreos/bbolt"
+	"github.com/davecgh/go-spew/spew"
+	"github.com/go-errors/errors"
 )
 
 // TestPaymentStatusesMigration checks that already completed payments will have
@@ -189,3 +195,276 @@ func TestPaymentStatusesMigration(t *testing.T) {
 		paymentStatusesMigration,
 		false)
 }
+
+// TestMigrateOptionalChannelCloseSummaryFields properly converts a
+// ChannelCloseSummary to the v7 format, where optional fields have their
+// presence indicated with boolean markers.
+func TestMigrateOptionalChannelCloseSummaryFields(t *testing.T) {
+	t.Parallel()
+
+	chanState, err := createTestChannelState(nil)
+	if err != nil {
+		t.Fatalf("unable to create channel state: %v", err)
+	}
+
+	var chanPointBuf bytes.Buffer
+	err = writeOutpoint(&chanPointBuf, &chanState.FundingOutpoint)
+	if err != nil {
+		t.Fatalf("unable to write outpoint: %v", err)
+	}
+
+	chanID := chanPointBuf.Bytes()
+
+	testCases := []struct {
+		closeSummary     *ChannelCloseSummary
+		oldSerialization func(c *ChannelCloseSummary) []byte
+	}{
+		{
+			// A close summary where none of the new fields are
+			// set.
+			closeSummary: &ChannelCloseSummary{
+				ChanPoint:      chanState.FundingOutpoint,
+				ShortChanID:    chanState.ShortChanID(),
+				ChainHash:      chanState.ChainHash,
+				ClosingTXID:    testTx.TxHash(),
+				CloseHeight:    100,
+				RemotePub:      chanState.IdentityPub,
+				Capacity:       chanState.Capacity,
+				SettledBalance: btcutil.Amount(50000),
+				CloseType:      RemoteForceClose,
+				IsPending:      true,
+
+				// The last fields will be unset.
+				RemoteCurrentRevocation: nil,
+				LocalChanConfig:         ChannelConfig{},
+				RemoteNextRevocation:    nil,
+			},
+
+			// In the old format the last field written is the
+			// IsPendingField. It should be converted by adding an
+			// extra boolean marker at the end to indicate that the
+			// remaining fields are not there.
+			oldSerialization: func(cs *ChannelCloseSummary) []byte {
+				var buf bytes.Buffer
+				err := WriteElements(&buf, cs.ChanPoint,
+					cs.ShortChanID, cs.ChainHash,
+					cs.ClosingTXID, cs.CloseHeight,
+					cs.RemotePub, cs.Capacity,
+					cs.SettledBalance, cs.TimeLockedBalance,
+					cs.CloseType, cs.IsPending,
+				)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				// For the old format, these are all the fields
+				// that are written.
+				return buf.Bytes()
+			},
+		},
+		{
+			// A close summary where the new fields are present,
+			// but the optional RemoteNextRevocation field is not
+			// set.
+			closeSummary: &ChannelCloseSummary{
+				ChanPoint:               chanState.FundingOutpoint,
+				ShortChanID:             chanState.ShortChanID(),
+				ChainHash:               chanState.ChainHash,
+				ClosingTXID:             testTx.TxHash(),
+				CloseHeight:             100,
+				RemotePub:               chanState.IdentityPub,
+				Capacity:                chanState.Capacity,
+				SettledBalance:          btcutil.Amount(50000),
+				CloseType:               RemoteForceClose,
+				IsPending:               true,
+				RemoteCurrentRevocation: chanState.RemoteCurrentRevocation,
+				LocalChanConfig:         chanState.LocalChanCfg,
+
+				// RemoteNextRevocation is optional, and here
+				// it is not set.
+				RemoteNextRevocation: nil,
+			},
+
+			// In the old format the last field written is the
+			// LocalChanConfig. This indicates that the optional
+			// RemoteNextRevocation field is not present. It should
+			// be converted by adding boolean markers for all these
+			// fields.
+			oldSerialization: func(cs *ChannelCloseSummary) []byte {
+				var buf bytes.Buffer
+				err := WriteElements(&buf, cs.ChanPoint,
+					cs.ShortChanID, cs.ChainHash,
+					cs.ClosingTXID, cs.CloseHeight,
+					cs.RemotePub, cs.Capacity,
+					cs.SettledBalance, cs.TimeLockedBalance,
+					cs.CloseType, cs.IsPending,
+				)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				err = WriteElements(&buf, cs.RemoteCurrentRevocation)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				err = writeChanConfig(&buf, &cs.LocalChanConfig)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				// RemoteNextRevocation is not written.
+				return buf.Bytes()
+			},
+		},
+		{
+			// A close summary where all fields are present.
+			closeSummary: &ChannelCloseSummary{
+				ChanPoint:               chanState.FundingOutpoint,
+				ShortChanID:             chanState.ShortChanID(),
+				ChainHash:               chanState.ChainHash,
+				ClosingTXID:             testTx.TxHash(),
+				CloseHeight:             100,
+				RemotePub:               chanState.IdentityPub,
+				Capacity:                chanState.Capacity,
+				SettledBalance:          btcutil.Amount(50000),
+				CloseType:               RemoteForceClose,
+				IsPending:               true,
+				RemoteCurrentRevocation: chanState.RemoteCurrentRevocation,
+				LocalChanConfig:         chanState.LocalChanCfg,
+
+				// RemoteNextRevocation is optional, and in
+				// this case we set it.
+				RemoteNextRevocation: chanState.RemoteNextRevocation,
+			},
+
+			// In the old format all the fields are written. It
+			// should be converted by adding boolean markers for
+			// all these fields.
+			oldSerialization: func(cs *ChannelCloseSummary) []byte {
+				var buf bytes.Buffer
+				err := WriteElements(&buf, cs.ChanPoint,
+					cs.ShortChanID, cs.ChainHash,
+					cs.ClosingTXID, cs.CloseHeight,
+					cs.RemotePub, cs.Capacity,
+					cs.SettledBalance, cs.TimeLockedBalance,
+					cs.CloseType, cs.IsPending,
+				)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				err = WriteElements(&buf, cs.RemoteCurrentRevocation)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				err = writeChanConfig(&buf, &cs.LocalChanConfig)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				err = WriteElements(&buf, cs.RemoteNextRevocation)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				return buf.Bytes()
+			},
+		},
+	}
+
+	for _, test := range testCases {
+
+		// Before the migration we must add the old format to the DB.
+		beforeMigrationFunc := func(d *DB) {
+
+			// Get the old serialization format for this test's
+			// close summary, and it to the closed channel bucket.
+			old := test.oldSerialization(test.closeSummary)
+			err = d.Update(func(tx *bolt.Tx) error {
+				closedChanBucket, err := tx.CreateBucketIfNotExists(
+					closedChannelBucket,
+				)
+				if err != nil {
+					return err
+				}
+				return closedChanBucket.Put(chanID, old)
+			})
+			if err != nil {
+				t.Fatalf("unable to add old serialization: %v",
+					err)
+			}
+		}
+
+		// After the migration it should be found in the new format.
+		afterMigrationFunc := func(d *DB) {
+			meta, err := d.FetchMeta(nil)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if meta.DbVersionNumber != 1 {
+				t.Fatal("migration wasn't applied")
+			}
+
+			// We generate the new serialized version, to check
+			// against what is found in the DB.
+			var b bytes.Buffer
+			err = serializeChannelCloseSummary(&b, test.closeSummary)
+			if err != nil {
+				t.Fatalf("unable to serialize: %v", err)
+			}
+			newSerialization := b.Bytes()
+
+			var dbSummary []byte
+			err = d.View(func(tx *bolt.Tx) error {
+				closedChanBucket := tx.Bucket(closedChannelBucket)
+				if closedChanBucket == nil {
+					return errors.New("unable to find bucket")
+				}
+
+				// Get the serialized verision from the DB and
+				// make sure it matches what we expected.
+				dbSummary = closedChanBucket.Get(chanID)
+				if !bytes.Equal(dbSummary, newSerialization) {
+					return fmt.Errorf("unexpected new " +
+						"serialization")
+				}
+				return nil
+			})
+			if err != nil {
+				t.Fatalf("unable to view DB: %v", err)
+			}
+
+			// Finally we fetch the deserialized summary from the
+			// DB and check that it is equal to our original one.
+			dbChannels, err := d.FetchClosedChannels(false)
+			if err != nil {
+				t.Fatalf("unable to fetch closed channels: %v",
+					err)
+			}
+
+			if len(dbChannels) != 1 {
+				t.Fatalf("expected 1 closed channels, found %v",
+					len(dbChannels))
+			}
+
+			dbChan := dbChannels[0]
+			if !reflect.DeepEqual(dbChan, test.closeSummary) {
+				dbChan.RemotePub.Curve = nil
+				test.closeSummary.RemotePub.Curve = nil
+				t.Fatalf("not equal: %v vs %v",
+					spew.Sdump(dbChan),
+					spew.Sdump(test.closeSummary))
+			}
+
+		}
+
+		applyMigration(t,
+			beforeMigrationFunc,
+			afterMigrationFunc,
+			migrateOptionalChannelCloseSummaryFields,
+			false)
+	}
+}

contractcourt/chain_watcher.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/btcsuite/btcd/btcec"
 	"github.com/btcsuite/btcd/chaincfg"
@@ -16,6 +17,16 @@ import (
 	"github.com/lightningnetwork/lnd/lnwallet"
 )
 
+const (
+	// minCommitPointPollTimeout is the minimum time we'll wait before
+	// polling the database for a channel's commitpoint.
+	minCommitPointPollTimeout = 1 * time.Second
+
+	// maxCommitPointPollTimeout is the maximum time we'll wait before
+	// polling the database for a channel's commitpoint.
+	maxCommitPointPollTimeout = 10 * time.Minute
+)
+
 // LocalUnilateralCloseInfo encapsulates all the informnation we need to act
 // on a local force close that gets confirmed.
 type LocalUnilateralCloseInfo struct {
@@ -402,16 +413,38 @@ func (c *chainWatcher) closeObserver(spendNtfn *chainntnfs.SpendEvent) {
 
 			// If we are lucky, the remote peer sent us the correct
 			// commitment point during channel sync, such that we
-			// can sweep our funds.
-			// TODO(halseth): must handle the case where we haven't
-			// yet processed the chan sync message.
-			commitPoint, err := c.cfg.chanState.DataLossCommitPoint()
-			if err != nil {
+			// can sweep our funds. If we cannot find the commit
+			// point, there's not much we can do other than wait
+			// for us to retrieve it. We will attempt to retrieve
+			// it from the peer each time we connect to it.
+			// TODO(halseth): actively initiate re-connection to
+			// the peer?
+			var commitPoint *btcec.PublicKey
+			backoff := minCommitPointPollTimeout
+			for {
+				commitPoint, err = c.cfg.chanState.DataLossCommitPoint()
+				if err == nil {
+					break
+				}
+
 				log.Errorf("Unable to retrieve commitment "+
 					"point for channel(%v) with lost "+
-					"state: %v",
-					c.cfg.chanState.FundingOutpoint, err)
-				return
+					"state: %v. Retrying in %v.",
+					c.cfg.chanState.FundingOutpoint,
+					err, backoff)
+
+				select {
+				// Wait before retrying, with an exponential
+				// backoff.
+				case <-time.After(backoff):
+					backoff = 2 * backoff
+					if backoff > maxCommitPointPollTimeout {
+						backoff = maxCommitPointPollTimeout
+					}
+
+				case <-c.quit:
+					return
+				}
 			}
 
 			log.Infof("Recovered commit point(%x) for "+
@@ -522,6 +555,15 @@ func (c *chainWatcher) dispatchCooperativeClose(commitSpend *chainntnfs.SpendDet
 		LocalChanConfig:         c.cfg.chanState.LocalChanCfg,
 	}
 
+	// Attempt to add a channel sync message to the close summary.
+	chanSync, err := lnwallet.ChanSyncMsg(c.cfg.chanState)
+	if err != nil {
+		log.Errorf("ChannelPoint(%v): unable to create channel sync "+
+			"message: %v", c.cfg.chanState.FundingOutpoint, err)
+	} else {
+		closeSummary.LastChanSyncMsg = chanSync
+	}
+
 	// Create a summary of all the information needed to handle the
 	// cooperative closure.
 	closeInfo := &CooperativeCloseInfo{
@@ -590,6 +632,15 @@ func (c *chainWatcher) dispatchLocalForceClose(
 		closeSummary.TimeLockedBalance += htlcValue
 	}
 
+	// Attempt to add a channel sync message to the close summary.
+	chanSync, err := lnwallet.ChanSyncMsg(c.cfg.chanState)
+	if err != nil {
+		log.Errorf("ChannelPoint(%v): unable to create channel sync "+
+			"message: %v", c.cfg.chanState.FundingOutpoint, err)
+	} else {
+		closeSummary.LastChanSyncMsg = chanSync
+	}
+
 	// With the event processed, we'll now notify all subscribers of the
 	// event.
 	closeInfo := &LocalUnilateralCloseInfo{
@@ -749,6 +800,15 @@ func (c *chainWatcher) dispatchContractBreach(spendEvent *chainntnfs.SpendDetail
 		LocalChanConfig:         c.cfg.chanState.LocalChanCfg,
 	}
 
+	// Attempt to add a channel sync message to the close summary.
+	chanSync, err := lnwallet.ChanSyncMsg(c.cfg.chanState)
+	if err != nil {
+		log.Errorf("ChannelPoint(%v): unable to create channel sync "+
+			"message: %v", c.cfg.chanState.FundingOutpoint, err)
+	} else {
+		closeSummary.LastChanSyncMsg = chanSync
+	}
+
 	if err := c.cfg.chanState.CloseChannel(&closeSummary); err != nil {
 		return err
 	}

htlcswitch/link.go

@@ -516,7 +516,7 @@ func (l *channelLink) syncChanStates() error {
 	// First, we'll generate our ChanSync message to send to the other
 	// side. Based on this message, the remote party will decide if they
 	// need to retransmit any data or not.
-	localChanSyncMsg, err := l.channel.ChanSyncMsg()
+	localChanSyncMsg, err := lnwallet.ChanSyncMsg(l.channel.State())
 	if err != nil {
 		return fmt.Errorf("unable to generate chan sync message for "+
 			"ChannelPoint(%v)", l.channel.ChannelPoint())
@@ -789,9 +789,6 @@ func (l *channelLink) htlcManager() {
 			// We failed syncing the commit chains, probably
 			// because the remote has lost state. We should force
 			// close the channel.
-			// TODO(halseth): store sent chanSync message to
-			// database, such that it can be resent to peer in case
-			// it tries to sync the channel again.
 			case err == lnwallet.ErrCommitSyncRemoteDataLoss:
 				fallthrough
 

lnwallet/channel.go

@@ -3445,19 +3445,22 @@ func (lc *LightningChannel) ProcessChanSyncMsg(
 //      it.
 //   3. We didn't get the last RevokeAndAck message they sent, so they'll
 //      re-send it.
-func (lc *LightningChannel) ChanSyncMsg() (*lnwire.ChannelReestablish, error) {
+func ChanSyncMsg(c *channeldb.OpenChannel) (*lnwire.ChannelReestablish, error) {
+	c.Lock()
+	defer c.Unlock()
+
 	// The remote commitment height that we'll send in the
 	// ChannelReestablish message is our current commitment height plus
 	// one. If the receiver thinks that our commitment height is actually
 	// *equal* to this value, then they'll re-send the last commitment that
 	// they sent but we never fully processed.
-	localHeight := lc.localCommitChain.tip().height
+	localHeight := c.LocalCommitment.CommitHeight
 	nextLocalCommitHeight := localHeight + 1
 
 	// The second value we'll send is the height of the remote commitment
 	// from our PoV. If the receiver thinks that their height is actually
 	// *one plus* this value, then they'll re-send their last revocation.
-	remoteChainTipHeight := lc.remoteCommitChain.tail().height
+	remoteChainTipHeight := c.RemoteCommitment.CommitHeight
 
 	// If this channel has undergone a commitment update, then in order to
 	// prove to the remote party our knowledge of their prior commitment
@@ -3465,7 +3468,7 @@ func (lc *LightningChannel) ChanSyncMsg() (*lnwire.ChannelReestablish, error) {
 	// remote party sent.
 	var lastCommitSecret [32]byte
 	if remoteChainTipHeight != 0 {
-		remoteSecret, err := lc.channelState.RevocationStore.LookUp(
+		remoteSecret, err := c.RevocationStore.LookUp(
 			remoteChainTipHeight - 1,
 		)
 		if err != nil {
@@ -3476,7 +3479,7 @@ func (lc *LightningChannel) ChanSyncMsg() (*lnwire.ChannelReestablish, error) {
 
 	// Additionally, we'll send over the current unrevoked commitment on
 	// our local commitment transaction.
-	currentCommitSecret, err := lc.channelState.RevocationProducer.AtIndex(
+	currentCommitSecret, err := c.RevocationProducer.AtIndex(
 		localHeight,
 	)
 	if err != nil {
@@ -3485,7 +3488,7 @@ func (lc *LightningChannel) ChanSyncMsg() (*lnwire.ChannelReestablish, error) {
 
 	return &lnwire.ChannelReestablish{
 		ChanID: lnwire.NewChanIDFromOutPoint(
-			&lc.channelState.FundingOutpoint,
+			&c.FundingOutpoint,
 		),
 		NextLocalCommitHeight:  nextLocalCommitHeight,
 		RemoteCommitTailHeight: remoteChainTipHeight,
@@ -5112,6 +5115,15 @@ func NewUnilateralCloseSummary(chanState *channeldb.OpenChannel, signer Signer,
 		LocalChanConfig:         chanState.LocalChanCfg,
 	}
 
+	// Attempt to add a channel sync message to the close summary.
+	chanSync, err := ChanSyncMsg(chanState)
+	if err != nil {
+		walletLog.Errorf("ChannelPoint(%v): unable to create channel sync "+
+			"message: %v", chanState.FundingOutpoint, err)
+	} else {
+		closeSummary.LastChanSyncMsg = chanSync
+	}
+
 	return &UnilateralCloseSummary{
 		SpendDetail:         commitSpend,
 		ChannelCloseSummary: closeSummary,
@@ -6187,7 +6199,7 @@ func (lc *LightningChannel) IsPending() bool {
 	return lc.channelState.IsPending
 }
 
-// State provides access to the channel's internal state for testing.
+// State provides access to the channel's internal state.
 func (lc *LightningChannel) State() *channeldb.OpenChannel {
 	return lc.channelState
 }

lnwallet/channel_test.go

@@ -2433,7 +2433,7 @@ func assertNoChanSyncNeeded(t *testing.T, aliceChannel *LightningChannel,
 
 	_, _, line, _ := runtime.Caller(1)
 
-	aliceChanSyncMsg, err := aliceChannel.ChanSyncMsg()
+	aliceChanSyncMsg, err := ChanSyncMsg(aliceChannel.channelState)
 	if err != nil {
 		t.Fatalf("line #%v: unable to produce chan sync msg: %v",
 			line, err)
@@ -2448,7 +2448,7 @@ func assertNoChanSyncNeeded(t *testing.T, aliceChannel *LightningChannel,
 			"instead wants to send: %v", line, spew.Sdump(bobMsgsToSend))
 	}
 
-	bobChanSyncMsg, err := bobChannel.ChanSyncMsg()
+	bobChanSyncMsg, err := ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("line #%v: unable to produce chan sync msg: %v",
 			line, err)
@@ -2681,11 +2681,11 @@ func TestChanSyncOweCommitment(t *testing.T) {
 	// Bob doesn't get this message so upon reconnection, they need to
 	// synchronize. Alice should conclude that she owes Bob a commitment,
 	// while Bob should think he's properly synchronized.
-	aliceSyncMsg, err := aliceChannel.ChanSyncMsg()
+	aliceSyncMsg, err := ChanSyncMsg(aliceChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
-	bobSyncMsg, err := bobChannel.ChanSyncMsg()
+	bobSyncMsg, err := ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -2995,11 +2995,11 @@ func TestChanSyncOweRevocation(t *testing.T) {
 	// If we fetch the channel sync messages at this state, then Alice
 	// should report that she owes Bob a revocation message, while Bob
 	// thinks they're fully in sync.
-	aliceSyncMsg, err := aliceChannel.ChanSyncMsg()
+	aliceSyncMsg, err := ChanSyncMsg(aliceChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
-	bobSyncMsg, err := bobChannel.ChanSyncMsg()
+	bobSyncMsg, err := ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -3164,11 +3164,11 @@ func TestChanSyncOweRevocationAndCommit(t *testing.T) {
 	// If we now attempt to resync, then Alice should conclude that she
 	// doesn't need any further updates, while Bob concludes that he needs
 	// to re-send both his revocation and commit sig message.
-	aliceSyncMsg, err := aliceChannel.ChanSyncMsg()
+	aliceSyncMsg, err := ChanSyncMsg(aliceChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
-	bobSyncMsg, err := bobChannel.ChanSyncMsg()
+	bobSyncMsg, err := ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -3374,11 +3374,11 @@ func TestChanSyncOweRevocationAndCommitForceTransition(t *testing.T) {
 	// Now if we attempt to synchronize states at this point, Alice should
 	// detect that she owes nothing, while Bob should re-send both his
 	// RevokeAndAck as well as his commitment message.
-	aliceSyncMsg, err := aliceChannel.ChanSyncMsg()
+	aliceSyncMsg, err := ChanSyncMsg(aliceChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
-	bobSyncMsg, err := bobChannel.ChanSyncMsg()
+	bobSyncMsg, err := ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -3573,11 +3573,11 @@ func TestChanSyncFailure(t *testing.T) {
 	// assertLocalDataLoss checks that aliceOld and bobChannel detects that
 	// Alice has lost state during sync.
 	assertLocalDataLoss := func(aliceOld *LightningChannel) {
-		aliceSyncMsg, err := aliceOld.ChanSyncMsg()
+		aliceSyncMsg, err := ChanSyncMsg(aliceOld.channelState)
 		if err != nil {
 			t.Fatalf("unable to produce chan sync msg: %v", err)
 		}
-		bobSyncMsg, err := bobChannel.ChanSyncMsg()
+		bobSyncMsg, err := ChanSyncMsg(bobChannel.channelState)
 		if err != nil {
 			t.Fatalf("unable to produce chan sync msg: %v", err)
 		}
@@ -3629,7 +3629,7 @@ func TestChanSyncFailure(t *testing.T) {
 	// If we remove the recovery options from Bob's message, Alice cannot
 	// tell if she lost state, since Bob might be lying. She still should
 	// be able to detect that chains cannot be synced.
-	bobSyncMsg, err := bobChannel.ChanSyncMsg()
+	bobSyncMsg, err := ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -3643,7 +3643,7 @@ func TestChanSyncFailure(t *testing.T) {
 	// If Bob lies about the NextLocalCommitHeight, making it greater than
 	// what Alice expect, she cannot tell for sure whether she lost state,
 	// but should detect the desync.
-	bobSyncMsg, err = bobChannel.ChanSyncMsg()
+	bobSyncMsg, err = ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -3656,7 +3656,7 @@ func TestChanSyncFailure(t *testing.T) {
 
 	// If Bob's NextLocalCommitHeight is lower than what Alice expects, Bob
 	// probably lost state.
-	bobSyncMsg, err = bobChannel.ChanSyncMsg()
+	bobSyncMsg, err = ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -3669,7 +3669,7 @@ func TestChanSyncFailure(t *testing.T) {
 
 	// If Alice and Bob's states are in sync, but Bob is sending the wrong
 	// LocalUnrevokedCommitPoint, Alice should detect this.
-	bobSyncMsg, err = bobChannel.ChanSyncMsg()
+	bobSyncMsg, err = ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -3695,7 +3695,7 @@ func TestChanSyncFailure(t *testing.T) {
 	// when there's a pending remote commit.
 	halfAdvance()
 
-	bobSyncMsg, err = bobChannel.ChanSyncMsg()
+	bobSyncMsg, err = ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -3785,11 +3785,11 @@ func TestChannelRetransmissionFeeUpdate(t *testing.T) {
 	// Bob doesn't get this message so upon reconnection, they need to
 	// synchronize. Alice should conclude that she owes Bob a commitment,
 	// while Bob should think he's properly synchronized.
-	aliceSyncMsg, err := aliceChannel.ChanSyncMsg()
+	aliceSyncMsg, err := ChanSyncMsg(aliceChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
-	bobSyncMsg, err := bobChannel.ChanSyncMsg()
+	bobSyncMsg, err := ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to produce chan sync msg: %v", err)
 	}
@@ -4015,11 +4015,11 @@ func TestChanSyncInvalidLastSecret(t *testing.T) {
 	}
 
 	// Next, we'll produce the ChanSync messages for both parties.
-	aliceChanSync, err := aliceChannel.ChanSyncMsg()
+	aliceChanSync, err := ChanSyncMsg(aliceChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to generate chan sync msg: %v", err)
 	}
-	bobChanSync, err := bobChannel.ChanSyncMsg()
+	bobChanSync, err := ChanSyncMsg(bobChannel.channelState)
 	if err != nil {
 		t.Fatalf("unable to generate chan sync msg: %v", err)
 	}

peer.go

@@ -836,11 +836,11 @@ func newChanMsgStream(p *peer, cid lnwire.ChannelID) *msgStream {
 		fmt.Sprintf("Update stream for ChannelID(%x) exiting", cid[:]),
 		1000,
 		func(msg lnwire.Message) {
-			_, isChanSycMsg := msg.(*lnwire.ChannelReestablish)
+			_, isChanSyncMsg := msg.(*lnwire.ChannelReestablish)
 
 			// If this is the chanSync message, then we'll deliver
 			// it immediately to the active link.
-			if !isChanSycMsg {
+			if !isChanSyncMsg {
 				// We'll send a message to the funding manager
 				// and wait iff an active funding process for
 				// this channel hasn't yet completed.  We do
@@ -874,9 +874,36 @@ func newChanMsgStream(p *peer, cid lnwire.ChannelID) *msgStream {
 			// active goroutine dedicated to this channel.
 			if chanLink == nil {
 				link, err := p.server.htlcSwitch.GetLink(cid)
-				if err != nil {
-					peerLog.Errorf("recv'd update for unknown "+
-						"channel %v from %v", cid, p)
+				switch {
+
+				// If we failed to find the link in question,
+				// and the message received was a channel sync
+				// message, then this might be a peer trying to
+				// resync closed channel. In this case we'll
+				// try to resend our last channel sync message,
+				// such that the peer can recover funds from
+				// the closed channel.
+				case err != nil && isChanSyncMsg:
+					peerLog.Debugf("Unable to find "+
+						"link(%v) to handle channel "+
+						"sync, attempting to resend "+
+						"last ChanSync message", cid)
+
+					err := p.resendChanSyncMsg(cid)
+					if err != nil {
+						// TODO(halseth): send error to
+						// peer?
+						peerLog.Errorf(
+							"resend failed: %v",
+							err,
+						)
+					}
+					return
+
+				case err != nil:
+					peerLog.Errorf("recv'd update for "+
+						"unknown channel %v from %v: "+
+						"%v", cid, p, err)
 					return
 				}
 				chanLink = link
@@ -2143,6 +2170,35 @@ func (p *peer) sendInitMsg() error {
 	return p.writeMessage(msg)
 }
 
+// resendChanSyncMsg will attempt to find a channel sync message for the closed
+// channel and resend it to our peer.
+func (p *peer) resendChanSyncMsg(cid lnwire.ChannelID) error {
+	// Check if we have any channel sync messages stored for this channel.
+	c, err := p.server.chanDB.FetchClosedChannelForID(cid)
+	if err != nil {
+		return fmt.Errorf("unable to fetch channel sync messages for "+
+			"peer %v: %v", p, err)
+	}
+
+	if c.LastChanSyncMsg == nil {
+		return fmt.Errorf("no chan sync message stored for channel %v",
+			cid)
+	}
+
+	peerLog.Debugf("Re-sending channel sync message for channel %v to "+
+		"peer %v", cid, p)
+
+	if err := p.SendMessage(true, c.LastChanSyncMsg); err != nil {
+		return fmt.Errorf("Failed resending channel sync "+
+			"message to peer %v: %v", p, err)
+	}
+
+	peerLog.Debugf("Re-sent channel sync message for channel %v to peer "+
+		"%v", cid, p)
+
+	return nil
+}
+
 // SendMessage sends a variadic number of message to remote peer. The first
 // argument denotes if the method should block until the message has been sent
 // to the remote peer.