chanstate: move OpenChannel behind store interface

channeldb/channel_test.go

@@ -414,7 +414,6 @@ func createTestChannelState(t *testing.T, cdb *ChannelStateDB) *OpenChannel {
 		RevocationProducer:      producer,
 		RevocationStore:         store,
 		Db:                      cdb,
-		Packager:                NewChannelPackager(chanID),
 		FundingTxn:              channels.TestFundingTx,
 		ThawHeight:              uint32(defaultPendingHeight),
 		InitialLocalBalance:     lnwire.MilliSatoshi(9000),
@@ -879,7 +878,7 @@ func TestChannelStateTransition(t *testing.T) {
 
 	// The state number recovered from the tail of the revocation log
 	// should be identical to this current state.
-	logTailHeight, err := channel.revocationLogTailCommitHeight()
+	logTailHeight, err := cdb.revocationLogTailCommitHeight(channel)
 	require.NoError(t, err, "unable to retrieve log")
 	if logTailHeight != oldRemoteCommit.CommitHeight {
 		t.Fatal("update number doesn't match")
@@ -922,7 +921,7 @@ func TestChannelStateTransition(t *testing.T) {
 
 	// Once again, state number recovered from the tail of the revocation
 	// log should be identical to this current state.
-	logTailHeight, err = channel.revocationLogTailCommitHeight()
+	logTailHeight, err = cdb.revocationLogTailCommitHeight(channel)
 	require.NoError(t, err, "unable to retrieve log")
 	if logTailHeight != oldRemoteCommit.CommitHeight {
 		t.Fatal("update number doesn't match")
@@ -939,7 +938,9 @@ func TestChannelStateTransition(t *testing.T) {
 	}
 
 	// At this point, we should have 2 forwarding packages added.
-	fwdPkgs := loadFwdPkgs(t, cdb.backend, channel.Packager)
+	fwdPkgs := loadFwdPkgs(
+		t, cdb.backend, NewChannelPackager(channel.ShortChanID()),
+	)
 	require.Len(t, fwdPkgs, 2, "wrong number of forwarding packages")
 
 	// Now attempt to delete the channel from the database.
@@ -974,7 +975,9 @@ func TestChannelStateTransition(t *testing.T) {
 	}
 
 	// All forwarding packages of this channel has been deleted too.
-	fwdPkgs = loadFwdPkgs(t, cdb.backend, channel.Packager)
+	fwdPkgs = loadFwdPkgs(
+		t, cdb.backend, NewChannelPackager(channel.ShortChanID()),
+	)
 	require.Empty(t, fwdPkgs, "no forwarding packages should exist")
 }
 
@@ -1424,16 +1427,6 @@ func TestRefresh(t *testing.T) {
 			"updated before refreshing short_chan_id")
 	}
 
-	// Now that the receiver's short channel id has been updated, check to
-	// ensure that the channel packager's source has been updated as well.
-	// This ensures that the packager will read and write to buckets
-	// corresponding to the new short chan id, instead of the prior.
-	if state.Packager.(*ChannelPackager).source != chanOpenLoc {
-		t.Fatalf("channel packager source was not updated: want %v, "+
-			"got %v", chanOpenLoc,
-			state.Packager.(*ChannelPackager).source)
-	}
-
 	// Now, refresh the state of the pending channel.
 	err = pendingChannel.Refresh()
 	require.NoError(t, err, "unable to refresh short_chan_id")
@@ -1446,16 +1439,6 @@ func TestRefresh(t *testing.T) {
 			pendingChannel.ShortChanID())
 	}
 
-	// Check to ensure that the _other_ OpenChannel channel packager's
-	// source has also been updated after the refresh. This ensures that the
-	// other packagers will read and write to buckets corresponding to the
-	// updated short chan id.
-	if pendingChannel.Packager.(*ChannelPackager).source != chanOpenLoc {
-		t.Fatalf("channel packager source was not updated: want %v, "+
-			"got %v", chanOpenLoc,
-			pendingChannel.Packager.(*ChannelPackager).source)
-	}
-
 	// Check to ensure that this channel is no longer pending and this field
 	// is up to date.
 	if pendingChannel.IsPending {
@@ -1559,7 +1542,7 @@ func TestCloseInitiator(t *testing.T) {
 				if !dbChans[0].HasChanStatus(status) {
 					t.Fatalf("expected channel to have "+
 						"status: %v, has status: %v",
-						status, dbChans[0].chanStatus)
+						status, dbChans[0].ChanStatus())
 				}
 			}
 		})
@@ -1642,9 +1625,8 @@ func TestHasChanStatus(t *testing.T) {
 		test := test
 
 		t.Run(test.name, func(t *testing.T) {
-			c := &OpenChannel{
-				chanStatus: test.status,
-			}
+			c := &OpenChannel{}
+			c.SetChannelStatusForStore(test.status)
 
 			for status, expHas := range test.expHas {
 				has := c.HasChanStatus(status)

channeldb/chanstate_assertions.go

@@ -0,0 +1,7 @@
+package channeldb
+
+import "github.com/lightningnetwork/lnd/chanstate"
+
+// Compile-time assertion that ChannelStateDB satisfies the channel-state store
+// contract while the KV implementation still lives in channeldb.
+var _ chanstate.Store = (*ChannelStateDB)(nil)

channeldb/close_channel_test.go

@@ -15,10 +15,12 @@ import (
 // revocationLogBucket of the given channel. The helper navigates the raw KV
 // tree so the test does not depend on the higher-level commit-chain
 // machinery.
-func writeTestRevlogEntries(t *testing.T, ch *OpenChannel, n int) {
+func writeTestRevlogEntries(t *testing.T, cdb *ChannelStateDB,
+	ch *OpenChannel, n int) {
+
 	t.Helper()
 
-	err := kvdb.Update(ch.Db.backend, func(tx kvdb.RwTx) error {
+	err := kvdb.Update(cdb.backend, func(tx kvdb.RwTx) error {
 		openChanBkt := tx.ReadWriteBucket(openChannelBucket)
 		require.NotNil(t, openChanBkt, "openChannelBucket missing")
 
@@ -56,11 +58,13 @@ func writeTestRevlogEntries(t *testing.T, ch *OpenChannel, n int) {
 
 // writeTestForwardingPackages writes n empty forwarding packages for the
 // given channel using distinct remote commitment heights.
-func writeTestForwardingPackages(t *testing.T, ch *OpenChannel, n int) {
+func writeTestForwardingPackages(t *testing.T, cdb *ChannelStateDB,
+	ch *OpenChannel, n int) {
+
 	t.Helper()
 
 	packager := NewChannelPackager(ch.ShortChanID())
-	err := kvdb.Update(ch.Db.backend, func(tx kvdb.RwTx) error {
+	err := kvdb.Update(cdb.backend, func(tx kvdb.RwTx) error {
 		for i := range n {
 			pkg := NewFwdPkg(
 				ch.ShortChanID(), uint64(i), nil, nil,
@@ -78,11 +82,13 @@ func writeTestForwardingPackages(t *testing.T, ch *OpenChannel, n int) {
 // countRevlogEntries returns the number of entries in the revocationLogBucket
 // for the given channel, or -1 if the channel bucket no longer exists in
 // openChannelBucket.
-func countRevlogEntries(t *testing.T, ch *OpenChannel) int {
+func countRevlogEntries(t *testing.T, cdb *ChannelStateDB,
+	ch *OpenChannel) int {
+
 	t.Helper()
 
 	count := -1
-	err := kvdb.View(ch.Db.backend, func(tx kvdb.RTx) error {
+	err := kvdb.View(cdb.backend, func(tx kvdb.RTx) error {
 		openChanBkt := tx.ReadBucket(openChannelBucket)
 		if openChanBkt == nil {
 			return nil
@@ -202,8 +208,8 @@ func TestCloseChannelTombstoneWritePath(t *testing.T) {
 
 	const numRevlogEntries = 5
 	const numFwdPkgs = 3
-	writeTestRevlogEntries(t, ch, numRevlogEntries)
-	writeTestForwardingPackages(t, ch, numFwdPkgs)
+	writeTestRevlogEntries(t, cdb, ch, numRevlogEntries)
+	writeTestForwardingPackages(t, cdb, ch, numFwdPkgs)
 
 	closeChannelForTest(t, cdb, ch)
 
@@ -224,7 +230,7 @@ func TestCloseChannelTombstoneWritePath(t *testing.T) {
 	require.Equal(t, ch.FundingOutpoint, closeSummary.ChanPoint)
 
 	// Bulk state preserved on disk — tombstoning's whole point.
-	require.Equal(t, numRevlogEntries, countRevlogEntries(t, ch))
+	require.Equal(t, numRevlogEntries, countRevlogEntries(t, cdb, ch))
 
 	packager := NewChannelPackager(ch.ShortChanID())
 	var fwdPkgs []*FwdPkg
@@ -281,7 +287,7 @@ func TestCloseChannelTombstoneRemovesFromOpenScans(t *testing.T) {
 	ch2 := createTestChannel(t, cdb, openChannelOption())
 
 	const numRevlogEntries = 5
-	writeTestRevlogEntries(t, ch1, numRevlogEntries)
+	writeTestRevlogEntries(t, cdb, ch1, numRevlogEntries)
 
 	openChans, err := cdb.FetchAllChannels()
 	require.NoError(t, err)
@@ -313,7 +319,7 @@ func TestCloseChannelTombstoneRemovesFromOpenScans(t *testing.T) {
 
 	// The bulk historical state stays put — that is the whole point of
 	// the tombstone path on these backends.
-	require.Equal(t, numRevlogEntries, countRevlogEntries(t, ch1))
+	require.Equal(t, numRevlogEntries, countRevlogEntries(t, cdb, ch1))
 
 	// The outpoint index for ch1 must flip to closed; ch2's stays open.
 	require.Equal(t, outpointClosed, readOutpointStatus(
@@ -380,14 +386,14 @@ func TestCloseChannelSync(t *testing.T) {
 	ch := createTestChannel(t, cdb, openChannelOption())
 
 	const numRevlogEntries = 4
-	writeTestRevlogEntries(t, ch, numRevlogEntries)
-	writeTestForwardingPackages(t, ch, 3)
+	writeTestRevlogEntries(t, cdb, ch, numRevlogEntries)
+	writeTestForwardingPackages(t, cdb, ch, 3)
 
 	closeChannelForTest(t, cdb, ch)
 
 	// The synchronous path wipes the chanBucket inline, so
 	// countRevlogEntries must report -1 (bucket is gone, not just empty).
-	require.Equal(t, -1, countRevlogEntries(t, ch),
+	require.Equal(t, -1, countRevlogEntries(t, cdb, ch),
 		"channel bucket must be deleted after sync close")
 
 	// Forwarding packages are wiped inline.

channeldb/db.go

@@ -32,6 +32,7 @@
 	"github.com/lightningnetwork/lnd/channeldb/migration34"
 	"github.com/lightningnetwork/lnd/channeldb/migration35"
 	"github.com/lightningnetwork/lnd/channeldb/migration_01_to_11"
+	"github.com/lightningnetwork/lnd/chanstate"
 	"github.com/lightningnetwork/lnd/clock"
 	graphdb "github.com/lightningnetwork/lnd/graph/db"
 	"github.com/lightningnetwork/lnd/invoices"
@@ -777,10 +778,7 @@
 }
 
 // ChanCount is used by the server in determining access control.
-type ChanCount struct {
-	HasOpenOrClosedChan bool
-	PendingOpenCount    uint64
-}
+type ChanCount = chanstate.ChanCount
 
 // FetchPermAndTempPeers returns a map where the key is the remote node's
 // public key and the value is a struct that has a tally of the pending-open
@@ -1678,17 +1676,8 @@
 }
 
 // ChannelShell is a shell of a channel that is meant to be used for channel
-// recovery purposes. It contains a minimal OpenChannel instance along with
-// addresses for that target node.
-type ChannelShell struct {
-	// NodeAddrs the set of addresses that this node has known to be
-	// reachable at in the past.
-	NodeAddrs []net.Addr
-
-	// Chan is a shell of an OpenChannel, it contains only the items
-	// required to restore the channel on disk.
-	Chan *OpenChannel
-}
+// recovery purposes.
+type ChannelShell = chanstate.ChannelShell
 
 // RestoreChannelShells is a method that allows the caller to reconstruct the
 // state of an OpenChannel from the ChannelShell. We'll attempt to write the
@@ -1705,15 +1694,17 @@
 			// been restored, this will signal to other sub-systems
 			// to not attempt to use the channel as if it was a
 			// regular one.
-			channel.chanStatus |= ChanStatusRestored
+			channel.SetChannelStatusForStore(
+				channel.ChannelStatusForStore() | ChanStatusRestored,
+			)
 
 			// First, we'll attempt to create a new open channel
 			// and link node for this channel. If the channel
 			// already exists, then in order to ensure this method
 			// is idempotent, we'll continue to the next step.
 			channel.Db = c
 			err := syncNewChannel(
-				tx, channel, channelShell.NodeAddrs,
+				tx, channel, channelShell.NodeAddrs, c.backend,
 			)
 			if err != nil {
 				return err

channeldb/db_test.go

@@ -307,33 +307,35 @@ func genRandomChannelShell() (*ChannelShell, error) {
 		CsvDelay: uint16(rand.Int63()),
 	}
 
+	channel := &OpenChannel{
+		ChainHash:       rev,
+		FundingOutpoint: chanPoint,
+		ShortChannelID: lnwire.NewShortChanIDFromInt(
+			uint64(rand.Int63()),
+		),
+		IdentityPub: pub,
+		LocalChanCfg: ChannelConfig{
+			CommitmentParams: commitParams,
+			PaymentBasePoint: keychain.KeyDescriptor{
+				KeyLocator: keychain.KeyLocator{
+					Family: keychain.KeyFamily(rand.Int63()),
+					Index:  uint32(rand.Int63()),
+				},
+			},
+		},
+		RemoteCurrentRevocation: pub,
+		IsPending:               false,
+		RevocationStore:         shachain.NewRevocationStore(),
+		RevocationProducer:      shaChainProducer,
+	}
+	channel.SetChannelStatusForStore(chanStatus)
+
 	return &ChannelShell{
 		NodeAddrs: []net.Addr{&net.TCPAddr{
 			IP:   net.ParseIP("127.0.0.1"),
 			Port: 18555,
 		}},
-		Chan: &OpenChannel{
-			chanStatus:      chanStatus,
-			ChainHash:       rev,
-			FundingOutpoint: chanPoint,
-			ShortChannelID: lnwire.NewShortChanIDFromInt(
-				uint64(rand.Int63()),
-			),
-			IdentityPub: pub,
-			LocalChanCfg: ChannelConfig{
-				CommitmentParams: commitParams,
-				PaymentBasePoint: keychain.KeyDescriptor{
-					KeyLocator: keychain.KeyLocator{
-						Family: keychain.KeyFamily(rand.Int63()),
-						Index:  uint32(rand.Int63()),
-					},
-				},
-			},
-			RemoteCurrentRevocation: pub,
-			IsPending:               false,
-			RevocationStore:         shachain.NewRevocationStore(),
-			RevocationProducer:      shaChainProducer,
-		},
+		Chan: channel,
 	}, nil
 }
 
@@ -403,7 +405,7 @@ func TestRestoreChannelShells(t *testing.T) {
 	}
 	if !nodeChans[0].HasChanStatus(ChanStatusRestored) {
 		t.Fatalf("node has wrong status flags: %v",
-			nodeChans[0].chanStatus)
+			nodeChans[0].ChanStatus())
 	}
 
 	// We should also be able to find the channel if we query for it