Merge pull request #9610 from lightningnetwork/rbf-staging

multi: integrate rbf changes from staging branch
This commit is contained in:
Olaoluwa Osuntokun 2025-03-19 15:01:23 -05:00 committed by GitHub
commit ea050d06f0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
47 changed files with 5868 additions and 2761 deletions

View file

@ -4104,6 +4104,34 @@ func (c *OpenChannel) AbsoluteThawHeight() (uint32, error) {
return c.ThawHeight, nil
}
// DeriveHeightHint derives the block height for the channel opening.
func (c *OpenChannel) DeriveHeightHint() uint32 {
// As a height hint, we'll try to use the opening height, but if the
// channel isn't yet open, then we'll use the height it was broadcast
// at. This may be an unconfirmed zero-conf channel.
heightHint := c.ShortChanID().BlockHeight
if heightHint == 0 {
heightHint = c.BroadcastHeight()
}
// Since no zero-conf state is stored in a channel backup, the below
// logic will not be triggered for restored, zero-conf channels. Set
// the height hint for zero-conf channels.
if c.IsZeroConf() {
if c.ZeroConfConfirmed() {
// If the zero-conf channel is confirmed, we'll use the
// confirmed SCID's block height.
heightHint = c.ZeroConfRealScid().BlockHeight
} else {
// The zero-conf channel is unconfirmed. We'll need to
// use the FundingBroadcastHeight.
heightHint = c.BroadcastHeight()
}
}
return heightHint
}
func putChannelCloseSummary(tx kvdb.RwTx, chanID []byte,
summary *ChannelCloseSummary, lastChanState *OpenChannel) error {

View file

@ -281,7 +281,7 @@ func newChainWatcher(cfg chainWatcherConfig) (*chainWatcher, error) {
}
// Get the channel opening block height.
heightHint := deriveHeightHint(chanState)
heightHint := chanState.DeriveHeightHint()
// We'll register for a notification to be dispatched if the funding
// output is spent.
@ -1328,34 +1328,6 @@ func deriveFundingPkScript(chanState *channeldb.OpenChannel) ([]byte, error) {
return fundingPkScript, nil
}
// deriveHeightHint derives the block height for the channel opening.
func deriveHeightHint(chanState *channeldb.OpenChannel) uint32 {
// As a height hint, we'll try to use the opening height, but if the
// channel isn't yet open, then we'll use the height it was broadcast
// at. This may be an unconfirmed zero-conf channel.
heightHint := chanState.ShortChanID().BlockHeight
if heightHint == 0 {
heightHint = chanState.BroadcastHeight()
}
// Since no zero-conf state is stored in a channel backup, the below
// logic will not be triggered for restored, zero-conf channels. Set
// the height hint for zero-conf channels.
if chanState.IsZeroConf() {
if chanState.ZeroConfConfirmed() {
// If the zero-conf channel is confirmed, we'll use the
// confirmed SCID's block height.
heightHint = chanState.ZeroConfRealScid().BlockHeight
} else {
// The zero-conf channel is unconfirmed. We'll need to
// use the FundingBroadcastHeight.
heightHint = chanState.BroadcastHeight()
}
}
return heightHint
}
// handleCommitSpend takes a spending tx of the funding output and handles the
// channel close based on the closure type.
func (c *chainWatcher) handleCommitSpend(

View file

@ -93,6 +93,27 @@
# New Features
* Add support for [archiving channel backup](https://github.com/lightningnetwork/lnd/pull/9232)
in a designated folder which allows for easy referencing in the future. A new
config is added `disable-backup-archive`, with default set to false, to
determine if previous channel backups should be archived or not.
## Protocol Updates
* `lnd` now [supports the new RBF cooperative close
flow](https://github.com/lightningnetwork/lnd/pull/9610). Unlike the old flow,
this version now uses RBF to enable either side to increase their fee rate using
their _own_ channel funds. This removes the old "negotiation" logic that could
fail, with a version where either side can increase the fee on their coop close
transaction using their channel balance.
This new feature can be activated with a new config flag:
`--protocol.rbf-coop-close`.
With this new co-op close type, users can issue multiple `lncli closechannnel`
commands with increasing fee rates to use RBF to bump an existing signed co-op
close transaction.
* [Support](https://github.com/lightningnetwork/lnd/pull/8390) for
[experimental endorsement](https://github.com/lightning/blips/pull/27)
signal relay was added. This signal has *no impact* on routing, and
@ -106,10 +127,7 @@
initial historical sync may be blocked due to a race condition in handling the
syncer's internal state.
* Add support for [archiving channel backup](https://github.com/lightningnetwork/lnd/pull/9232)
in a designated folder which allows for easy referencing in the future. A new
config is added `disable-backup-archive`, with default set to false, to
determine if previous channel backups should be archived or not.
* [The max fee rate](https://github.com/lightningnetwork/lnd/pull/9491) is now
respected when a coop close is initiated. Before the max fee rate would only
@ -432,6 +450,7 @@ The underlying functionality between those two options remain the same.
* Keagan McClelland
* Nishant Bansal
* Oliver Gugger
* Olaoluwa Osuntokun
* Pins
* Viktor Tigerström
* Yong Yu

View file

@ -103,4 +103,8 @@ var defaultSetDesc = setDesc{
lnwire.ExperimentalEndorsementOptional: {
SetNodeAnn: {}, // N
},
lnwire.RbfCoopCloseOptionalStaging: {
SetInit: {}, // I
SetNodeAnn: {}, // N
},
}

View file

@ -73,6 +73,10 @@ type Config struct {
// forwarding experimental endorsement.
NoExperimentalEndorsement bool
// NoRbfCoopClose unsets any bits that signal support for using RBF for
// coop close.
NoRbfCoopClose bool
// CustomFeatures is a set of custom features to advertise in each
// set.
CustomFeatures map[Set][]lnwire.FeatureBit
@ -209,11 +213,13 @@ func newManager(cfg Config, desc setDesc) (*Manager, error) {
raw.Unset(lnwire.SimpleTaprootOverlayChansOptional)
raw.Unset(lnwire.SimpleTaprootOverlayChansRequired)
}
if cfg.NoExperimentalEndorsement {
raw.Unset(lnwire.ExperimentalEndorsementOptional)
raw.Unset(lnwire.ExperimentalEndorsementRequired)
}
if cfg.NoRbfCoopClose {
raw.Unset(lnwire.RbfCoopCloseOptionalStaging)
}
for _, custom := range cfg.CustomFeatures[set] {
if custom > set.Maximum() {

View file

@ -196,7 +196,7 @@ type FlushHookID uint64
// LinkDirection is used to query and change any link state on a per-direction
// basis.
type LinkDirection bool
type LinkDirection = bool
const (
// Incoming is the direction from the remote peer to our node.

View file

@ -2,6 +2,7 @@ package htlcswitch
import (
"bytes"
"context"
"errors"
"fmt"
"math/rand"
@ -125,6 +126,9 @@ type ChanClose struct {
// Err is used by request creator to receive request execution error.
Err chan error
// Ctx is a context linked to the lifetime of the caller.
Ctx context.Context //nolint:containedctx
}
// Config defines the configuration for the service. ALL elements within the
@ -1413,7 +1417,7 @@ func (s *Switch) teardownCircuit(pkt *htlcPacket) error {
// targetFeePerKw parameter should be the ideal fee-per-kw that will be used as
// a starting point for close negotiation. The deliveryScript parameter is an
// optional parameter which sets a user specified script to close out to.
func (s *Switch) CloseLink(chanPoint *wire.OutPoint,
func (s *Switch) CloseLink(ctx context.Context, chanPoint *wire.OutPoint,
closeType contractcourt.ChannelCloseType,
targetFeePerKw, maxFee chainfee.SatPerKWeight,
deliveryScript lnwire.DeliveryAddress) (chan interface{}, chan error) {
@ -1427,9 +1431,10 @@ func (s *Switch) CloseLink(chanPoint *wire.OutPoint,
ChanPoint: chanPoint,
Updates: updateChan,
TargetFeePerKw: targetFeePerKw,
MaxFee: maxFee,
DeliveryScript: deliveryScript,
Err: errChan,
MaxFee: maxFee,
Ctx: ctx,
}
select {

View file

@ -0,0 +1,51 @@
package input
import (
"testing"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcd/txscript"
"github.com/stretchr/testify/require"
)
// FuzzScriptIsOpReturn fuzzes the ScriptIsOpReturn function.
func FuzzScriptIsOpReturn(f *testing.F) {
// Seed the corpus with some representative inputs.
f.Add([]byte{})
f.Add([]byte{txscript.OP_RETURN})
// Use our canonical ScriptBuilder to produce a valid OP_RETURN script.
builder := txscript.NewScriptBuilder()
builder.AddOp(txscript.OP_RETURN)
builder.AddData([]byte("valid data"))
script, err := builder.Script()
require.NoError(f, err)
f.Add(script)
// An example of a script that does not start with OP_RETURN.
f.Add([]byte{txscript.OP_DUP, txscript.OP_RETURN})
f.Fuzz(func(t *testing.T, pkScript []byte) {
result := ScriptIsOpReturn(pkScript)
var expected bool
if len(script) == 0 || script[0] != txscript.OP_RETURN {
expected = false
} else {
scriptClass, _, _, err := txscript.ExtractPkScriptAddrs(
pkScript, &chaincfg.MainNetParams,
)
if err != nil {
t.Fatalf("unable to extract pk script "+
"addresses: %v", err)
}
expected = scriptClass == txscript.NullDataTy
}
if result != expected {
t.Fatalf("for script %x, expected ScriptIsOpReturn=%v;"+
" got %v", script, expected, result)
}
})
}

View file

@ -3239,3 +3239,36 @@ func ComputeCommitmentPoint(commitSecret []byte) *btcec.PublicKey {
_, pubKey := btcec.PrivKeyFromBytes(commitSecret)
return pubKey
}
// ScriptIsOpReturn returns true if the passed script is an OP_RETURN script.
//
// Lifted from the txscript package:
// https://github.com/btcsuite/btcd/blob/cc26860b40265e1332cca8748c5dbaf3c81cc094/txscript/standard.go#L493-L526.
//
//nolint:ll
func ScriptIsOpReturn(script []byte) bool {
// A null script is of the form:
// OP_RETURN <optional data>
//
// Thus, it can either be a single OP_RETURN or an OP_RETURN followed by
// a data push up to MaxDataCarrierSize bytes.
// The script can't possibly be a null data script if it doesn't start
// with OP_RETURN. Fail fast to avoid more work below.
if len(script) < 1 || script[0] != txscript.OP_RETURN {
return false
}
// Single OP_RETURN.
if len(script) == 1 {
return true
}
// OP_RETURN followed by data push up to MaxDataCarrierSize bytes.
tokenizer := txscript.MakeScriptTokenizer(0, script[1:])
return tokenizer.Next() && tokenizer.Done() &&
(txscript.IsSmallInt(tokenizer.Opcode()) ||
tokenizer.Opcode() <= txscript.OP_PUSHDATA4) &&
len(tokenizer.Data()) <= txscript.MaxDataCarrierSize
}

View file

@ -678,6 +678,10 @@ var allTestCases = []*lntest.TestCase{
Name: "access perm",
TestFunc: testAccessPerm,
},
{
Name: "rbf coop close",
TestFunc: testCoopCloseRbf,
},
}
// appendPrefixed is used to add a prefix to each test name in the subtests

View file

@ -0,0 +1,133 @@
package itest
import (
"github.com/btcsuite/btcd/btcutil"
"github.com/lightningnetwork/lnd/lntest"
"github.com/lightningnetwork/lnd/lnwallet/chainfee"
"github.com/stretchr/testify/require"
)
func testCoopCloseRbf(ht *lntest.HarnessTest) {
rbfCoopFlags := []string{"--protocol.rbf-coop-close"}
// Set the fee estimate to 1sat/vbyte. This ensures that our manually
// initiated RBF attempts will always be successful.
ht.SetFeeEstimate(250)
ht.SetFeeEstimateWithConf(250, 6)
// To kick things off, we'll create two new nodes, then fund them with
// enough coins to make a 50/50 channel.
cfgs := [][]string{rbfCoopFlags, rbfCoopFlags}
params := lntest.OpenChannelParams{
Amt: btcutil.Amount(1000000),
PushAmt: btcutil.Amount(1000000 / 2),
}
chanPoints, nodes := ht.CreateSimpleNetwork(cfgs, params)
alice, bob := nodes[0], nodes[1]
chanPoint := chanPoints[0]
// Now that both sides are active with a funded channel, we can kick
// off the test.
//
// To start, we'll have Alice try to close the channel, with a fee rate
// of 5 sat/byte.
aliceFeeRate := chainfee.SatPerVByte(5)
aliceCloseStream, aliceCloseUpdate := ht.CloseChannelAssertPending(
alice, chanPoint, false,
lntest.WithCoopCloseFeeRate(aliceFeeRate),
lntest.WithLocalTxNotify(),
)
// Confirm that this new update was at 5 sat/vb.
alicePendingUpdate := aliceCloseUpdate.GetClosePending()
require.NotNil(ht, aliceCloseUpdate)
require.Equal(
ht, int64(aliceFeeRate), alicePendingUpdate.FeePerVbyte,
)
require.True(ht, alicePendingUpdate.LocalCloseTx)
// Now, we'll have Bob attempt to RBF the close transaction with a
// higher fee rate, double that of Alice's.
bobFeeRate := aliceFeeRate * 2
bobCloseStream, bobCloseUpdate := ht.CloseChannelAssertPending(
bob, chanPoint, false, lntest.WithCoopCloseFeeRate(bobFeeRate),
lntest.WithLocalTxNotify(),
)
// Confirm that this new update was at 10 sat/vb.
bobPendingUpdate := bobCloseUpdate.GetClosePending()
require.NotNil(ht, bobCloseUpdate)
require.Equal(ht, bobPendingUpdate.FeePerVbyte, int64(bobFeeRate))
require.True(ht, bobPendingUpdate.LocalCloseTx)
var err error
// Alice should've also received a similar update that Bob has
// increased the closing fee rate to 10 sat/vb with his settled funds.
aliceCloseUpdate, err = ht.ReceiveCloseChannelUpdate(aliceCloseStream)
require.NoError(ht, err)
alicePendingUpdate = aliceCloseUpdate.GetClosePending()
require.NotNil(ht, aliceCloseUpdate)
require.Equal(ht, alicePendingUpdate.FeePerVbyte, int64(bobFeeRate))
require.False(ht, alicePendingUpdate.LocalCloseTx)
// We'll now attempt to make a fee update that increases Alice's fee
// rate by 6 sat/vb, which should be rejected as it is too small of an
// increase for the RBF rules. The RPC API however will return the new
// fee. We'll skip the mempool check here as it won't make it in.
aliceRejectedFeeRate := aliceFeeRate + 1
_, aliceCloseUpdate = ht.CloseChannelAssertPending(
alice, chanPoint, false,
lntest.WithCoopCloseFeeRate(aliceRejectedFeeRate),
lntest.WithLocalTxNotify(), lntest.WithSkipMempoolCheck(),
)
alicePendingUpdate = aliceCloseUpdate.GetClosePending()
require.NotNil(ht, aliceCloseUpdate)
require.Equal(
ht, alicePendingUpdate.FeePerVbyte,
int64(aliceRejectedFeeRate),
)
require.True(ht, alicePendingUpdate.LocalCloseTx)
_, err = ht.ReceiveCloseChannelUpdate(bobCloseStream)
require.NoError(ht, err)
// We'll now attempt a fee update that we can't actually pay for. This
// will actually show up as an error to the remote party.
aliceRejectedFeeRate = 100_000
_, _ = ht.CloseChannelAssertPending(
alice, chanPoint, false,
lntest.WithCoopCloseFeeRate(aliceRejectedFeeRate),
lntest.WithLocalTxNotify(),
lntest.WithExpectedErrString("cannot pay for fee"),
)
// At this point, we'll have Alice+Bob reconnect so we can ensure that
// we can continue to do RBF bumps even after a reconnection.
ht.DisconnectNodes(alice, bob)
ht.ConnectNodes(alice, bob)
// Next, we'll have Alice double that fee rate again to 20 sat/vb.
aliceFeeRate = bobFeeRate * 2
aliceCloseStream, aliceCloseUpdate = ht.CloseChannelAssertPending(
alice, chanPoint, false,
lntest.WithCoopCloseFeeRate(aliceFeeRate),
lntest.WithLocalTxNotify(),
)
alicePendingUpdate = aliceCloseUpdate.GetClosePending()
require.NotNil(ht, aliceCloseUpdate)
require.Equal(
ht, alicePendingUpdate.FeePerVbyte, int64(aliceFeeRate),
)
require.True(ht, alicePendingUpdate.LocalCloseTx)
// To conclude, we'll mine a block which should now confirm Alice's
// version of the coop close transaction.
block := ht.MineBlocksAndAssertNumTxes(1, 1)[0]
// Both Alice and Bob should trigger a final close update to signal the
// closing transaction has confirmed.
aliceClosingTxid := ht.WaitForChannelCloseEvent(aliceCloseStream)
ht.AssertTxInBlock(block, aliceClosingTxid)
}

View file

@ -1,6 +1,7 @@
package itest
import (
"fmt"
"testing"
"github.com/btcsuite/btcd/btcutil"
@ -10,6 +11,7 @@ import (
"github.com/lightningnetwork/lnd/lnrpc/routerrpc"
"github.com/lightningnetwork/lnd/lnrpc/walletrpc"
"github.com/lightningnetwork/lnd/lntest"
"github.com/lightningnetwork/lnd/lntest/node"
"github.com/lightningnetwork/lnd/lntest/wait"
"github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwallet/chainfee"
@ -23,23 +25,47 @@ import (
// will have the receiver settle the invoice and observe that the channel gets
// torn down after settlement.
func testCoopCloseWithHtlcs(ht *lntest.HarnessTest) {
ht.Run("no restart", func(t *testing.T) {
tt := ht.Subtest(t)
coopCloseWithHTLCs(tt)
})
rbfCoopFlags := []string{"--protocol.rbf-coop-close"}
ht.Run("with restart", func(t *testing.T) {
tt := ht.Subtest(t)
coopCloseWithHTLCsWithRestart(tt)
})
for _, isRbf := range []bool{true, false} {
testName := fmt.Sprintf("no restart is_rbf=%v", isRbf)
ht.Run(testName, func(t *testing.T) {
tt := ht.Subtest(t)
var flags []string
if isRbf {
flags = rbfCoopFlags
}
alice := ht.NewNodeWithCoins("Alice", flags)
bob := ht.NewNodeWithCoins("bob", flags)
coopCloseWithHTLCs(tt, alice, bob)
})
}
for _, isRbf := range []bool{true, false} {
testName := fmt.Sprintf("with restart is_rbf=%v", isRbf)
ht.Run(testName, func(t *testing.T) {
tt := ht.Subtest(t)
var flags []string
if isRbf {
flags = rbfCoopFlags
}
alice := ht.NewNodeWithCoins("Alice", flags)
bob := ht.NewNodeWithCoins("bob", flags)
coopCloseWithHTLCsWithRestart(tt, alice, bob)
})
}
}
// coopCloseWithHTLCs tests the basic coop close scenario which occurs when one
// channel party initiates a channel shutdown while an HTLC is still pending on
// the channel.
func coopCloseWithHTLCs(ht *lntest.HarnessTest) {
alice := ht.NewNodeWithCoins("Alice", nil)
bob := ht.NewNodeWithCoins("bob", nil)
func coopCloseWithHTLCs(ht *lntest.HarnessTest, alice, bob *node.HarnessNode) {
ht.ConnectNodes(alice, bob)
// Here we set up a channel between Alice and Bob, beginning with a
@ -131,9 +157,9 @@ func coopCloseWithHTLCs(ht *lntest.HarnessTest) {
// is still pending on the channel but this time it ensures that the shutdown
// process continues as expected even if a channel re-establish happens after
// one party has already initiated the shutdown.
func coopCloseWithHTLCsWithRestart(ht *lntest.HarnessTest) {
alice := ht.NewNodeWithCoins("Alice", nil)
bob := ht.NewNodeWithCoins("bob", nil)
func coopCloseWithHTLCsWithRestart(ht *lntest.HarnessTest, alice,
bob *node.HarnessNode) {
ht.ConnectNodes(alice, bob)
// Open a channel between Alice and Bob with the balance split equally.
@ -219,8 +245,8 @@ func coopCloseWithHTLCsWithRestart(ht *lntest.HarnessTest) {
}, defaultTimeout)
require.NoError(ht, err)
// Wait for the close tx to be in the Mempool and then mine 6 blocks
// to confirm the close.
// Wait for the close tx to be in the Mempool and then mine 6 blocks to
// confirm the close.
closingTx := ht.AssertClosingTxInMempool(
chanPoint, lnrpc.CommitmentType_LEGACY,
)

View file

@ -109,7 +109,11 @@ func breachRetributionTestCase(ht *lntest.HarnessTest,
// broadcasting his current channel state. This is actually the
// commitment transaction of a prior *revoked* state, so he'll soon
// feel the wrath of Carol's retribution.
_, breachTXID := ht.CloseChannelAssertPending(bob, chanPoint, true)
_, breachCloseUpd := ht.CloseChannelAssertPending(bob, chanPoint, true)
closeUpd := breachCloseUpd.GetClosePending()
require.NotNil(ht, closeUpd)
breachTXID, err := chainhash.NewHash(closeUpd.Txid)
require.NoError(ht, err)
// Here, Carol sees Bob's breach transaction in the mempool, but is
// waiting for it to confirm before continuing her retribution. We
@ -122,13 +126,13 @@ func breachRetributionTestCase(ht *lntest.HarnessTest,
// update, then ensure that the closing transaction was included in the
// block.
block := ht.MineBlocksAndAssertNumTxes(1, 1)[0]
ht.AssertTxInBlock(block, breachTXID)
ht.AssertTxInBlock(block, *breachTXID)
// Construct to_remote output which pays to Bob. Based on the output
// ordering, the first output in this breach tx is the to_remote
// output.
toRemoteOp := wire.OutPoint{
Hash: breachTXID,
Hash: *breachTXID,
Index: 0,
}
@ -151,7 +155,7 @@ func breachRetributionTestCase(ht *lntest.HarnessTest,
// Assert that all the inputs of this transaction are spending outputs
// generated by Bob's breach transaction above.
for _, txIn := range justiceTx.TxIn {
require.Equal(ht, breachTXID, txIn.PreviousOutPoint.Hash,
require.Equal(ht, *breachTXID, txIn.PreviousOutPoint.Hash,
"justice tx not spending commitment utxo")
}
@ -296,7 +300,7 @@ func revokedCloseRetributionZeroValueRemoteOutputCase(ht *lntest.HarnessTest,
// broadcasting her current channel state. This is actually the
// commitment transaction of a prior *revoked* state, so she'll soon
// feel the wrath of Dave's retribution.
stream, closeTxID := ht.CloseChannelAssertPending(
stream, closeUpdate := ht.CloseChannelAssertPending(
carol, chanPoint, true,
)
@ -310,9 +314,11 @@ func revokedCloseRetributionZeroValueRemoteOutputCase(ht *lntest.HarnessTest,
// state and continues exacting justice after his node restarts.
ht.RestartNode(dave)
closeTxID := closeUpdate.GetClosePending().Txid
// The breachTXID should match the above closeTxID.
breachTXID := ht.WaitForChannelCloseEvent(stream)
require.EqualValues(ht, breachTXID, closeTxID)
require.EqualValues(ht, breachTXID[:], closeTxID)
// Construct to_local output which pays to Dave. Based on the output
// ordering, the first output in this breach tx is the to_local
@ -543,9 +549,13 @@ func revokedCloseRetributionRemoteHodlCase(ht *lntest.HarnessTest,
// broadcasting her current channel state. This is actually the
// commitment transaction of a prior *revoked* state, so she'll soon
// feel the wrath of Dave's retribution.
closeUpdates, closeTxID := ht.CloseChannelAssertPending(
closeUpdates, closeUpd := ht.CloseChannelAssertPending(
carol, chanPoint, true,
)
pendingCloseUpd := closeUpd.GetClosePending()
require.NotNil(ht, pendingCloseUpd)
closeTxID, err := chainhash.NewHash(pendingCloseUpd.Txid)
require.NoError(ht, err)
// Generate a single block to mine the breach transaction.
block := ht.MineBlocksAndAssertNumTxes(1, 1)[0]
@ -593,7 +603,7 @@ func revokedCloseRetributionRemoteHodlCase(ht *lntest.HarnessTest,
return nil, errNotFound
}
err := wait.NoError(func() error {
err = wait.NoError(func() error {
txid, err := findJusticeTx()
if err != nil {
return err

View file

@ -5,6 +5,7 @@ import (
"time"
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/lightningnetwork/lnd/contractcourt"
"github.com/lightningnetwork/lnd/fn/v2"
@ -1963,8 +1964,8 @@ func testBumpForceCloseFee(ht *lntest.HarnessTest) {
ht.FundCoinsP2TR(btcutil.SatoshiPerBitcoin, alice)
// Alice force closes the channel which has no HTLCs at stake.
_, closingTxID := ht.CloseChannelAssertPending(alice, chanPoint, true)
require.NotNil(ht, closingTxID)
_, closeUpdates := ht.CloseChannelAssertPending(alice, chanPoint, true)
require.NotNil(ht, closeUpdates)
// Alice should see one waiting close channel.
ht.AssertNumWaitingClose(alice, 1)
@ -1974,7 +1975,10 @@ func testBumpForceCloseFee(ht *lntest.HarnessTest) {
ht.AssertNumPendingSweeps(alice, 2)
// Calculate the commitment tx fee rate.
closingTx := ht.AssertTxInMempool(closingTxID)
pendingClose := closeUpdates.GetClosePending()
closeTxid, err := chainhash.NewHash(pendingClose.Txid)
require.NoError(ht, err)
closingTx := ht.AssertTxInMempool(*closeTxid)
require.NotNil(ht, closingTx)
// The default commitment fee for anchor channels is capped at 2500

View file

@ -491,7 +491,7 @@ func testRevokedCloseRetributionAltruistWatchtowerCase(ht *lntest.HarnessTest,
// broadcasting his current channel state. This is actually the
// commitment transaction of a prior *revoked* state, so he'll soon
// feel the wrath of Dave's retribution.
closeUpdates, closeTxID := ht.CloseChannelAssertPending(
closeUpdates, pendingClose := ht.CloseChannelAssertPending(
carol, chanPoint, true,
)
@ -504,7 +504,8 @@ func testRevokedCloseRetributionAltruistWatchtowerCase(ht *lntest.HarnessTest,
ht.AssertTxInBlock(block, breachTXID)
// The breachTXID should match the above closeTxID.
require.EqualValues(ht, breachTXID, closeTxID)
closeTxID := pendingClose.GetClosePending().Txid
require.EqualValues(ht, breachTXID[:], closeTxID)
// Query the mempool for Dave's justice transaction, this should be
// broadcast as Carol's contract breaching transaction gets confirmed

View file

@ -35,6 +35,10 @@ type ProtocolOptions struct {
// the experimental taproot overlay chan type.
TaprootOverlayChans bool `long:"simple-taproot-overlay-chans" description:"if set, then lnd will create and accept requests for channels using the taproot overlay commitment type"`
// RbfCoopClose should be set if we want to signal that we support for
// the new experimental RBF coop close feature.
RbfCoopClose bool `long:"rbf-coop-close" description:"if set, then lnd will signal that it supports the new RBF based coop close protocol"`
// NoAnchors should be set if we don't want to support opening or accepting
// channels having the anchor commitment type.
NoAnchors bool `long:"no-anchors" description:"disable support for anchor commitments"`

View file

@ -41,6 +41,10 @@ type ProtocolOptions struct {
// TODO(halseth): transition itests to anchors instead!
Anchors bool `long:"anchors" description:"enable support for anchor commitments"`
// RbfCoopClose should be set if we want to signal that we support for
// the new experimental RBF coop close feature.
RbfCoopClose bool `long:"rbf-coop-close" description:"if set, then lnd will signal that it supports the new RBF based coop close protocol"`
// ScriptEnforcedLease enables script enforced commitments for channel
// leases.
//

File diff suppressed because it is too large Load diff

View file

@ -2167,6 +2167,8 @@ message CloseStatusUpdate {
message PendingUpdate {
bytes txid = 1;
uint32 output_index = 2;
int64 fee_per_vbyte = 3;
bool local_close_tx = 4;
}
message InstantUpdate {

View file

@ -6739,6 +6739,13 @@
"output_index": {
"type": "integer",
"format": "int64"
},
"fee_per_vbyte": {
"type": "string",
"format": "int64"
},
"local_close_tx": {
"type": "boolean"
}
}
},

View file

@ -1208,14 +1208,79 @@ func (h *HarnessTest) OpenChannelAssertErr(srcNode, destNode *node.HarnessNode,
"error returned, want %v, got %v", expectedErr, err)
}
// closeChannelOpts holds the options for closing a channel.
type closeChannelOpts struct {
feeRate fn.Option[chainfee.SatPerVByte]
// localTxOnly is a boolean indicating if we should only attempt to
// consume close pending notifications for the local transaction.
localTxOnly bool
// skipMempoolCheck is a boolean indicating if we should skip the normal
// mempool check after a coop close.
skipMempoolCheck bool
// errString is an expected error. If this is non-blank, then we'll
// assert that the coop close wasn't possible, and returns an error that
// contains this err string.
errString string
}
// CloseChanOpt is a functional option to modify the way we close a channel.
type CloseChanOpt func(*closeChannelOpts)
// WithCoopCloseFeeRate is a functional option to set the fee rate for a coop
// close attempt.
func WithCoopCloseFeeRate(rate chainfee.SatPerVByte) CloseChanOpt {
return func(o *closeChannelOpts) {
o.feeRate = fn.Some(rate)
}
}
// WithLocalTxNotify is a functional option to indicate that we should only
// notify for the local txn. This is useful for the RBF coop close type, as
// it'll notify for both local and remote txns.
func WithLocalTxNotify() CloseChanOpt {
return func(o *closeChannelOpts) {
o.localTxOnly = true
}
}
// WithSkipMempoolCheck is a functional option to indicate that we should skip
// the mempool check. This can be used when a coop close iteration may not
// result in a newly broadcast transaction.
func WithSkipMempoolCheck() CloseChanOpt {
return func(o *closeChannelOpts) {
o.skipMempoolCheck = true
}
}
// WithExpectedErrString is a functional option that can be used to assert that
// an error occurs during the coop close process.
func WithExpectedErrString(errString string) CloseChanOpt {
return func(o *closeChannelOpts) {
o.errString = errString
}
}
// defaultCloseOpts returns the set of default close options.
func defaultCloseOpts() *closeChannelOpts {
return &closeChannelOpts{}
}
// CloseChannelAssertPending attempts to close the channel indicated by the
// passed channel point, initiated by the passed node. Once the CloseChannel
// rpc is called, it will consume one event and assert it's a close pending
// event. In addition, it will check that the closing tx can be found in the
// mempool.
func (h *HarnessTest) CloseChannelAssertPending(hn *node.HarnessNode,
cp *lnrpc.ChannelPoint,
force bool) (rpc.CloseChanClient, chainhash.Hash) {
cp *lnrpc.ChannelPoint, force bool,
opts ...CloseChanOpt) (rpc.CloseChanClient, *lnrpc.CloseStatusUpdate) {
closeOpts := defaultCloseOpts()
for _, optFunc := range opts {
optFunc(closeOpts)
}
// Calls the rpc to close the channel.
closeReq := &lnrpc.CloseChannelRequest{
@ -1224,10 +1289,9 @@ func (h *HarnessTest) CloseChannelAssertPending(hn *node.HarnessNode,
NoWait: true,
}
// For coop close, we use a default confg target of 6.
if !force {
closeReq.TargetConf = 6
}
closeOpts.feeRate.WhenSome(func(feeRate chainfee.SatPerVByte) {
closeReq.SatPerVbyte = uint64(feeRate)
})
var (
stream rpc.CloseChanClient
@ -1242,25 +1306,52 @@ func (h *HarnessTest) CloseChannelAssertPending(hn *node.HarnessNode,
_, err = h.ReceiveCloseChannelUpdate(stream)
require.NoError(h, err, "close channel update got error: %v", err)
event, err = h.ReceiveCloseChannelUpdate(stream)
if err != nil {
h.Logf("Test: %s, close channel got error: %v",
h.manager.currentTestCase, err)
var closeTxid *chainhash.Hash
for {
event, err = h.ReceiveCloseChannelUpdate(stream)
if err != nil {
h.Logf("Test: %s, close channel got error: %v",
h.manager.currentTestCase, err)
}
if err != nil && closeOpts.errString == "" {
require.NoError(h, err, "retry closing channel failed")
} else if err != nil && closeOpts.errString != "" {
require.ErrorContains(h, err, closeOpts.errString)
return nil, nil
}
pendingClose, ok := event.Update.(*lnrpc.CloseStatusUpdate_ClosePending) //nolint:ll
require.Truef(h, ok, "expected channel close "+
"update, instead got %v", pendingClose)
if !pendingClose.ClosePending.LocalCloseTx &&
closeOpts.localTxOnly {
continue
}
notifyRate := pendingClose.ClosePending.FeePerVbyte
if closeOpts.localTxOnly &&
notifyRate != int64(closeReq.SatPerVbyte) {
continue
}
closeTxid, err = chainhash.NewHash(
pendingClose.ClosePending.Txid,
)
require.NoErrorf(h, err, "unable to decode closeTxid: %v",
pendingClose.ClosePending.Txid)
break
}
require.NoError(h, err, "retry closing channel failed")
pendingClose, ok := event.Update.(*lnrpc.CloseStatusUpdate_ClosePending)
require.Truef(h, ok, "expected channel close update, instead got %v",
pendingClose)
if !closeOpts.skipMempoolCheck {
// Assert the closing tx is in the mempool.
h.miner.AssertTxInMempool(*closeTxid)
}
closeTxid, err := chainhash.NewHash(pendingClose.ClosePending.Txid)
require.NoErrorf(h, err, "unable to decode closeTxid: %v",
pendingClose.ClosePending.Txid)
// Assert the closing tx is in the mempool.
h.miner.AssertTxInMempool(*closeTxid)
return stream, *closeTxid
return stream, event
}
// CloseChannel attempts to coop close a non-anchored channel identified by the

View file

@ -534,8 +534,8 @@ func (h HarnessTest) WaitForChannelCloseEvent(
require.NoError(h, err)
resp, ok := event.Update.(*lnrpc.CloseStatusUpdate_ChanClose)
require.Truef(h, ok, "expected channel open update, instead got %v",
resp)
require.Truef(h, ok, "expected channel close update, instead got %v",
event.Update)
txid, err := chainhash.NewHash(resp.ChanClose.ClosingTxid)
require.NoErrorf(h, err, "wrong format found in closing txid: %v",

View file

@ -0,0 +1,219 @@
# RBF Co-op Close State Machine
## Abstract
The Lightning Network Daemon (lnd) implements a cooperative channel closing
mechanism that allows two connected peers to negotiate a mutually agreed-upon
closing transaction. This document outlines the state machine that governs the
cooperative closing process, defining the various states a channel progresses
through from active operation to final closure.
Cooperative closing is distinct from force closing in that it requires both
channel participants to agree on the final channel state and closing
transaction, resulting in lower fees and immediate fund availability compared to
unilateral channel closures.
## State Machine Overview
The cooperative closing state machine manages the lifecycle of a channel from
the moment a closing request is initiated until the closing transaction is
confirmed on the blockchain. The state transitions ensure proper negotiation of
closing fees, validation of signatures, and handling of protocol violations.
### High-Level Flow
1. An active channel receives a shutdown request (locally or remotely)
2. Channel stops accepting new HTLCs and begins flushing existing ones
3. Once flushed, nodes begin negotiating closing transaction fee
4. Upon agreement, nodes sign and broadcast the transaction
5. **RBF Iteration (optional)**: While waiting for confirmation, either node can
propose a fee increase via Replace-By-Fee (RBF), restarting the negotiation
process with higher fees
6. Process completes when transaction confirms on-chain
## States and Transitions
```mermaid
---
title: Co-Op Close V2
---
stateDiagram-v2
state CoopCloseV2 {
[*] --> ChannelActive
ChannelActive --> ShutdownPending: send_shutdown
ChannelActive --> ShutdownPending: shutdown_received
ShutdownPending --> ChannelFlushing: shutdown_received
ShutdownPending --> ChannelFlushing: shutdown_complete
ChannelFlushing --> ClosingNegotiation: channel_flushed
state ClosingNegotiation {
state LocalSide {
[*] --> LocalCloseStart
LocalCloseStart --> LocalOfferSent: send_offer
LocalOfferSent --> ClosePending: local_sig_received
}
state RemoteSide {
[*] --> RemoteCloseStart
RemoteCloseStart --> ClosePending: offer_received
}
CloseErr --> LocalCloseStart: send_offer_rbf
CloseErr --> RemoteCloseStart: offer_received_rbf
LocalCloseStart --> CloseErr: protocol_violation
RemoteCloseStart --> CloseErr: protocol_violation
ClosePending --> LocalCloseStart: send_offer_rbf
ClosePending --> RemoteCloseStart: offer_received_rbf
}
ClosingNegotiation --> CloseFin: txn_confirmation
}
```
### ChannelActive
The initial state where the channel is fully operational.
- **Transitions**:
- `send_shutdown``ShutdownPending` (Local node initiates closing process)
- `shutdown_received``ShutdownPending` (Remote node initiates closing process)
### ShutdownPending
The channel has entered the closing process but is still processing existing HTLCs.
- **Transitions**:
- `shutdown_received``ChannelFlushing` (When both parties have exchanged
shutdown messages)
- `shutdown_complete``ChannelFlushing` (When local shutdown processing is
complete)
### ChannelFlushing
The channel no longer accepts new HTLCs and is waiting for existing HTLCs to
resolve.
- **Transitions**:
- `channel_flushed``ClosingNegotiation` (When all HTLCs are settled)
### ClosingNegotiation
This composite state encompasses multiple substates related to the negotiation
of the closing transaction fee. It's divided into local and remote operation
flows.
#### LocalSide
- **LocalCloseStart**: Ready to initiate or respond to closing negotiation from
local perspective.
- **Transitions**:
- `send_offer``LocalOfferSent` (Local node proposes a closing fee)
- `protocol_violation``CloseErr` (Invalid message or state detected)
- **LocalOfferSent**: Local node has sent a closing offer and is awaiting
confirmation.
- **Transitions**:
- `local_sig_received``ClosePending` (Local signature validated and
accepted)
#### RemoteSide
- **RemoteCloseStart**: Ready to process closing negotiation initiated by remote
node.
- **Transitions**:
- `offer_received``ClosePending` (Remote node proposal received and
accepted)
- `protocol_violation``CloseErr` (Invalid message or state detected)
#### Shared States
- **ClosePending**: Closing transaction has been negotiated, signed by both
parties, and broadcast to the network. Waiting for on-chain confirmation.
- **Transitions**:
- `send_offer_rbf``LocalCloseStart` (Local node initiates fee increase
via RBF)
- `offer_received_rbf``RemoteCloseStart` (Remote node initiates fee
increase via RBF)
- **CloseErr**: Error state for handling protocol violations.
- **Transitions**:
- `send_offer_rbf``LocalCloseStart` (Local node attempts recovery with
new offer)
- `offer_received_rbf``RemoteCloseStart` (Remote node attempts recovery
with new offer)
### CloseFin
The closing transaction has been confirmed on the blockchain, and the channel is
considered closed.
- **Transitions**: None (terminal state)
## Fee Negotiation Process
Fee negotiation is a critical component of the cooperative close process. The
protocol supports Replace-By-Fee (RBF) to handle changing network fee
conditions:
1. Initial fee proposal is based on current network fee estimation
2. Either party can propose higher fees using RBF if confirmation is taking too
long
3. Nodes can alternate proposals until reaching agreement
4. Each proposal must satisfy RBF requirements (incremental fee increase)
## Protocol Violation Handling
The `CloseErr` state provides recovery paths when protocol violations occur:
- Invalid signature formats
- Out-of-sequence messages
- Timeout violations
- Malformed closing transactions
Recovery typically involves restarting the negotiation with a new closing offer.
## Example Scenarios
### Standard Cooperative Close
1. Local node initiates closing: `ChannelActive``ShutdownPending` (via
`send_shutdown`)
2. Remote acknowledges: `ShutdownPending``ChannelFlushing` (via
`shutdown_received`)
3. HTLCs resolve: `ChannelFlushing``ClosingNegotiation` (via
`channel_flushed`)
4. Local proposes fee: `LocalCloseStart``LocalOfferSent` (via `send_offer`)
5. Local signature validated: `LocalOfferSent``ClosePending` (via
`local_sig_received`)
6. Transaction confirms: `ClosePending``CloseFin` (via `txn_confirmation`)
### Fee Bump Due to Network Congestion
1. ... (steps 1-5 same as above)
2. While in `ClosePending`, network fees increase
3. Local node initiates RBF: `ClosePending``LocalCloseStart` (via
`send_offer_rbf`)
4. New proposal cycle begins with higher fees
5. When agreement is reached on new fees: `ClosePending``CloseFin` (via
`txn_confirmation`)
## Implementation Notes
- This state machine is implemented in the `peer.go` and `channel.go` files
within the lnd codebase
- State transitions are logged at the debug level
- The `ChanCloser` interface manages the state machine execution

View file

@ -1,8 +1,10 @@
package chancloser
import (
"github.com/btcsuite/btcd/btcec/v2"
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/msgmux"
)
// RbfMsgMapper is a struct that implements the MsgMapper interface for the
@ -16,16 +18,21 @@ type RbfMsgMapper struct {
// chanID is the channel ID of the channel being closed.
chanID lnwire.ChannelID
// peerPub is the public key of the peer that the channel is being
// closed.
peerPub btcec.PublicKey
}
// NewRbfMsgMapper creates a new RbfMsgMapper instance given the current block
// height when the co-op close request was initiated.
func NewRbfMsgMapper(blockHeight uint32,
chanID lnwire.ChannelID) *RbfMsgMapper {
chanID lnwire.ChannelID, peerPub btcec.PublicKey) *RbfMsgMapper {
return &RbfMsgMapper{
blockHeight: blockHeight,
chanID: chanID,
peerPub: peerPub,
}
}
@ -34,18 +41,20 @@ func someEvent[T ProtocolEvent](m T) fn.Option[ProtocolEvent] {
return fn.Some(ProtocolEvent(m))
}
// isExpectedChanID returns true if the channel ID of the message matches the
// isForUs returns true if the channel ID + pubkey of the message matches the
// bound instance.
func (r *RbfMsgMapper) isExpectedChanID(chanID lnwire.ChannelID) bool {
return r.chanID == chanID
func (r *RbfMsgMapper) isForUs(chanID lnwire.ChannelID,
fromPub btcec.PublicKey) bool {
return r.chanID == chanID && r.peerPub.IsEqual(&fromPub)
}
// MapMsg maps a wire message into a FSM event. If the message is not mappable,
// then an error is returned.
func (r *RbfMsgMapper) MapMsg(wireMsg lnwire.Message) fn.Option[ProtocolEvent] {
switch msg := wireMsg.(type) {
func (r *RbfMsgMapper) MapMsg(wireMsg msgmux.PeerMsg) fn.Option[ProtocolEvent] {
switch msg := wireMsg.Message.(type) {
case *lnwire.Shutdown:
if !r.isExpectedChanID(msg.ChannelID) {
if !r.isForUs(msg.ChannelID, wireMsg.PeerPub) {
return fn.None[ProtocolEvent]()
}
@ -55,7 +64,7 @@ func (r *RbfMsgMapper) MapMsg(wireMsg lnwire.Message) fn.Option[ProtocolEvent] {
})
case *lnwire.ClosingComplete:
if !r.isExpectedChanID(msg.ChannelID) {
if !r.isForUs(msg.ChannelID, wireMsg.PeerPub) {
return fn.None[ProtocolEvent]()
}
@ -64,7 +73,7 @@ func (r *RbfMsgMapper) MapMsg(wireMsg lnwire.Message) fn.Option[ProtocolEvent] {
})
case *lnwire.ClosingSig:
if !r.isExpectedChanID(msg.ChannelID) {
if !r.isForUs(msg.ChannelID, wireMsg.PeerPub) {
return fn.None[ProtocolEvent]()
}

View file

@ -49,6 +49,11 @@ var (
// ErrCloserAndClosee is returned when we expect a sig covering both
// outputs, it isn't present.
ErrCloserAndClosee = fmt.Errorf("expected CloserAndClosee sig")
// ErrWrongLocalScript is returned when the remote party sends a
// ClosingComplete message that doesn't carry our last local script
// sent.
ErrWrongLocalScript = fmt.Errorf("wrong local script")
)
// ProtocolEvent is a special interface used to create the equivalent of a
@ -361,6 +366,9 @@ type ProtocolState interface {
// ProcessEvent takes a protocol event, and implements a state
// transition for the state.
ProcessEvent(ProtocolEvent, *Environment) (*CloseStateTransition, error)
// String returns the name of the state.
String() string
}
// AsymmetricPeerState is an extension of the normal ProtocolState interface
@ -379,7 +387,7 @@ type AsymmetricPeerState interface {
type ProtocolStates interface {
ChannelActive | ShutdownPending | ChannelFlushing | ClosingNegotiation |
LocalCloseStart | LocalOfferSent | RemoteCloseStart |
ClosePending | CloseFin
ClosePending | CloseFin | CloseErr
}
// ChannelActive is the base state for the channel closer state machine. In
@ -401,6 +409,11 @@ type ProtocolStates interface {
type ChannelActive struct {
}
// String returns the name of the state for ChannelActive.
func (c *ChannelActive) String() string {
return "ChannelActive"
}
// IsTerminal returns true if the target state is a terminal state.
func (c *ChannelActive) IsTerminal() bool {
return false
@ -441,6 +454,16 @@ type ShutdownPending struct {
// IdealFeeRate is the ideal fee rate we'd like to use for the closing
// attempt.
IdealFeeRate fn.Option[chainfee.SatPerVByte]
// EarlyRemoteOffer is the offer we received from the remote party
// before we received their shutdown message. We'll stash it to process
// later.
EarlyRemoteOffer fn.Option[OfferReceivedEvent]
}
// String returns the name of the state for ShutdownPending.
func (s *ShutdownPending) String() string {
return "ShutdownPending"
}
// IsTerminal returns true if the target state is a terminal state.
@ -478,6 +501,11 @@ type ChannelFlushing struct {
IdealFeeRate fn.Option[chainfee.SatPerVByte]
}
// String returns the name of the state for ChannelFlushing.
func (c *ChannelFlushing) String() string {
return "ChannelFlushing"
}
// protocolStateSealed indicates that this struct is a ProtocolEvent instance.
func (c *ChannelFlushing) protocolStateSealed() {}
@ -505,6 +533,22 @@ type ClosingNegotiation struct {
// the ShouldRouteTo method to determine which state route incoming
// events to.
PeerState lntypes.Dual[AsymmetricPeerState]
// CloseChannelTerms is the terms we'll use to close the channel. We
// hold a value here which is pointed to by the various
// AsymmetricPeerState instances. This allows us to update this value if
// the remote peer sends a new address, with each of the state noting
// the new value via a pointer.
*CloseChannelTerms
}
// String returns the name of the state for ClosingNegotiation.
func (c *ClosingNegotiation) String() string {
localState := c.PeerState.GetForParty(lntypes.Local)
remoteState := c.PeerState.GetForParty(lntypes.Remote)
return fmt.Sprintf("ClosingNegotiation(local=%v, remote=%v)",
localState, remoteState)
}
// IsTerminal returns true if the target state is a terminal state.
@ -515,6 +559,56 @@ func (c *ClosingNegotiation) IsTerminal() bool {
// protocolSealed indicates that this struct is a ProtocolEvent instance.
func (c *ClosingNegotiation) protocolStateSealed() {}
// ErrState can be used to introspect into a benign error related to a state
// transition.
type ErrState interface {
sealed()
error
// Err returns an error for the ErrState.
Err() error
}
// ErrStateCantPayForFee is sent when the local party attempts a fee update
// that they can't actually party for.
type ErrStateCantPayForFee struct {
localBalance btcutil.Amount
attemptedFee btcutil.Amount
}
// NewErrStateCantPayForFee returns a new NewErrStateCantPayForFee error.
func NewErrStateCantPayForFee(localBalance, attemptedFee btcutil.Amount,
) *ErrStateCantPayForFee {
return &ErrStateCantPayForFee{
localBalance: localBalance,
attemptedFee: attemptedFee,
}
}
// sealed makes this a sealed interface.
func (e *ErrStateCantPayForFee) sealed() {
}
// Err returns an error for the ErrState.
func (e *ErrStateCantPayForFee) Err() error {
return fmt.Errorf("cannot pay for fee of %v, only have %v local "+
"balance", e.attemptedFee, e.localBalance)
}
// Error returns the error string for the ErrState.
func (e *ErrStateCantPayForFee) Error() string {
return e.Err().Error()
}
// String returns the string for the ErrStateCantPayForFee.
func (e *ErrStateCantPayForFee) String() string {
return fmt.Sprintf("ErrStateCantPayForFee(local_balance=%v, "+
"attempted_fee=%v)", e.localBalance, e.attemptedFee)
}
// CloseChannelTerms is a set of terms that we'll use to close the channel. This
// includes the balances of the channel, and the scripts we'll use to send each
// party's funds to.
@ -526,11 +620,11 @@ type CloseChannelTerms struct {
// DeriveCloseTxOuts takes the close terms, and returns the local and remote tx
// out for the close transaction. If an output is dust, then it'll be nil.
//
// TODO(roasbeef): add func for w/e heuristic to not manifest own output?
func (c *CloseChannelTerms) DeriveCloseTxOuts() (*wire.TxOut, *wire.TxOut) {
//nolint:ll
deriveTxOut := func(balance btcutil.Amount, pkScript []byte) *wire.TxOut {
// We'll base the existence of the output on our normal dust
// check.
dustLimit := lnwallet.DustLimitForSize(len(pkScript))
if balance >= dustLimit {
return &wire.TxOut{
@ -591,7 +685,13 @@ func (c *CloseChannelTerms) RemoteCanPayFees(absoluteFee btcutil.Amount) bool {
// input events:
// - SendOfferEvent
type LocalCloseStart struct {
CloseChannelTerms
*CloseChannelTerms
}
// String returns the name of the state for LocalCloseStart, including proposed
// fee details.
func (l *LocalCloseStart) String() string {
return "LocalCloseStart"
}
// ShouldRouteTo returns true if the target state should process the target
@ -625,15 +725,23 @@ func (l *LocalCloseStart) protocolStateSealed() {}
// input events:
// - LocalSigReceived
type LocalOfferSent struct {
CloseChannelTerms
*CloseChannelTerms
// ProposedFee is the fee we proposed to the remote party.
ProposedFee btcutil.Amount
// ProposedFeeRate is the fee rate we proposed to the remote party.
ProposedFeeRate chainfee.SatPerVByte
// LocalSig is the signature we sent to the remote party.
LocalSig lnwire.Sig
}
// String returns the name of the state for LocalOfferSent, including proposed.
func (l *LocalOfferSent) String() string {
return fmt.Sprintf("LocalOfferSent(proposed_fee=%v)", l.ProposedFee)
}
// ShouldRouteTo returns true if the target state should process the target
// event.
func (l *LocalOfferSent) ShouldRouteTo(event ProtocolEvent) bool {
@ -668,6 +776,27 @@ func (l *LocalOfferSent) IsTerminal() bool {
type ClosePending struct {
// CloseTx is the pending close transaction.
CloseTx *wire.MsgTx
*CloseChannelTerms
// FeeRate is the fee rate of the closing transaction.
FeeRate chainfee.SatPerVByte
// Party indicates which party is at this state. This is used to
// implement the state transition properly, based on ShouldRouteTo.
Party lntypes.ChannelParty
}
// String returns the name of the state for ClosePending.
func (c *ClosePending) String() string {
return fmt.Sprintf("ClosePending(txid=%v, party=%v, fee_rate=%v)",
c.CloseTx.TxHash(), c.Party, c.FeeRate)
}
// isType returns true if the value is of type T.
func isType[T any](value any) bool {
_, ok := value.(T)
return ok
}
// ShouldRouteTo returns true if the target state should process the target
@ -677,6 +806,17 @@ func (c *ClosePending) ShouldRouteTo(event ProtocolEvent) bool {
case *SpendEvent:
return true
default:
switch {
case c.Party == lntypes.Local && isType[*SendOfferEvent](event):
return true
case c.Party == lntypes.Remote && isType[*OfferReceivedEvent](
event,
):
return true
}
return false
}
}
@ -696,6 +836,11 @@ type CloseFin struct {
ConfirmedTx *wire.MsgTx
}
// String returns the name of the state for CloseFin.
func (c *CloseFin) String() string {
return "CloseFin"
}
// protocolStateSealed indicates that this struct is a ProtocolEvent instance.
func (c *CloseFin) protocolStateSealed() {}
@ -711,7 +856,12 @@ func (c *CloseFin) IsTerminal() bool {
// - fromState: ChannelFlushing
// - toState: ClosePending
type RemoteCloseStart struct {
CloseChannelTerms
*CloseChannelTerms
}
// String returns the name of the state for RemoteCloseStart.
func (r *RemoteCloseStart) String() string {
return "RemoteCloseStart"
}
// ShouldRouteTo returns true if the target state should process the target
@ -733,6 +883,54 @@ func (l *RemoteCloseStart) IsTerminal() bool {
return false
}
// CloseErr is an error state in the protocol. We enter this state when a
// protocol constraint is violated, or an upfront sanity check fails.
type CloseErr struct {
ErrState
*CloseChannelTerms
// Party indicates which party is at this state. This is used to
// implement the state transition properly, based on ShouldRouteTo.
Party lntypes.ChannelParty
}
// String returns the name of the state for CloseErr, including error and party
// details.
func (c *CloseErr) String() string {
return fmt.Sprintf("CloseErr(party=%v, err=%v)", c.Party, c.ErrState)
}
// ShouldRouteTo returns true if the target state should process the target
// event.
func (c *CloseErr) ShouldRouteTo(event ProtocolEvent) bool {
switch event.(type) {
case *SpendEvent:
return true
default:
switch {
case c.Party == lntypes.Local && isType[*SendOfferEvent](event):
return true
case c.Party == lntypes.Remote && isType[*OfferReceivedEvent](
event,
):
return true
}
return false
}
}
// protocolStateSealed indicates that this struct is a ProtocolEvent instance.
func (c *CloseErr) protocolStateSealed() {}
// IsTerminal returns true if the target state is a terminal state.
func (c *CloseErr) IsTerminal() bool {
return true
}
// RbfChanCloser is a state machine that handles the RBF-enabled cooperative
// channel close protocol.
type RbfChanCloser = protofsm.StateMachine[ProtocolEvent, *Environment]
@ -761,3 +959,7 @@ type RbfState = protofsm.State[ProtocolEvent, *Environment]
// RbfEvent is a type alias for the event type of the RBF channel closer.
type RbfEvent = protofsm.EmittedEvent[ProtocolEvent]
// RbfStateSub is a type alias for the state subscription type of the RBF chan
// closer.
type RbfStateSub = protofsm.StateSubscriber[ProtocolEvent, *Environment]

View file

@ -110,7 +110,10 @@ func assertStateTransitions[Event any, Env protofsm.Environment](
t.Helper()
for _, expectedState := range expectedStates {
newState := <-stateSub.NewItemCreated.ChanOut()
newState, err := fn.RecvOrTimeout(
stateSub.NewItemCreated.ChanOut(), 10*time.Millisecond,
)
require.NoError(t, err, "expected state: %T", expectedState)
require.IsType(t, expectedState, newState)
}
@ -154,6 +157,27 @@ func assertUnknownEventFail(t *testing.T, startingState ProtocolState) {
})
}
// assertSpendEventCloseFin asserts that the state machine transitions to the
// CloseFin state when a spend event is received.
func assertSpendEventCloseFin(t *testing.T, startingState ProtocolState) {
t.Helper()
// If a spend event is received, the state machine should transition to
// the CloseFin state.
t.Run("spend_event", func(t *testing.T) {
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some(startingState),
})
defer closeHarness.stopAndAssert()
closeHarness.chanCloser.SendEvent(
context.Background(), &SpendEvent{},
)
closeHarness.assertStateTransitions(&CloseFin{})
})
}
type harnessCfg struct {
initialState fn.Option[ProtocolState]
@ -248,6 +272,8 @@ func (r *rbfCloserTestHarness) assertNoStateTransitions() {
}
func (r *rbfCloserTestHarness) assertStateTransitions(states ...RbfState) {
r.T.Helper()
assertStateTransitions(r.T, r.stateSub, states)
}
@ -598,6 +624,8 @@ func (r *rbfCloserTestHarness) assertSingleRbfIteration(
// response of the remote party, which completes one iteration
localSigEvent := &LocalSigReceived{
SigMsg: lnwire.ClosingSig{
CloserScript: localAddr,
CloseeScript: remoteAddr,
ClosingSigs: lnwire.ClosingSigs{
CloserAndClosee: newSigTlv[tlv.TlvType3](
remoteWireSig,
@ -620,34 +648,34 @@ func (r *rbfCloserTestHarness) assertSingleRbfIteration(
}
func (r *rbfCloserTestHarness) assertSingleRemoteRbfIteration(
initEvent ProtocolEvent, balanceAfterClose, absoluteFee btcutil.Amount,
sequence uint32, iteration bool) {
initEvent *OfferReceivedEvent, balanceAfterClose,
absoluteFee btcutil.Amount, sequence uint32, iteration bool,
sendInit bool) {
ctx := context.Background()
// If this is an iteration, then we expect some intermediate states,
// before we enter the main RBF/sign loop.
if iteration {
r.expectFeeEstimate(absoluteFee, 1)
r.assertStateTransitions(
&ChannelActive{}, &ShutdownPending{},
&ChannelFlushing{}, &ClosingNegotiation{},
)
}
// When we receive the signature below, our local state machine should
// move to finalize the close.
r.expectRemoteCloseFinalized(
&localSig, &remoteSig, localAddr, remoteAddr,
&localSig, &remoteSig, initEvent.SigMsg.CloseeScript,
initEvent.SigMsg.CloserScript,
absoluteFee, balanceAfterClose, false,
)
r.chanCloser.SendEvent(ctx, initEvent)
if sendInit {
r.chanCloser.SendEvent(ctx, initEvent)
}
// Our outer state should transition to ClosingNegotiation state.
r.assertStateTransitions(&ClosingNegotiation{})
// If this is an iteration, then we'll go from ClosePending ->
// RemoteCloseStart -> ClosePending. So we'll assert an extra transition
// here.
if iteration {
r.assertStateTransitions(&ClosingNegotiation{})
}
// If we examine the final resting state, we should see that the we're
// now in the ClosePending state for the remote peer.
currentState := assertStateT[*ClosingNegotiation](r)
@ -687,7 +715,7 @@ func newRbfCloserTestHarness(t *testing.T,
peerPub := randPubKey(t)
msgMapper := NewRbfMsgMapper(uint32(startingHeight), chanID)
msgMapper := NewRbfMsgMapper(uint32(startingHeight), chanID, *peerPub)
initialState := cfg.initialState.UnwrapOr(&ChannelActive{})
@ -858,7 +886,28 @@ func TestRbfChannelActiveTransitions(t *testing.T) {
closeHarness.waitForMsgSent()
})
// TODO(roasbeef): thaw height fail
// If the remote party attempts to close, and a thaw height is active,
// but not yet met, then we should fail.
t.Run("remote_initiated_thaw_height_close_fail", func(t *testing.T) {
closeHarness := newCloser(t, &harnessCfg{
localUpfrontAddr: fn.Some(localAddr),
thawHeight: fn.Some(uint32(100000)),
})
defer closeHarness.stopAndAssert()
// Next, we'll emit the recv event, with the addr of the remote
// party.
closeHarness.chanCloser.SendEvent(
ctx, &ShutdownReceived{
ShutdownScript: remoteAddr,
BlockHeight: 1,
},
)
// We expect a failure as the block height is less than the
// start height.
closeHarness.expectFailure(ErrThawHeightNotReached)
})
// When we receive a shutdown, we should transition to the shutdown
// pending state, with the local+remote shutdown addrs known.
@ -902,6 +951,9 @@ func TestRbfChannelActiveTransitions(t *testing.T) {
// Any other event should be ignored.
assertUnknownEventFail(t, &ChannelActive{})
// Sending a Spend event should transition to CloseFin.
assertSpendEventCloseFin(t, &ChannelActive{})
}
// TestRbfShutdownPendingTransitions tests the transitions of the RBF closer
@ -1033,8 +1085,106 @@ func TestRbfShutdownPendingTransitions(t *testing.T) {
closeHarness.assertStateTransitions(&ChannelFlushing{})
})
// If we an early offer from the remote party, then we should stash
// that, transition to the channel flushing state. Once there, another
// self transition should emit the stashed offer.
t.Run("early_remote_offer_shutdown_complete", func(t *testing.T) {
firstState := *startingState
firstState.IdealFeeRate = fn.Some(
chainfee.FeePerKwFloor.FeePerVByte(),
)
firstState.ShutdownScripts = ShutdownScripts{
LocalDeliveryScript: localAddr,
RemoteDeliveryScript: remoteAddr,
}
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some[ProtocolState](
&firstState,
),
})
defer closeHarness.stopAndAssert()
// In this case we're doing the shutdown dance for the first
// time, so we'll mark the channel as not being flushed.
closeHarness.expectFinalBalances(fn.None[ShutdownBalances]())
// Before we send the shutdown complete event, we'll send in an
// early offer from the remote party.
closeHarness.chanCloser.SendEvent(ctx, &OfferReceivedEvent{})
// This will cause a self transition back to ShutdownPending.
closeHarness.assertStateTransitions(&ShutdownPending{})
// Next, we'll send in a shutdown complete event.
closeHarness.chanCloser.SendEvent(ctx, &ShutdownComplete{})
// We should transition to the channel flushing state, then the
// self event to have this state cache he early offer should
// follow.
closeHarness.assertStateTransitions(
&ChannelFlushing{}, &ChannelFlushing{},
)
// If we get the current state, we should see that the offer is
// cached.
currentState := assertStateT[*ChannelFlushing](closeHarness)
require.NotNil(t, currentState.EarlyRemoteOffer)
})
// If we an early offer from the remote party, then we should stash
// that, transition to the channel flushing state. Once there, another
// self transition should emit the stashed offer.
t.Run("early_remote_offer_shutdown_received", func(t *testing.T) {
firstState := *startingState
firstState.IdealFeeRate = fn.Some(
chainfee.FeePerKwFloor.FeePerVByte(),
)
firstState.ShutdownScripts = ShutdownScripts{
LocalDeliveryScript: localAddr,
RemoteDeliveryScript: remoteAddr,
}
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some[ProtocolState](
&firstState,
),
})
defer closeHarness.stopAndAssert()
// In this case we're doing the shutdown dance for the first
// time, so we'll mark the channel as not being flushed.
closeHarness.expectFinalBalances(fn.None[ShutdownBalances]())
closeHarness.expectIncomingAddsDisabled()
// Before we send the shutdown complete event, we'll send in an
// early offer from the remote party.
closeHarness.chanCloser.SendEvent(ctx, &OfferReceivedEvent{})
// This will cause a self transition back to ShutdownPending.
closeHarness.assertStateTransitions(&ShutdownPending{})
// Next, we'll send in a shutdown complete event.
closeHarness.chanCloser.SendEvent(ctx, &ShutdownReceived{})
// We should transition to the channel flushing state, then the
// self event to have this state cache he early offer should
// follow.
closeHarness.assertStateTransitions(
&ChannelFlushing{}, &ChannelFlushing{},
)
// If we get the current state, we should see that the offer is
// cached.
currentState := assertStateT[*ChannelFlushing](closeHarness)
require.NotNil(t, currentState.EarlyRemoteOffer)
})
// Any other event should be ignored.
assertUnknownEventFail(t, startingState)
// Sending a Spend event should transition to CloseFin.
assertSpendEventCloseFin(t, startingState)
}
// TestRbfChannelFlushingTransitions tests the transitions of the RBF closer
@ -1142,7 +1292,7 @@ func TestRbfChannelFlushingTransitions(t *testing.T) {
closeHarness.expectChanPendingClose()
}
// From where, we expect the state transition to go
// From here, we expect the state transition to go
// back to closing negotiated, for a ClosingComplete
// message to be sent and then for us to terminate at
// that state. This is 1/2 of the normal RBF signer
@ -1154,8 +1304,65 @@ func TestRbfChannelFlushingTransitions(t *testing.T) {
})
}
// This tests that if we receive an `OfferReceivedEvent` while in the
// flushing state, then we'll cache that, and once we receive
// ChannelFlushed, we'll emit an internal `OfferReceivedEvent` in the
// negotiation state.
t.Run("early_offer", func(t *testing.T) {
firstState := *startingState
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some[ProtocolState](
&firstState,
),
})
defer closeHarness.stopAndAssert()
flushEvent := *flushTemplate
// Set up the fee estimate s.t the local party doesn't have
// balance to close.
closeHarness.expectFeeEstimate(absoluteFee, 1)
// First, we'll emit an `OfferReceivedEvent` to simulate an
// early offer (async network, they determine the channel is
// "flushed" before we do, and send their offer over).
remoteOffer := &OfferReceivedEvent{
SigMsg: lnwire.ClosingComplete{
FeeSatoshis: absoluteFee,
CloserScript: remoteAddr,
CloseeScript: localAddr,
ClosingSigs: lnwire.ClosingSigs{
CloserAndClosee: newSigTlv[tlv.TlvType3]( //nolint:ll
remoteWireSig,
),
},
},
}
closeHarness.chanCloser.SendEvent(ctx, remoteOffer)
// We should do a self transition, and still be in the
// ChannelFlushing state.
closeHarness.assertStateTransitions(&ChannelFlushing{})
sequence := uint32(mempool.MaxRBFSequence)
// Now we'll send in the channel flushed event, and assert that
// this triggers a remote RBF iteration (we process their early
// offer and send our sig).
closeHarness.chanCloser.SendEvent(ctx, &flushEvent)
closeHarness.assertSingleRemoteRbfIteration(
remoteOffer, absoluteFee, absoluteFee, sequence, false,
true,
)
})
// Any other event should be ignored.
assertUnknownEventFail(t, startingState)
// Sending a Spend event should transition to CloseFin.
assertSpendEventCloseFin(t, startingState)
}
// TestRbfCloseClosingNegotiationLocal tests the local portion of the primary
@ -1184,9 +1391,10 @@ func TestRbfCloseClosingNegotiationLocal(t *testing.T) {
startingState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: *closeTerms,
CloseChannelTerms: closeTerms,
},
},
CloseChannelTerms: closeTerms,
}
sendOfferEvent := &SendOfferEvent{
@ -1195,6 +1403,8 @@ func TestRbfCloseClosingNegotiationLocal(t *testing.T) {
balanceAfterClose := localBalance.ToSatoshis() - absoluteFee
// TODO(roasbeef): add test case for error state validation, then resume
// In this state, we'll simulate deciding that we need to send a new
// offer to the remote party.
t.Run("send_offer_iteration_no_dust", func(t *testing.T) {
@ -1231,6 +1441,8 @@ func TestRbfCloseClosingNegotiationLocal(t *testing.T) {
// we'll specify 2 signature fields.
localSigEvent := &LocalSigReceived{
SigMsg: lnwire.ClosingSig{
CloserScript: localAddr,
CloseeScript: remoteAddr,
ClosingSigs: lnwire.ClosingSigs{
CloserNoClosee: newSigTlv[tlv.TlvType1](
remoteWireSig,
@ -1261,9 +1473,10 @@ func TestRbfCloseClosingNegotiationLocal(t *testing.T) {
firstState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: newCloseTerms,
CloseChannelTerms: &newCloseTerms,
},
},
CloseChannelTerms: &newCloseTerms,
}
closeHarness := newCloser(t, &harnessCfg{
@ -1286,9 +1499,10 @@ func TestRbfCloseClosingNegotiationLocal(t *testing.T) {
firstState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: *closeTerms,
CloseChannelTerms: closeTerms,
},
},
CloseChannelTerms: closeTerms,
}
closeHarness := newCloser(t, &harnessCfg{
@ -1306,15 +1520,55 @@ func TestRbfCloseClosingNegotiationLocal(t *testing.T) {
)
})
// In this test, we'll assert that we're able to restart the RBF loop
// to trigger additional signature iterations.
t.Run("send_offer_rbf_wrong_local_script", func(t *testing.T) {
firstState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: closeTerms,
},
},
CloseChannelTerms: closeTerms,
}
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some[ProtocolState](firstState),
localUpfrontAddr: fn.Some(localAddr),
})
defer closeHarness.stopAndAssert()
// The remote party will send a ClosingSig message, but with the
// wrong local script. We should expect an error.
closeHarness.expectFailure(ErrWrongLocalScript)
// We'll send this message in directly, as we shouldn't get any
// further in the process.
// assuming we start in this negotiation state.
localSigEvent := &LocalSigReceived{
SigMsg: lnwire.ClosingSig{
CloserScript: remoteAddr,
CloseeScript: remoteAddr,
ClosingSigs: lnwire.ClosingSigs{
CloserAndClosee: newSigTlv[tlv.TlvType3]( //nolint:ll
remoteWireSig,
),
},
},
}
closeHarness.chanCloser.SendEvent(ctx, localSigEvent)
})
// In this test, we'll assert that we're able to restart the RBF loop
// to trigger additional signature iterations.
t.Run("send_offer_rbf_iteration_loop", func(t *testing.T) {
firstState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: *closeTerms,
CloseChannelTerms: closeTerms,
},
},
CloseChannelTerms: closeTerms,
}
closeHarness := newCloser(t, &harnessCfg{
@ -1330,44 +1584,18 @@ func TestRbfCloseClosingNegotiationLocal(t *testing.T) {
noDustExpect,
)
// Next, we'll send in a new SendShutdown event which simulates
// the user requesting a RBF fee bump. We'll use 10x the fee we
// used in the last iteration.
// Next, we'll send in a new SendOfferEvent event which
// simulates the user requesting a RBF fee bump. We'll use 10x
// the fee we used in the last iteration.
rbfFeeBump := chainfee.FeePerKwFloor.FeePerVByte() * 10
sendShutdown := &SendShutdown{
IdealFeeRate: rbfFeeBump,
localOffer := &SendOfferEvent{
TargetFeeRate: rbfFeeBump,
}
// We should send shutdown as normal, but skip some other
// checks as we know the close is in progress.
closeHarness.expectShutdownEvents(shutdownExpect{
allowSend: true,
finalBalances: fn.Some(closeTerms.ShutdownBalances),
recvShutdown: true,
})
closeHarness.expectMsgSent(
singleMsgMatcher[*lnwire.Shutdown](nil),
)
closeHarness.chanCloser.SendEvent(ctx, sendShutdown)
// We should first transition to the Channel Active state
// momentarily, before transitioning to the shutdown pending
// state.
closeHarness.assertStateTransitions(
&ChannelActive{}, &ShutdownPending{},
)
// Next, we'll send in the shutdown received event, which
// should transition us to the channel flushing state.
shutdownEvent := &ShutdownReceived{
ShutdownScript: remoteAddr,
}
// Now we expect that aanother full RBF iteration takes place
// (we initiatea a new local sig).
// Now we expect that another full RBF iteration takes place (we
// initiate a new local sig).
closeHarness.assertSingleRbfIteration(
shutdownEvent, balanceAfterClose, absoluteFee,
localOffer, balanceAfterClose, absoluteFee,
noDustExpect,
)
@ -1376,6 +1604,59 @@ func TestRbfCloseClosingNegotiationLocal(t *testing.T) {
&ClosingNegotiation{},
)
})
// Make sure that we'll go to the error state if we try to try a close
// that we can't pay for.
t.Run("send_offer_cannot_pay_for_fees", func(t *testing.T) {
firstState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: closeTerms,
},
},
CloseChannelTerms: closeTerms,
}
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some[ProtocolState](firstState),
localUpfrontAddr: fn.Some(localAddr),
})
defer closeHarness.stopAndAssert()
// We'll prep to return an absolute fee that's much higher than
// the amount we have in the channel.
closeHarness.expectFeeEstimate(btcutil.SatoshiPerBitcoin, 1)
rbfFeeBump := chainfee.FeePerKwFloor.FeePerVByte()
localOffer := &SendOfferEvent{
TargetFeeRate: rbfFeeBump,
}
// Next, we'll send in this event, which should fail as we can't
// actually pay for fees.
closeHarness.chanCloser.SendEvent(ctx, localOffer)
// We should transition to the CloseErr (within
// ClosingNegotiation) state.
closeHarness.assertStateTransitions(&ClosingNegotiation{})
// If we get the state, we should see the expected ErrState.
currentState := assertStateT[*ClosingNegotiation](closeHarness)
closeErrState, ok := currentState.PeerState.GetForParty(
lntypes.Local,
).(*CloseErr)
require.True(t, ok)
require.IsType(
t, &ErrStateCantPayForFee{}, closeErrState.ErrState,
)
})
// Any other event should be ignored.
assertUnknownEventFail(t, startingState)
// Sending a Spend event should transition to CloseFin.
assertSpendEventCloseFin(t, startingState)
}
// TestRbfCloseClosingNegotiationRemote tests that state machine is able to
@ -1383,6 +1664,7 @@ func TestRbfCloseClosingNegotiationLocal(t *testing.T) {
// party.
func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
t.Parallel()
ctx := context.Background()
localBalance := lnwire.NewMSatFromSatoshis(40_000)
@ -1403,16 +1685,16 @@ func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
startingState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: *closeTerms,
CloseChannelTerms: closeTerms,
},
Remote: &RemoteCloseStart{
CloseChannelTerms: *closeTerms,
CloseChannelTerms: closeTerms,
},
},
CloseChannelTerms: closeTerms,
}
balanceAfterClose := remoteBalance.ToSatoshis() - absoluteFee
sequence := uint32(mempool.MaxRBFSequence)
// This case tests that if we receive a signature from the remote
@ -1430,7 +1712,9 @@ func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
// be higher than the remote party's balance.
feeOffer := &OfferReceivedEvent{
SigMsg: lnwire.ClosingComplete{
FeeSatoshis: absoluteFee * 10,
CloserScript: remoteAddr,
CloseeScript: localAddr,
FeeSatoshis: absoluteFee * 10,
},
}
closeHarness.chanCloser.SendEvent(ctx, feeOffer)
@ -1449,12 +1733,13 @@ func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
firstState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: closingTerms,
CloseChannelTerms: &closingTerms,
},
Remote: &RemoteCloseStart{
CloseChannelTerms: closingTerms,
CloseChannelTerms: &closingTerms,
},
},
CloseChannelTerms: &closingTerms,
}
closeHarness := newCloser(t, &harnessCfg{
@ -1469,7 +1754,9 @@ func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
// includes our output.
feeOffer := &OfferReceivedEvent{
SigMsg: lnwire.ClosingComplete{
FeeSatoshis: absoluteFee,
FeeSatoshis: absoluteFee,
CloserScript: remoteAddr,
CloseeScript: localAddr,
ClosingSigs: lnwire.ClosingSigs{
CloserAndClosee: newSigTlv[tlv.TlvType3]( //nolint:ll
remoteWireSig,
@ -1498,7 +1785,9 @@ func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
// signature as it excludes an output.
feeOffer := &OfferReceivedEvent{
SigMsg: lnwire.ClosingComplete{
FeeSatoshis: absoluteFee,
FeeSatoshis: absoluteFee,
CloserScript: remoteAddr,
CloseeScript: localAddr,
ClosingSigs: lnwire.ClosingSigs{
CloserNoClosee: newSigTlv[tlv.TlvType1]( //nolint:ll
remoteWireSig,
@ -1516,8 +1805,8 @@ func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
// loops to enable the remote party to sign.new versions of the co-op
// close transaction.
t.Run("recv_offer_rbf_loop_iterations", func(t *testing.T) {
// We'll modify our s.t we're unable to pay for fees, but
// aren't yet dust.
// We'll modify our balance s.t we're unable to pay for fees,
// but aren't yet dust.
closingTerms := *closeTerms
closingTerms.ShutdownBalances.LocalBalance = lnwire.NewMSatFromSatoshis( //nolint:ll
9000,
@ -1526,12 +1815,13 @@ func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
firstState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: closingTerms,
CloseChannelTerms: &closingTerms,
},
Remote: &RemoteCloseStart{
CloseChannelTerms: closingTerms,
CloseChannelTerms: &closingTerms,
},
},
CloseChannelTerms: &closingTerms,
}
closeHarness := newCloser(t, &harnessCfg{
@ -1542,8 +1832,10 @@ func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
feeOffer := &OfferReceivedEvent{
SigMsg: lnwire.ClosingComplete{
FeeSatoshis: absoluteFee,
LockTime: 1,
CloserScript: remoteAddr,
CloseeScript: localAddr,
FeeSatoshis: absoluteFee,
LockTime: 1,
ClosingSigs: lnwire.ClosingSigs{
CloserAndClosee: newSigTlv[tlv.TlvType3]( //nolint:ll
remoteWireSig,
@ -1557,43 +1849,214 @@ func TestRbfCloseClosingNegotiationRemote(t *testing.T) {
// sig.
closeHarness.assertSingleRemoteRbfIteration(
feeOffer, balanceAfterClose, absoluteFee, sequence,
false,
false, true,
)
// At this point, we've completed a single RBF iteration, and
// want to test further iterations, so we'll use a shutdown
// even tot kick it all off.
//
// Before we send the shutdown messages below, we'll mark the
// balances as so we fast track to the negotiation state.
closeHarness.expectShutdownEvents(shutdownExpect{
allowSend: true,
finalBalances: fn.Some(closingTerms.ShutdownBalances),
recvShutdown: true,
})
closeHarness.expectMsgSent(
singleMsgMatcher[*lnwire.Shutdown](nil),
)
// We'll now simulate the start of the RBF loop, by receiving a
// new Shutdown message from the remote party. This signals
// that they want to obtain a new commit sig.
closeHarness.chanCloser.SendEvent(
ctx, &ShutdownReceived{ShutdownScript: remoteAddr},
)
// Next, we'll receive an offer from the remote party, and
// drive another RBF iteration. This time, we'll increase the
// absolute fee by 1k sats.
// Next, we'll receive an offer from the remote party, and drive
// another RBF iteration. This time, we'll increase the absolute
// fee by 1k sats.
feeOffer.SigMsg.FeeSatoshis += 1000
absoluteFee = feeOffer.SigMsg.FeeSatoshis
closeHarness.assertSingleRemoteRbfIteration(
feeOffer, balanceAfterClose, absoluteFee, sequence,
true,
true, true,
)
closeHarness.assertNoStateTransitions()
})
// TODO(roasbeef): cross sig case? tested isolation, so wolog?
// This tests that if we get an offer that has the wrong local script,
// then we'll emit a hard error.
t.Run("recv_offer_wrong_local_script", func(t *testing.T) {
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some[ProtocolState](startingState),
})
defer closeHarness.stopAndAssert()
// The remote party will send a ClosingComplete message, but
// with the wrong local script. We should expect an error.
closeHarness.expectFailure(ErrWrongLocalScript)
// We'll send our remote addr as the Closee script, which should
// trigger an error.
feeOffer := &OfferReceivedEvent{
SigMsg: lnwire.ClosingComplete{
FeeSatoshis: absoluteFee,
CloserScript: remoteAddr,
CloseeScript: remoteAddr,
ClosingSigs: lnwire.ClosingSigs{
CloserNoClosee: newSigTlv[tlv.TlvType1]( //nolint:ll
remoteWireSig,
),
},
},
}
closeHarness.chanCloser.SendEvent(ctx, feeOffer)
// We shouldn't have transitioned to a new state.
closeHarness.assertNoStateTransitions()
})
// If we receive an offer from the remote party with a different remote
// script, then this ensures that we'll process that and use that create
// the next offer.
t.Run("recv_offer_remote_addr_change", func(t *testing.T) {
closingTerms := *closeTerms
firstState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: &closingTerms,
},
Remote: &RemoteCloseStart{
CloseChannelTerms: &closingTerms,
},
},
CloseChannelTerms: &closingTerms,
}
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some[ProtocolState](firstState),
localUpfrontAddr: fn.Some(localAddr),
})
defer closeHarness.stopAndAssert()
// This time, the close request sent by the remote party will
// modify their normal remote address. This should cause us to
// recognize this, and counter sign the proper co-op close
// transaction.
newRemoteAddr := lnwire.DeliveryAddress(append(
[]byte{txscript.OP_1, txscript.OP_DATA_32},
bytes.Repeat([]byte{0x03}, 32)...,
))
feeOffer := &OfferReceivedEvent{
SigMsg: lnwire.ClosingComplete{
CloserScript: newRemoteAddr,
CloseeScript: localAddr,
FeeSatoshis: absoluteFee,
LockTime: 1,
ClosingSigs: lnwire.ClosingSigs{
CloserAndClosee: newSigTlv[tlv.TlvType3]( //nolint:ll
remoteWireSig,
),
},
},
}
// As we're already in the negotiation phase, we'll now trigger
// a new iteration by having the remote party send a new offer
// sig.
closeHarness.assertSingleRemoteRbfIteration(
feeOffer, balanceAfterClose, absoluteFee, sequence,
false, true,
)
})
// Any other event should be ignored.
assertUnknownEventFail(t, startingState)
// Sending a Spend event should transition to CloseFin.
assertSpendEventCloseFin(t, startingState)
}
// TestRbfCloseErr tests that the state machine is able to properly restart
// the state machine if we encounter an error.
func TestRbfCloseErr(t *testing.T) {
localBalance := lnwire.NewMSatFromSatoshis(40_000)
remoteBalance := lnwire.NewMSatFromSatoshis(50_000)
closeTerms := &CloseChannelTerms{
ShutdownBalances: ShutdownBalances{
LocalBalance: localBalance,
RemoteBalance: remoteBalance,
},
ShutdownScripts: ShutdownScripts{
LocalDeliveryScript: localAddr,
RemoteDeliveryScript: remoteAddr,
},
}
startingState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &CloseErr{
CloseChannelTerms: closeTerms,
},
},
CloseChannelTerms: closeTerms,
}
absoluteFee := btcutil.Amount(10_100)
balanceAfterClose := localBalance.ToSatoshis() - absoluteFee
// From the error state, we should be able to kick off a new iteration
// for a local fee bump.
t.Run("send_offer_restart", func(t *testing.T) {
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some[ProtocolState](startingState),
})
defer closeHarness.stopAndAssert()
rbfFeeBump := chainfee.FeePerKwFloor.FeePerVByte()
localOffer := &SendOfferEvent{
TargetFeeRate: rbfFeeBump,
}
// Now we expect that another full RBF iteration takes place (we
// initiate a new local sig).
closeHarness.assertSingleRbfIteration(
localOffer, balanceAfterClose, absoluteFee,
noDustExpect,
)
// We should terminate in the negotiation state.
closeHarness.assertStateTransitions(
&ClosingNegotiation{},
)
})
// From the error state, we should be able to handle the remote party
// kicking off a new iteration for a fee bump.
t.Run("recv_offer_restart", func(t *testing.T) {
startingState := &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Remote: &CloseErr{
CloseChannelTerms: closeTerms,
Party: lntypes.Remote,
},
},
CloseChannelTerms: closeTerms,
}
closeHarness := newCloser(t, &harnessCfg{
initialState: fn.Some[ProtocolState](startingState),
localUpfrontAddr: fn.Some(localAddr),
})
defer closeHarness.stopAndAssert()
feeOffer := &OfferReceivedEvent{
SigMsg: lnwire.ClosingComplete{
CloserScript: remoteAddr,
CloseeScript: localAddr,
FeeSatoshis: absoluteFee,
LockTime: 1,
ClosingSigs: lnwire.ClosingSigs{
CloserAndClosee: newSigTlv[tlv.TlvType3]( //nolint:ll
remoteWireSig,
),
},
},
}
sequence := uint32(mempool.MaxRBFSequence)
// As we're already in the negotiation phase, we'll now trigger
// a new iteration by having the remote party send a new offer
// sig.
closeHarness.assertSingleRemoteRbfIteration(
feeOffer, balanceAfterClose, absoluteFee, sequence,
false, true,
)
})
// Sending a Spend event should transition to CloseFin.
assertSpendEventCloseFin(t, startingState)
}

View file

@ -1,9 +1,11 @@
package chancloser
import (
"bytes"
"fmt"
"github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/chaincfg"
"github.com/btcsuite/btcd/mempool"
"github.com/btcsuite/btcd/wire"
@ -14,11 +16,18 @@ import (
"github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnutils"
"github.com/lightningnetwork/lnd/lnwallet"
"github.com/lightningnetwork/lnd/lnwallet/chainfee"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/protofsm"
"github.com/lightningnetwork/lnd/tlv"
)
var (
// ErrInvalidStateTransition is returned if the remote party tries to
// close, but the thaw height hasn't been matched yet.
ErrThawHeightNotReached = fmt.Errorf("thaw height not reached")
)
// sendShutdownEvents is a helper function that returns a set of daemon events
// we need to emit when we decide that we should send a shutdown message. We'll
// also mark the channel as borked as well, as at this point, we no longer want
@ -104,11 +113,11 @@ func validateShutdown(chanThawHeight fn.Option[uint32],
// reject the shutdown message as we can't yet co-op close the
// channel.
if msg.BlockHeight < thawHeight {
return fmt.Errorf("initiator attempting to "+
return fmt.Errorf("%w: initiator attempting to "+
"co-op close frozen ChannelPoint(%v) "+
"(current_height=%v, thaw_height=%v)",
chanPoint, msg.BlockHeight,
thawHeight)
ErrThawHeightNotReached, chanPoint,
msg.BlockHeight, thawHeight)
}
return nil
@ -171,7 +180,7 @@ func (c *ChannelActive) ProcessEvent(event ProtocolEvent, env *Environment,
}
chancloserLog.Infof("ChannelPoint(%v): sending shutdown msg, "+
"delivery_script=%v", env.ChanPoint, shutdownScript)
"delivery_script=%x", env.ChanPoint, shutdownScript)
// From here, we'll transition to the shutdown pending state. In
// this state we await their shutdown message (self loop), then
@ -193,7 +202,8 @@ func (c *ChannelActive) ProcessEvent(event ProtocolEvent, env *Environment,
// also emit similar events like the above to send out shutdown, and
// also disable the channel.
case *ShutdownReceived:
chancloserLog.Infof("ChannelPoint(%v): received shutdown msg")
chancloserLog.Infof("ChannelPoint(%v): received shutdown msg",
env.ChanPoint)
// Validate that they can send the message now, and also that
// they haven't violated their commitment to a prior upfront
@ -204,7 +214,7 @@ func (c *ChannelActive) ProcessEvent(event ProtocolEvent, env *Environment,
)
if err != nil {
chancloserLog.Errorf("ChannelPoint(%v): rejecting "+
"shutdown attempt: %v", err)
"shutdown attempt: %v", env.ChanPoint, err)
return nil, err
}
@ -287,6 +297,22 @@ func (s *ShutdownPending) ProcessEvent(event ProtocolEvent, env *Environment,
},
}, nil
// The remote party sent an offer early. We'll go to the ChannelFlushing
// case, and then emit the offer as a internal event, which'll be
// handled as an early offer.
case *OfferReceivedEvent:
chancloserLog.Infof("ChannelPoint(%v): got an early offer "+
"in ShutdownPending, emitting as external event",
env.ChanPoint)
s.EarlyRemoteOffer = fn.Some(*msg)
// We'll perform a noop update so we can wait for the actual
// channel flushed event.
return &CloseStateTransition{
NextState: s,
}, nil
// When we receive a shutdown from the remote party, we'll validate the
// shutdown message, then transition to the ChannelFlushing state.
case *ShutdownReceived:
@ -302,7 +328,7 @@ func (s *ShutdownPending) ProcessEvent(event ProtocolEvent, env *Environment,
)
if err != nil {
chancloserLog.Errorf("ChannelPoint(%v): rejecting "+
"shutdown attempt: %v", err)
"shutdown attempt: %v", env.ChanPoint, err)
return nil, err
}
@ -310,7 +336,7 @@ func (s *ShutdownPending) ProcessEvent(event ProtocolEvent, env *Environment,
// If the channel is *already* flushed, and the close is
// go straight into negotiation, as this is the RBF loop.
// already in progress, then we can skip the flushing state and
var eventsToEmit fn.Option[protofsm.EmittedEvent[ProtocolEvent]]
var eventsToEmit []ProtocolEvent
finalBalances := env.ChanObserver.FinalBalances().UnwrapOr(
unknownBalance,
)
@ -318,11 +344,7 @@ func (s *ShutdownPending) ProcessEvent(event ProtocolEvent, env *Environment,
channelFlushed := ProtocolEvent(&ChannelFlushed{
ShutdownBalances: finalBalances,
})
eventsToEmit = fn.Some(RbfEvent{
InternalEvent: []ProtocolEvent{
channelFlushed,
},
})
eventsToEmit = append(eventsToEmit, channelFlushed)
}
chancloserLog.Infof("ChannelPoint(%v): disabling incoming adds",
@ -338,6 +360,19 @@ func (s *ShutdownPending) ProcessEvent(event ProtocolEvent, env *Environment,
chancloserLog.Infof("ChannelPoint(%v): waiting for channel to "+
"be flushed...", env.ChanPoint)
// If we received a remote offer early from the remote party,
// then we'll add that to the set of internal events to emit.
s.EarlyRemoteOffer.WhenSome(func(offer OfferReceivedEvent) {
eventsToEmit = append(eventsToEmit, &offer)
})
var newEvents fn.Option[RbfEvent]
if len(eventsToEmit) > 0 {
newEvents = fn.Some(RbfEvent{
InternalEvent: eventsToEmit,
})
}
// We transition to the ChannelFlushing state, where we await
// the ChannelFlushed event.
return &CloseStateTransition{
@ -348,7 +383,7 @@ func (s *ShutdownPending) ProcessEvent(event ProtocolEvent, env *Environment,
RemoteDeliveryScript: msg.ShutdownScript, //nolint:ll
},
},
NewEvents: eventsToEmit,
NewEvents: newEvents,
}, nil
// If we get this message, then this means that we were finally able to
@ -361,7 +396,7 @@ func (s *ShutdownPending) ProcessEvent(event ProtocolEvent, env *Environment,
// If the channel is *already* flushed, and the close is
// already in progress, then we can skip the flushing state and
// go straight into negotiation, as this is the RBF loop.
var eventsToEmit fn.Option[protofsm.EmittedEvent[ProtocolEvent]]
var eventsToEmit []ProtocolEvent
finalBalances := env.ChanObserver.FinalBalances().UnwrapOr(
unknownBalance,
)
@ -369,10 +404,19 @@ func (s *ShutdownPending) ProcessEvent(event ProtocolEvent, env *Environment,
channelFlushed := ProtocolEvent(&ChannelFlushed{
ShutdownBalances: finalBalances,
})
eventsToEmit = fn.Some(RbfEvent{
InternalEvent: []ProtocolEvent{
channelFlushed,
},
eventsToEmit = append(eventsToEmit, channelFlushed)
}
// If we received a remote offer early from the remote party,
// then we'll add that to the set of internal events to emit.
s.EarlyRemoteOffer.WhenSome(func(offer OfferReceivedEvent) {
eventsToEmit = append(eventsToEmit, &offer)
})
var newEvents fn.Option[RbfEvent]
if len(eventsToEmit) > 0 {
newEvents = fn.Some(RbfEvent{
InternalEvent: eventsToEmit,
})
}
@ -383,7 +427,7 @@ func (s *ShutdownPending) ProcessEvent(event ProtocolEvent, env *Environment,
IdealFeeRate: s.IdealFeeRate,
ShutdownScripts: s.ShutdownScripts,
},
NewEvents: eventsToEmit,
NewEvents: newEvents,
}, nil
// Any other messages in this state will result in an error, as this is
@ -423,9 +467,6 @@ func (c *ChannelFlushing) ProcessEvent(event ProtocolEvent, env *Environment,
c.EarlyRemoteOffer = fn.Some(*msg)
// TODO(roasbeef): unit test!
// * actually do this ^
// We'll perform a noop update so we can wait for the actual
// channel flushed event.
return &CloseStateTransition{
@ -466,8 +507,6 @@ func (c *ChannelFlushing) ProcessEvent(event ProtocolEvent, env *Environment,
// We'll then use that fee rate to determine the absolute fee
// we'd propose.
//
// TODO(roasbeef): need to sign the 3 diff versions of this?
localTxOut, remoteTxOut := closeTerms.DeriveCloseTxOuts()
absoluteFee := env.FeeEstimator.EstimateFee(
env.ChanType, localTxOut, remoteTxOut,
@ -519,12 +558,13 @@ func (c *ChannelFlushing) ProcessEvent(event ProtocolEvent, env *Environment,
NextState: &ClosingNegotiation{
PeerState: lntypes.Dual[AsymmetricPeerState]{
Local: &LocalCloseStart{
CloseChannelTerms: closeTerms,
CloseChannelTerms: &closeTerms,
},
Remote: &RemoteCloseStart{
CloseChannelTerms: closeTerms,
CloseChannelTerms: &closeTerms,
},
},
CloseChannelTerms: &closeTerms,
},
NewEvents: newEvents,
}, nil
@ -568,6 +608,63 @@ func processNegotiateEvent(c *ClosingNegotiation, event ProtocolEvent,
}, nil
}
// updateAndValidateCloseTerms is a helper function that validates examines the
// incoming event, and decide if we need to update the remote party's address,
// or reject it if it doesn't include our latest address.
func (c *ClosingNegotiation) updateAndValidateCloseTerms(event ProtocolEvent,
) error {
assertLocalScriptMatches := func(localScriptInMsg []byte) error {
if !bytes.Equal(
c.LocalDeliveryScript, localScriptInMsg,
) {
return fmt.Errorf("%w: remote party sent wrong "+
"script, expected %x, got %x",
ErrWrongLocalScript, c.LocalDeliveryScript,
localScriptInMsg,
)
}
return nil
}
switch msg := event.(type) {
// The remote party is sending us a new request to counter sign their
// version of the commitment transaction.
case *OfferReceivedEvent:
// Make sure that they're sending our local script, and not
// something else.
err := assertLocalScriptMatches(msg.SigMsg.CloseeScript)
if err != nil {
return err
}
oldRemoteAddr := c.RemoteDeliveryScript
newRemoteAddr := msg.SigMsg.CloserScript
// If they're sending a new script, then we'll update to the new
// one.
if !bytes.Equal(oldRemoteAddr, newRemoteAddr) {
c.RemoteDeliveryScript = newRemoteAddr
}
// The remote party responded to our sig request with a signature for
// our version of the commitment transaction.
case *LocalSigReceived:
// Make sure that they're sending our local script, and not
// something else.
err := assertLocalScriptMatches(msg.SigMsg.CloserScript)
if err != nil {
return err
}
return nil
}
return nil
}
// ProcessEvent drives forward the composite states for the local and remote
// party in response to new events. From this state, we'll continue to drive
// forward the local and remote states until we arrive at the StateFin stage,
@ -579,6 +676,12 @@ func (c *ClosingNegotiation) ProcessEvent(event ProtocolEvent, env *Environment,
// we receive a confirmation event, or we receive a signal to restart
// the co-op close process.
switch msg := event.(type) {
// Ignore any potential duplicate channel flushed events.
case *ChannelFlushed:
return &CloseStateTransition{
NextState: c,
}, nil
// If we get a confirmation, then the spend request we issued when we
// were leaving the ChannelFlushing state has been confirmed. We'll
// now transition to the StateFin state.
@ -588,45 +691,45 @@ func (c *ClosingNegotiation) ProcessEvent(event ProtocolEvent, env *Environment,
ConfirmedTx: msg.Tx,
},
}, nil
}
// Otherwise, if we receive a shutdown, or receive an event to send a
// shutdown, then we'll go back up to the ChannelActive state, and have
// it handle this event by emitting an internal event.
//
// TODO(roasbeef): both will have fee rate specified, so ok?
case *ShutdownReceived, *SendShutdown:
chancloserLog.Infof("ChannelPoint(%v): RBF case triggered, "+
"restarting negotiation", env.ChanPoint)
// At this point, we know its a new signature message. We'll validate,
// and maybe update the set of close terms based on what we receive. We
// might update the remote party's address for example.
if err := c.updateAndValidateCloseTerms(event); err != nil {
return nil, fmt.Errorf("event violates close terms: %w", err)
}
return &CloseStateTransition{
NextState: &ChannelActive{},
NewEvents: fn.Some(RbfEvent{
InternalEvent: []ProtocolEvent{event},
}),
}, nil
shouldRouteTo := func(party lntypes.ChannelParty) bool {
state := c.PeerState.GetForParty(party)
if state == nil {
return false
}
return state.ShouldRouteTo(event)
}
// If we get to this point, then we have an event that'll drive forward
// the negotiation process. Based on the event, we'll figure out which
// state we'll be modifying.
switch {
case c.PeerState.GetForParty(lntypes.Local).ShouldRouteTo(event):
case shouldRouteTo(lntypes.Local):
chancloserLog.Infof("ChannelPoint(%v): routing %T to local "+
"chan state", env.ChanPoint, event)
// Drive forward the local state based on the next event.
return processNegotiateEvent(c, event, env, lntypes.Local)
case c.PeerState.GetForParty(lntypes.Remote).ShouldRouteTo(event):
case shouldRouteTo(lntypes.Remote):
chancloserLog.Infof("ChannelPoint(%v): routing %T to remote "+
"chan state", env.ChanPoint, event)
"chan state", env.ChanPoint, event)
// Drive forward the remote state based on the next event.
return processNegotiateEvent(c, event, env, lntypes.Remote)
}
return nil, fmt.Errorf("%w: received %T while in ClosingNegotiation",
ErrInvalidStateTransition, event)
return nil, fmt.Errorf("%w: received %T while in %v",
ErrInvalidStateTransition, event, c)
}
// newSigTlv is a helper function that returns a new optional TLV sig field for
@ -645,14 +748,34 @@ func (l *LocalCloseStart) ProcessEvent(event ProtocolEvent, env *Environment,
// rate to generate for the closing transaction with our ideal fee
// rate.
case *SendOfferEvent:
// First, we'll figure out the absolute fee rate we should pay
// given the state of the local/remote outputs.
// First, we'll figure out the absolute fee rate we should pay
localTxOut, remoteTxOut := l.DeriveCloseTxOuts()
absoluteFee := env.FeeEstimator.EstimateFee(
env.ChanType, localTxOut, remoteTxOut,
msg.TargetFeeRate.FeePerKWeight(),
)
// If we can't actually pay for fees here, then we'll just do a
// noop back to the same state to await a new fee rate.
if !l.LocalCanPayFees(absoluteFee) {
chancloserLog.Infof("ChannelPoint(%v): unable to pay "+
"fee=%v with local balance %v, skipping "+
"closing_complete", env.ChanPoint, absoluteFee,
l.LocalBalance)
return &CloseStateTransition{
NextState: &CloseErr{
CloseChannelTerms: l.CloseChannelTerms,
Party: lntypes.Local,
ErrState: NewErrStateCantPayForFee(
l.LocalBalance.ToSatoshis(),
absoluteFee,
),
},
}, nil
}
// Now that we know what fee we want to pay, we'll create a new
// signature over our co-op close transaction. For our
// proposals, we'll just always use the known RBF sequence
@ -713,12 +836,13 @@ func (l *LocalCloseStart) ProcessEvent(event ProtocolEvent, env *Environment,
// TODO(roasbeef): type alias for protocol event
sendEvent := protofsm.DaemonEventSet{&protofsm.SendMsgEvent[ProtocolEvent]{ //nolint:ll
TargetPeer: env.ChanPeer,
// TODO(roasbeef): mew new func
Msgs: []lnwire.Message{&lnwire.ClosingComplete{
ChannelID: env.ChanID,
FeeSatoshis: absoluteFee,
LockTime: env.BlockHeight,
ClosingSigs: closingSigs,
ChannelID: env.ChanID,
CloserScript: l.LocalDeliveryScript,
CloseeScript: l.RemoteDeliveryScript,
FeeSatoshis: absoluteFee,
LockTime: env.BlockHeight,
ClosingSigs: closingSigs,
}},
}}
@ -729,6 +853,7 @@ func (l *LocalCloseStart) ProcessEvent(event ProtocolEvent, env *Environment,
return &CloseStateTransition{
NextState: &LocalOfferSent{
ProposedFee: absoluteFee,
ProposedFeeRate: msg.TargetFeeRate,
LocalSig: wireSig,
CloseChannelTerms: l.CloseChannelTerms,
},
@ -837,7 +962,10 @@ func (l *LocalOfferSent) ProcessEvent(event ProtocolEvent, env *Environment,
return &CloseStateTransition{
NextState: &ClosePending{
CloseTx: closeTx,
CloseTx: closeTx,
FeeRate: l.ProposedFeeRate,
CloseChannelTerms: l.CloseChannelTerms,
Party: lntypes.Local,
},
NewEvents: fn.Some(protofsm.EmittedEvent[ProtocolEvent]{
ExternalEvents: broadcastEvent,
@ -980,8 +1108,12 @@ func (l *RemoteCloseStart) ProcessEvent(event ProtocolEvent, env *Environment,
sendEvent := &protofsm.SendMsgEvent[ProtocolEvent]{
TargetPeer: env.ChanPeer,
Msgs: []lnwire.Message{&lnwire.ClosingSig{
ChannelID: env.ChanID,
ClosingSigs: closingSigs,
ChannelID: env.ChanID,
CloserScript: l.RemoteDeliveryScript,
CloseeScript: l.LocalDeliveryScript,
FeeSatoshis: msg.SigMsg.FeeSatoshis,
LockTime: msg.SigMsg.LockTime,
ClosingSigs: closingSigs,
}},
}
broadcastEvent := &protofsm.BroadcastTxn{
@ -994,11 +1126,22 @@ func (l *RemoteCloseStart) ProcessEvent(event ProtocolEvent, env *Environment,
sendEvent, broadcastEvent,
}
// We'll also compute the final fee rate that the remote party
// paid based off the absolute fee and the size of the closing
// transaction.
vSize := mempool.GetTxVirtualSize(btcutil.NewTx(closeTx))
feeRate := chainfee.SatPerVByte(
int64(msg.SigMsg.FeeSatoshis) / vSize,
)
// Now that we've extracted the signature, we'll transition to
// the next state where we'll sign+broadcast the sig.
return &CloseStateTransition{
NextState: &ClosePending{
CloseTx: closeTx,
CloseTx: closeTx,
FeeRate: feeRate,
CloseChannelTerms: l.CloseChannelTerms,
Party: lntypes.Remote,
},
NewEvents: fn.Some(protofsm.EmittedEvent[ProtocolEvent]{
ExternalEvents: daemonEvents,
@ -1026,6 +1169,32 @@ func (c *ClosePending) ProcessEvent(event ProtocolEvent, env *Environment,
},
}, nil
// If we get a send offer event in this state, then we're doing a state
// transition to the LocalCloseStart state, so we can sign a new closing
// tx.
case *SendOfferEvent:
return &CloseStateTransition{
NextState: &LocalCloseStart{
CloseChannelTerms: c.CloseChannelTerms,
},
NewEvents: fn.Some(protofsm.EmittedEvent[ProtocolEvent]{
InternalEvent: []ProtocolEvent{msg},
}),
}, nil
// If we get an offer received event, then we're doing a state
// transition to the RemoteCloseStart, as the remote peer wants to sign
// a new closing tx.
case *OfferReceivedEvent:
return &CloseStateTransition{
NextState: &RemoteCloseStart{
CloseChannelTerms: c.CloseChannelTerms,
},
NewEvents: fn.Some(protofsm.EmittedEvent[ProtocolEvent]{
InternalEvent: []ProtocolEvent{msg},
}),
}, nil
default:
return &CloseStateTransition{
@ -1043,3 +1212,44 @@ func (c *CloseFin) ProcessEvent(event ProtocolEvent, env *Environment,
NextState: c,
}, nil
}
// ProcessEvent is a semi-terminal state in the rbf-coop close state machine.
// In this state, we hit a validation error in an earlier state, so we'll remain
// in this state for the user to examine. We may also process new requests to
// continue the state machine.
func (c *CloseErr) ProcessEvent(event ProtocolEvent, env *Environment,
) (*CloseStateTransition, error) {
switch msg := event.(type) {
// If we get a send offer event in this state, then we're doing a state
// transition to the LocalCloseStart state, so we can sign a new closing
// tx.
case *SendOfferEvent:
return &CloseStateTransition{
NextState: &LocalCloseStart{
CloseChannelTerms: c.CloseChannelTerms,
},
NewEvents: fn.Some(protofsm.EmittedEvent[ProtocolEvent]{
InternalEvent: []ProtocolEvent{msg},
}),
}, nil
// If we get an offer received event, then we're doing a state
// transition to the RemoteCloseStart, as the remote peer wants to sign
// a new closing tx.
case *OfferReceivedEvent:
return &CloseStateTransition{
NextState: &RemoteCloseStart{
CloseChannelTerms: c.CloseChannelTerms,
},
NewEvents: fn.Some(protofsm.EmittedEvent[ProtocolEvent]{
InternalEvent: []ProtocolEvent{msg},
}),
}, nil
default:
return &CloseStateTransition{
NextState: c,
}, nil
}
}

View file

@ -9269,13 +9269,19 @@ func CreateCooperativeCloseTx(fundingTxIn wire.TxIn,
closeTx.LockTime = lockTime
})
// TODO(roasbeef): needs support for dropping inputs
// Create both cooperative closure outputs, properly respecting the
// dust limits of both parties.
// Create both cooperative closure outputs, properly respecting the dust
// limits of both parties.
var localOutputIdx fn.Option[int]
haveLocalOutput := ourBalance >= localDust
if haveLocalOutput {
// If our script is an OP_RETURN, then we set our balance to
// zero.
if opts.customSequence.IsSome() &&
input.ScriptIsOpReturn(ourDeliveryScript) {
ourBalance = 0
}
closeTx.AddTxOut(&wire.TxOut{
PkScript: ourDeliveryScript,
Value: int64(ourBalance),
@ -9287,6 +9293,14 @@ func CreateCooperativeCloseTx(fundingTxIn wire.TxIn,
var remoteOutputIdx fn.Option[int]
haveRemoteOutput := theirBalance >= remoteDust
if haveRemoteOutput {
// If a party's script is an OP_RETURN, then we set their
// balance to zero.
if opts.customSequence.IsSome() &&
input.ScriptIsOpReturn(theirDeliveryScript) {
theirBalance = 0
}
closeTx.AddTxOut(&wire.TxOut{
PkScript: theirDeliveryScript,
Value: int64(theirBalance),
@ -9874,6 +9888,14 @@ func (lc *LightningChannel) FundingTxOut() *wire.TxOut {
return &lc.fundingOutput
}
// DeriveHeightHint derives the block height for the channel opening.
func (lc *LightningChannel) DeriveHeightHint() uint32 {
lc.RLock()
defer lc.RUnlock()
return lc.channelState.DeriveHeightHint()
}
// MultiSigKeys returns the set of multi-sig keys for an channel.
func (lc *LightningChannel) MultiSigKeys() (keychain.KeyDescriptor,
keychain.KeyDescriptor) {
@ -9913,3 +9935,16 @@ func (lc *LightningChannel) FundingBlob() fn.Option[tlv.Blob] {
return newBlob
})(lc.channelState.CustomBlob)
}
// ZeroConfRealScid returns an optional real scid for the channel. If this
// returns None, then this isn't a zero conf channel. Otherwise, the real scid
// value will be returned.
//
//nolint:ll
func (lc *LightningChannel) ZeroConfRealScid() fn.Option[lnwire.ShortChannelID] {
if lc.channelState.IsZeroConf() {
return fn.Some(lc.channelState.ZeroConfRealScid())
}
return fn.None[lnwire.ShortChannelID]()
}

View file

@ -20,6 +20,7 @@ import (
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/btcutil/txsort"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/mempool"
"github.com/btcsuite/btcd/txscript"
"github.com/btcsuite/btcd/wire"
"github.com/davecgh/go-spew/spew"
@ -2450,6 +2451,72 @@ func TestCooperativeCloseDustAdherence(t *testing.T) {
}
}
// TestCooperativeCloseOpReturn tests that if either party's script is an
// OP_RETURN script, then we'll set their output value as zero on the closing
// transaction.
func TestCooperativeCloseOpReturn(t *testing.T) {
t.Parallel()
// Create a test channel which will be used for the duration of this
// unittest. The channel will be funded evenly with Alice having 5 BTC,
// and Bob having 5 BTC.
aliceChannel, bobChannel, err := CreateTestChannels(
t, channeldb.SingleFunderTweaklessBit,
)
require.NoError(t, err, "unable to create test channels")
// Alice will have a "normal" looking script, while Bob will have a
// script that's just an OP_RETURN.
aliceDeliveryScript := bobsPrivKey
bobDeliveryScript := []byte{txscript.OP_RETURN}
aliceFeeRate := chainfee.SatPerKWeight(
aliceChannel.channelState.LocalCommitment.FeePerKw,
)
aliceFee := aliceChannel.CalcFee(aliceFeeRate) + 1000
assertBobOpReturn := func(tx *wire.MsgTx) {
// We should still have two outputs on the commitment
// transaction, as Alice's is non-dust.
require.Len(t, tx.TxOut, 2)
// We should find that Bob's output has a zero value.
bobTxOut := fn.Filter(tx.TxOut, func(txOut *wire.TxOut) bool {
return bytes.Equal(txOut.PkScript, bobDeliveryScript)
})
require.Len(t, bobTxOut, 1)
require.True(t, bobTxOut[0].Value == 0)
}
// Next, we'll make a new co-op close proposal, initiated by Alice.
aliceSig, closeTxAlice, _, err := aliceChannel.CreateCloseProposal(
aliceFee, aliceDeliveryScript, bobDeliveryScript,
// We use a custom sequence as this rule only applies to the RBF
// coop channel type.
WithCustomSequence(mempool.MaxRBFSequence),
)
require.NoError(t, err, "unable to close channel")
assertBobOpReturn(closeTxAlice)
bobSig, _, _, err := bobChannel.CreateCloseProposal(
aliceFee, bobDeliveryScript, aliceDeliveryScript,
WithCustomSequence(mempool.MaxRBFSequence),
)
require.NoError(t, err, "unable to close channel")
// We should now be able to complete the cooperative channel closure,
// finding that the close tx still only has a single output.
closeTx, _, err := bobChannel.CompleteCooperativeClose(
bobSig, aliceSig, bobDeliveryScript, aliceDeliveryScript,
aliceFee, WithCustomSequence(mempool.MaxRBFSequence),
)
require.NoError(t, err, "unable to accept channel close")
assertBobOpReturn(closeTx)
}
// TestUpdateFeeAdjustments tests that the state machine is able to properly
// accept valid fee changes, as well as reject any invalid fee updates.
func TestUpdateFeeAdjustments(t *testing.T) {

View file

@ -30,6 +30,14 @@ type ClosingComplete struct {
// ChannelID serves to identify which channel is to be closed.
ChannelID ChannelID
// CloserScript is the script to which the channel funds will be paid
// for the closer (the person sending the ClosingComplete) message.
CloserScript DeliveryAddress
// CloseeScript is the script to which the channel funds will be paid
// (the person receiving the ClosingComplete message).
CloseeScript DeliveryAddress
// FeeSatoshis is the total fee in satoshis that the party to the
// channel would like to propose for the close transaction.
FeeSatoshis btcutil.Amount
@ -79,7 +87,10 @@ func decodeClosingSigs(c *ClosingSigs, tlvRecords ExtraOpaqueData) error {
// passed io.Reader.
func (c *ClosingComplete) Decode(r io.Reader, _ uint32) error {
// First, read out all the fields that are hard coded into the message.
err := ReadElements(r, &c.ChannelID, &c.FeeSatoshis, &c.LockTime)
err := ReadElements(
r, &c.ChannelID, &c.CloserScript, &c.CloseeScript,
&c.FeeSatoshis, &c.LockTime,
)
if err != nil {
return err
}
@ -125,6 +136,13 @@ func (c *ClosingComplete) Encode(w *bytes.Buffer, _ uint32) error {
return err
}
if err := WriteDeliveryAddress(w, c.CloserScript); err != nil {
return err
}
if err := WriteDeliveryAddress(w, c.CloseeScript); err != nil {
return err
}
if err := WriteSatoshi(w, c.FeeSatoshis); err != nil {
return err
}

View file

@ -3,6 +3,8 @@ package lnwire
import (
"bytes"
"io"
"github.com/btcsuite/btcd/btcutil"
)
// ClosingSig is sent in response to a ClosingComplete message. It carries the
@ -11,6 +13,22 @@ type ClosingSig struct {
// ChannelID serves to identify which channel is to be closed.
ChannelID ChannelID
// CloserScript is the script to which the channel funds will be paid
// for the closer (the person sending the ClosingComplete) message.
CloserScript DeliveryAddress
// CloseeScript is the script to which the channel funds will be paid
// (the person receiving the ClosingComplete message).
CloseeScript DeliveryAddress
// FeeSatoshis is the total fee in satoshis that the party to the
// channel proposed for the close transaction.
FeeSatoshis btcutil.Amount
// LockTime is the locktime number to be used in the input spending the
// funding transaction.
LockTime uint32
// ClosingSigs houses the 3 possible signatures that can be sent.
ClosingSigs
@ -24,7 +42,10 @@ type ClosingSig struct {
// io.Reader.
func (c *ClosingSig) Decode(r io.Reader, _ uint32) error {
// First, read out all the fields that are hard coded into the message.
err := ReadElements(r, &c.ChannelID)
err := ReadElements(
r, &c.ChannelID, &c.CloserScript, &c.CloseeScript,
&c.FeeSatoshis, &c.LockTime,
)
if err != nil {
return err
}
@ -53,6 +74,21 @@ func (c *ClosingSig) Encode(w *bytes.Buffer, _ uint32) error {
return err
}
if err := WriteDeliveryAddress(w, c.CloserScript); err != nil {
return err
}
if err := WriteDeliveryAddress(w, c.CloseeScript); err != nil {
return err
}
if err := WriteSatoshi(w, c.FeeSatoshis); err != nil {
return err
}
if err := WriteUint32(w, c.LockTime); err != nil {
return err
}
recordProducers := closingSigRecords(&c.ClosingSigs)
err := EncodeMessageExtraData(&c.ExtraData, recordProducers...)

View file

@ -233,6 +233,22 @@ const (
// able and willing to accept keysend payments.
KeysendOptional = 55
// RbfCoopCloseRequired is a required feature bit that signals that
// the new RBF-based co-op close protocol is supported.
RbfCoopCloseRequired = 60
// RbfCoopCloseOptional is an optional feature bit that signals that the
// new RBF-based co-op close protocol is supported.
RbfCoopCloseOptional = 61
// RbfCoopCloseRequiredStaging is a required feature bit that signals
// that the new RBF-based co-op close protocol is supported.
RbfCoopCloseRequiredStaging = 160
// RbfCoopCloseOptionalStaging is an optional feature bit that signals
// that the new RBF-based co-op close protocol is supported.
RbfCoopCloseOptionalStaging = 161
// ScriptEnforcedLeaseRequired is a required feature bit that signals
// that the node requires channels having zero-fee second-level HTLC
// transactions, which also imply anchor commitments, along with an
@ -373,6 +389,10 @@ var Features = map[FeatureBit]string{
ExperimentalEndorsementOptional: "endorsement-x",
Bolt11BlindedPathsOptional: "bolt-11-blinded-paths",
Bolt11BlindedPathsRequired: "bolt-11-blinded-paths",
RbfCoopCloseOptional: "rbf-coop-close",
RbfCoopCloseRequired: "rbf-coop-close",
RbfCoopCloseOptionalStaging: "rbf-coop-close-x",
RbfCoopCloseRequiredStaging: "rbf-coop-close-x",
}
// RawFeatureVector represents a set of feature bits as defined in BOLT-09. A

View file

@ -1355,6 +1355,18 @@ func TestLightningWireProtocol(t *testing.T) {
LockTime: uint32(r.Int63()),
ClosingSigs: ClosingSigs{},
}
req.CloserScript, err = randDeliveryAddress(r)
if err != nil {
t.Fatalf("unable to generate delivery "+
"address: %v", err)
return
}
req.CloseeScript, err = randDeliveryAddress(r)
if err != nil {
t.Fatalf("unable to generate delivery "+
"address: %v", err)
return
}
if r.Intn(2) == 0 {
sig := req.CloserNoClosee.Zero()
@ -1403,6 +1415,20 @@ func TestLightningWireProtocol(t *testing.T) {
req := ClosingSig{
ChannelID: ChannelID(c),
ClosingSigs: ClosingSigs{},
FeeSatoshis: btcutil.Amount(r.Int63()),
LockTime: uint32(r.Int63()),
}
req.CloserScript, err = randDeliveryAddress(r)
if err != nil {
t.Fatalf("unable to generate delivery "+
"address: %v", err)
return
}
req.CloseeScript, err = randDeliveryAddress(r)
if err != nil {
t.Fatalf("unable to generate delivery "+
"address: %v", err)
return
}
if r.Intn(2) == 0 {

4
log.go
View file

@ -44,9 +44,11 @@ import (
"github.com/lightningnetwork/lnd/lnwallet/chanfunding"
"github.com/lightningnetwork/lnd/lnwallet/rpcwallet"
"github.com/lightningnetwork/lnd/monitoring"
"github.com/lightningnetwork/lnd/msgmux"
"github.com/lightningnetwork/lnd/netann"
"github.com/lightningnetwork/lnd/peer"
"github.com/lightningnetwork/lnd/peernotifier"
"github.com/lightningnetwork/lnd/protofsm"
"github.com/lightningnetwork/lnd/routing"
"github.com/lightningnetwork/lnd/routing/blindedpath"
"github.com/lightningnetwork/lnd/routing/localchans"
@ -178,6 +180,7 @@ func SetupLoggers(root *build.SubLoggerManager, interceptor signal.Interceptor)
AddSubLogger(root, "PEER", interceptor, peer.UseLogger)
AddSubLogger(root, "CHCL", interceptor, chancloser.UseLogger)
AddSubLogger(root, "LCHN", interceptor, localchans.UseLogger)
AddSubLogger(root, "PFSM", interceptor, protofsm.UseLogger)
AddSubLogger(root, routing.Subsystem, interceptor, routing.UseLogger)
AddSubLogger(root, routerrpc.Subsystem, interceptor, routerrpc.UseLogger)
@ -200,6 +203,7 @@ func SetupLoggers(root *build.SubLoggerManager, interceptor signal.Interceptor)
)
AddV1SubLogger(root, graphdb.Subsystem, interceptor, graphdb.UseLogger)
AddSubLogger(root, chainio.Subsystem, interceptor, chainio.UseLogger)
AddSubLogger(root, msgmux.Subsystem, interceptor, msgmux.UseLogger)
}
// AddSubLogger is a helper method to conveniently create and register the

View file

@ -1,6 +1,7 @@
package msgmux
import (
"context"
"fmt"
"maps"
"sync"
@ -46,7 +47,7 @@ type Endpoint interface {
// SendMessage handles the target message, and returns true if the
// message was able being processed.
SendMessage(msg PeerMsg) bool
SendMessage(ctx context.Context, msg PeerMsg) bool
}
// MsgRouter is an interface that represents a message router, which is generic
@ -66,7 +67,7 @@ type Router interface {
RouteMsg(PeerMsg) error
// Start starts the peer message router.
Start()
Start(ctx context.Context)
// Stop stops the peer message router.
Stop()
@ -137,12 +138,12 @@ func NewMultiMsgRouter() *MultiMsgRouter {
}
// Start starts the peer message router.
func (p *MultiMsgRouter) Start() {
func (p *MultiMsgRouter) Start(ctx context.Context) {
log.Infof("Starting Router")
p.startOnce.Do(func() {
p.wg.Add(1)
go p.msgRouter()
go p.msgRouter(ctx)
})
}
@ -179,7 +180,7 @@ func (p *MultiMsgRouter) endpoints() fn.Result[EndpointsMap] {
}
// msgRouter is the main goroutine that handles all incoming messages.
func (p *MultiMsgRouter) msgRouter() {
func (p *MultiMsgRouter) msgRouter(ctx context.Context) {
defer p.wg.Done()
// endpoints is a map of all registered endpoints.
@ -235,7 +236,7 @@ func (p *MultiMsgRouter) msgRouter() {
"msg %T to endpoint %s", msg,
endpoint.Name())
sent := endpoint.SendMessage(msg)
sent := endpoint.SendMessage(ctx, msg)
couldSend = couldSend || sent
}
}
@ -243,7 +244,7 @@ func (p *MultiMsgRouter) msgRouter() {
var err error
if !couldSend {
log.Tracef("MsgRouter: unable to route "+
"msg %T", msg)
"msg %T", msg.Message)
err = ErrUnableToRouteMsg
}

View file

@ -1,6 +1,7 @@
package msgmux
import (
"context"
"testing"
"github.com/lightningnetwork/lnd/lnwire"
@ -24,8 +25,8 @@ func (m *mockEndpoint) CanHandle(msg PeerMsg) bool {
return args.Bool(0)
}
func (m *mockEndpoint) SendMessage(msg PeerMsg) bool {
args := m.Called(msg)
func (m *mockEndpoint) SendMessage(ctx context.Context, msg PeerMsg) bool {
args := m.Called(ctx, msg)
return args.Bool(0)
}
@ -33,8 +34,9 @@ func (m *mockEndpoint) SendMessage(msg PeerMsg) bool {
// TestMessageRouterOperation tests the basic operation of the message router:
// add new endpoints, route to them, remove, them, etc.
func TestMessageRouterOperation(t *testing.T) {
ctx := context.Background()
msgRouter := NewMultiMsgRouter()
msgRouter.Start()
msgRouter.Start(ctx)
defer msgRouter.Stop()
openChanMsg := PeerMsg{
@ -57,7 +59,7 @@ func TestMessageRouterOperation(t *testing.T) {
fundingEndpoint.On("CanHandle", openChanMsg).Return(true)
fundingEndpoint.On("CanHandle", errorMsg).Return(false)
fundingEndpoint.On("CanHandle", commitSigMsg).Return(false)
fundingEndpoint.On("SendMessage", openChanMsg).Return(true)
fundingEndpoint.On("SendMessage", ctx, openChanMsg).Return(true)
commitEndpoint := &mockEndpoint{}
commitEndpointName := "commit"
@ -65,7 +67,7 @@ func TestMessageRouterOperation(t *testing.T) {
commitEndpoint.On("CanHandle", commitSigMsg).Return(true)
commitEndpoint.On("CanHandle", openChanMsg).Return(false)
commitEndpoint.On("CanHandle", errorMsg).Return(false)
commitEndpoint.On("SendMessage", commitSigMsg).Return(true)
commitEndpoint.On("SendMessage", ctx, commitSigMsg).Return(true)
t.Run("add endpoints", func(t *testing.T) {
// First, we'll add the funding endpoint to the router.
@ -113,8 +115,10 @@ func TestMessageRouterOperation(t *testing.T) {
fundingEndpoint.AssertCalled(t, "CanHandle", openChanMsg)
commitEndpoint.AssertCalled(t, "CanHandle", openChanMsg)
fundingEndpoint.AssertCalled(t, "SendMessage", openChanMsg)
commitEndpoint.AssertNotCalled(t, "SendMessage", openChanMsg)
fundingEndpoint.AssertCalled(t, "SendMessage", ctx, openChanMsg)
commitEndpoint.AssertNotCalled(
t, "SendMessage", ctx, openChanMsg,
)
// We'll do the same for the commit sig message.
require.NoError(t, msgRouter.RouteMsg(commitSigMsg))
@ -122,8 +126,10 @@ func TestMessageRouterOperation(t *testing.T) {
fundingEndpoint.AssertCalled(t, "CanHandle", commitSigMsg)
commitEndpoint.AssertCalled(t, "CanHandle", commitSigMsg)
commitEndpoint.AssertCalled(t, "SendMessage", commitSigMsg)
fundingEndpoint.AssertNotCalled(t, "SendMessage", commitSigMsg)
commitEndpoint.AssertCalled(t, "SendMessage", ctx, commitSigMsg)
fundingEndpoint.AssertNotCalled(
t, "SendMessage", ctx, commitSigMsg,
)
})
t.Run("remove endpoints", func(t *testing.T) {

File diff suppressed because it is too large Load diff

View file

@ -689,15 +689,9 @@ func TestChooseDeliveryScript(t *testing.T) {
userScript lnwire.DeliveryAddress
shutdownScript lnwire.DeliveryAddress
expectedScript lnwire.DeliveryAddress
newAddr func() ([]byte, error)
expectedError error
}{
{
name: "Neither set",
userScript: nil,
shutdownScript: nil,
expectedScript: nil,
expectedError: nil,
},
{
name: "Both set and equal",
userScript: script1,
@ -726,6 +720,16 @@ func TestChooseDeliveryScript(t *testing.T) {
expectedScript: script2,
expectedError: nil,
},
{
name: "no script generate new one",
userScript: nil,
shutdownScript: nil,
expectedScript: script2,
newAddr: func() ([]byte, error) {
return script2, nil
},
expectedError: nil,
},
}
for _, test := range tests {
@ -734,13 +738,16 @@ func TestChooseDeliveryScript(t *testing.T) {
t.Run(test.name, func(t *testing.T) {
script, err := chooseDeliveryScript(
test.shutdownScript, test.userScript,
test.newAddr,
)
if err != test.expectedError {
t.Fatalf("Expected: %v, got: %v", test.expectedError, err)
t.Fatalf("Expected: %v, got: %v",
test.expectedError, err)
}
if !bytes.Equal(script, test.expectedScript) {
t.Fatalf("Expected: %x, got: %x", test.expectedScript, script)
t.Fatalf("Expected: %x, got: %x",
test.expectedScript, script)
}
})
}
@ -853,8 +860,10 @@ func TestCustomShutdownScript(t *testing.T) {
t.Fatalf("did not receive shutdown message")
case err := <-errChan:
// Fail if we do not expect an error.
if err != test.expectedError {
t.Fatalf("error closing channel: %v", err)
if test.expectedError != nil {
require.ErrorIs(
t, err, test.expectedError,
)
}
// Terminate the test early if have received an error, no

174
peer/chan_observer.go Normal file
View file

@ -0,0 +1,174 @@
package peer
import (
"github.com/btcsuite/btcd/wire"
"github.com/lightningnetwork/lnd/channeldb"
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/lightningnetwork/lnd/htlcswitch"
"github.com/lightningnetwork/lnd/lntypes"
"github.com/lightningnetwork/lnd/lnwallet/chancloser"
)
// channelView is a view into the current active/global channel state machine
// for a given link.
type channelView interface {
// OweCommitment returns a boolean value reflecting whether we need to
// send out a commitment signature because there are outstanding local
// updates and/or updates in the local commit tx that aren't reflected
// in the remote commit tx yet.
OweCommitment() bool
// IsChannelClean returns true if neither side has pending commitments,
// neither side has HTLC's, and all updates are locked in irrevocably.
IsChannelClean() bool
// MarkCoopBroadcasted persistently marks that the channel close
// transaction has been broadcast.
MarkCoopBroadcasted(*wire.MsgTx, lntypes.ChannelParty) error
// StateSnapshot returns a snapshot of the current fully committed
// state within the channel.
StateSnapshot() *channeldb.ChannelSnapshot
// MarkShutdownSent persists the given ShutdownInfo. The existence of
// the ShutdownInfo represents the fact that the Shutdown message has
// been sent by us and so should be re-sent on re-establish.
MarkShutdownSent(info *channeldb.ShutdownInfo) error
}
// linkController is capable of controlling the flow out incoming/outgoing
// HTLCs to/from the link.
type linkController interface {
// DisableAdds sets the ChannelUpdateHandler state to allow/reject
// UpdateAddHtlc's in the specified direction. It returns true if the
// state was changed and false if the desired state was already set
// before the method was called.
DisableAdds(outgoing bool) bool
// IsFlushing returns true when UpdateAddHtlc's are disabled in the
// direction of the argument.
IsFlushing(direction bool) bool
}
// linkNetworkController is an interface that represents an object capable of
// managing interactions with the active channel links from the PoV of the
// gossip network.
type linkNetworkController interface {
// RequestDisable disables a channel by its channel point.
RequestDisable(wire.OutPoint, bool) error
}
// chanObserver implements the chancloser.ChanObserver interface for the
// existing LightningChannel struct/instance.
type chanObserver struct {
chanView channelView
link linkController
linkNetwork linkNetworkController
}
// newChanObserver creates a new instance of a chanObserver from an active
// channelView.
func newChanObserver(chanView channelView,
link linkController, linkNetwork linkNetworkController) *chanObserver {
return &chanObserver{
chanView: chanView,
link: link,
linkNetwork: linkNetwork,
}
}
// NoDanglingUpdates returns true if there are no dangling updates in the
// channel. In other words, there are no active update messages that haven't
// already been covered by a commit sig.
func (l *chanObserver) NoDanglingUpdates() bool {
return !l.chanView.OweCommitment()
}
// DisableIncomingAdds instructs the channel link to disable process new
// incoming add messages.
func (l *chanObserver) DisableIncomingAdds() error {
// If there's no link, then we don't need to disable any adds.
if l.link == nil {
return nil
}
disabled := l.link.DisableAdds(htlcswitch.Incoming)
if disabled {
chanPoint := l.chanView.StateSnapshot().ChannelPoint
peerLog.Debugf("ChannelPoint(%v): link already disabled",
chanPoint)
}
return nil
}
// DisableOutgoingAdds instructs the channel link to disable process new
// outgoing add messages.
func (l *chanObserver) DisableOutgoingAdds() error {
// If there's no link, then we don't need to disable any adds.
if l.link == nil {
return nil
}
_ = l.link.DisableAdds(htlcswitch.Outgoing)
return nil
}
// MarkCoopBroadcasted persistently marks that the channel close transaction
// has been broadcast.
func (l *chanObserver) MarkCoopBroadcasted(tx *wire.MsgTx, local bool) error {
return l.chanView.MarkCoopBroadcasted(tx, lntypes.Local)
}
// MarkShutdownSent persists the given ShutdownInfo. The existence of the
// ShutdownInfo represents the fact that the Shutdown message has been sent by
// us and so should be re-sent on re-establish.
func (l *chanObserver) MarkShutdownSent(deliveryAddr []byte,
isInitiator bool) error {
shutdownInfo := channeldb.NewShutdownInfo(deliveryAddr, isInitiator)
return l.chanView.MarkShutdownSent(shutdownInfo)
}
// FinalBalances is the balances of the channel once it has been flushed. If
// Some, then this indicates that the channel is now in a state where it's
// always flushed, so we can accelerate the state transitions.
func (l *chanObserver) FinalBalances() fn.Option[chancloser.ShutdownBalances] {
chanClean := l.chanView.IsChannelClean()
switch {
// If we have a link, then the balances are final if both the incoming
// and outgoing adds are disabled _and_ the channel is clean.
case l.link != nil && l.link.IsFlushing(htlcswitch.Incoming) &&
l.link.IsFlushing(htlcswitch.Outgoing) && chanClean:
fallthrough
// If we don't have a link, then this is a restart case, so the
// balances are final.
case l.link == nil:
snapshot := l.chanView.StateSnapshot()
return fn.Some(chancloser.ShutdownBalances{
LocalBalance: snapshot.LocalBalance,
RemoteBalance: snapshot.RemoteBalance,
})
// Otherwise, the link is still active and not flushed, so the balances
// aren't yet final.
default:
return fn.None[chancloser.ShutdownBalances]()
}
}
// DisableChannel disables the target channel.
func (l *chanObserver) DisableChannel() error {
op := l.chanView.StateSnapshot().ChannelPoint
return l.linkNetwork.RequestDisable(op, false)
}
// A compile-time assertion to ensure that chanObserver meets the
// chancloser.ChanStateObserver interface.
var _ chancloser.ChanStateObserver = (*chanObserver)(nil)

141
peer/daemon_adapters.go Normal file
View file

@ -0,0 +1,141 @@
package peer
import (
"fmt"
"github.com/btcsuite/btcd/btcec/v2"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/protofsm"
)
// MessageSender is an interface that represents an object capable of sending
// p2p messages to a destination.
type MessageSender interface {
// SendMessages sends the target set of messages to the target peer.
//
// TODO(roasbeef): current impl bound to single peer, need server
// pointer otherwise
SendMessages(btcec.PublicKey, []lnwire.Message) error
}
// flexMessageSender is a message sender-like interface that is aware of
// sync/async semantics, and is bound to a single peer.
type flexMessageSender interface {
// SendMessage sends a variadic number of high-priority messages to the
// remote peer. The first argument denotes if the method should block
// until the messages have been sent to the remote peer or an error is
// returned, otherwise it returns immediately after queuing.
SendMessage(sync bool, msgs ...lnwire.Message) error
}
// peerMsgSender implements the MessageSender interface for a single peer.
// It'll return an error if the target public isn't equal to public key of the
// backing peer.
type peerMsgSender struct {
sender flexMessageSender
peerPub btcec.PublicKey
}
// newPeerMsgSender creates a new instance of a peerMsgSender.
func newPeerMsgSender(peerPub btcec.PublicKey,
msgSender flexMessageSender) *peerMsgSender {
return &peerMsgSender{
sender: msgSender,
peerPub: peerPub,
}
}
// SendMessages sends the target set of messages to the target peer.
//
// TODO(roasbeef): current impl bound to single peer, need server pointer
// otherwise?
func (p *peerMsgSender) SendMessages(pub btcec.PublicKey,
msgs []lnwire.Message) error {
if !p.peerPub.IsEqual(&pub) {
return fmt.Errorf("wrong peer pubkey: got %x, can only send "+
"to %x", pub.SerializeCompressed(),
p.peerPub.SerializeCompressed())
}
return p.sender.SendMessage(true, msgs...)
}
// TxBroadcaster is an interface that represents an object capable of
// broadcasting transactions to the network.
type TxBroadcaster interface {
// PublishTransaction broadcasts a transaction to the network.
PublishTransaction(*wire.MsgTx, string) error
}
// LndAdapterCfg is a struct that holds the configuration for the
// LndDaemonAdapters instance.
type LndAdapterCfg struct {
// MsgSender is capable of sending messages to an arbitrary peer.
MsgSender MessageSender
// TxBroadcaster is capable of broadcasting a transaction to the
// network.
TxBroadcaster TxBroadcaster
// ChainNotifier is capable of receiving notifications for on-chain
// events.
ChainNotifier chainntnfs.ChainNotifier
}
// LndDaemonAdapters is a struct that implements the protofsm.DaemonAdapters
// interface using common lnd abstractions.
type LndDaemonAdapters struct {
cfg LndAdapterCfg
}
// NewLndDaemonAdapters creates a new instance of the lndDaemonAdapters struct.
func NewLndDaemonAdapters(cfg LndAdapterCfg) *LndDaemonAdapters {
return &LndDaemonAdapters{
cfg: cfg,
}
}
// SendMessages sends the target set of messages to the target peer.
func (l *LndDaemonAdapters) SendMessages(pub btcec.PublicKey,
msgs []lnwire.Message) error {
return l.cfg.MsgSender.SendMessages(pub, msgs)
}
// BroadcastTransaction broadcasts a transaction with the target label.
func (l *LndDaemonAdapters) BroadcastTransaction(tx *wire.MsgTx,
label string) error {
return l.cfg.TxBroadcaster.PublishTransaction(tx, label)
}
// RegisterConfirmationsNtfn registers an intent to be notified once txid
// reaches numConfs confirmations.
func (l *LndDaemonAdapters) RegisterConfirmationsNtfn(txid *chainhash.Hash,
pkScript []byte, numConfs, heightHint uint32,
opts ...chainntnfs.NotifierOption,
) (*chainntnfs.ConfirmationEvent, error) {
return l.cfg.ChainNotifier.RegisterConfirmationsNtfn(
txid, pkScript, numConfs, heightHint, opts...,
)
}
// RegisterSpendNtfn registers an intent to be notified once the target
// outpoint is successfully spent within a transaction.
func (l *LndDaemonAdapters) RegisterSpendNtfn(outpoint *wire.OutPoint,
pkScript []byte, heightHint uint32) (*chainntnfs.SpendEvent, error) {
return l.cfg.ChainNotifier.RegisterSpendNtfn(
outpoint, pkScript, heightHint,
)
}
// A compile time check to ensure that lndDaemonAdapters fully implements the
// DaemonAdapters interface.
var _ protofsm.DaemonAdapters = (*LndDaemonAdapters)(nil)

View file

@ -337,7 +337,7 @@ func createTestPeerWithChannel(t *testing.T, updateChan func(a,
chanID := lnwire.NewChanIDFromOutPoint(channelAlice.ChannelPoint())
alicePeer.activeChannels.Store(chanID, channelAlice)
alicePeer.wg.Add(1)
alicePeer.cg.WgAdd(1)
go alicePeer.channelManager()
return &peerTestCtx{

View file

@ -2,7 +2,7 @@ package protofsm
import (
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/msgmux"
)
// MsgMapper is used to map incoming wire messages into a FSM event. This is
@ -11,5 +11,5 @@ import (
type MsgMapper[Event any] interface {
// MapMsg maps a wire message into a FSM event. If the message is not
// mappable, then an None is returned.
MapMsg(msg lnwire.Message) fn.Option[Event]
MapMsg(msg msgmux.PeerMsg) fn.Option[Event]
}

View file

@ -14,6 +14,7 @@ import (
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/lightningnetwork/lnd/lnutils"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/msgmux"
)
const (
@ -77,7 +78,8 @@ type State[Event any, Env Environment] interface {
// otherwise.
IsTerminal() bool
// TODO(roasbeef): also add state serialization?
// String returns a human readable string that represents the state.
String() string
}
// DaemonAdapters is a set of methods that server as adapters to bridge the
@ -249,7 +251,7 @@ func (s *StateMachine[Event, Env]) SendEvent(ctx context.Context, event Event) {
// CanHandle returns true if the target message can be routed to the state
// machine.
func (s *StateMachine[Event, Env]) CanHandle(msg lnwire.Message) bool {
func (s *StateMachine[Event, Env]) CanHandle(msg msgmux.PeerMsg) bool {
cfgMapper := s.cfg.MsgMapper
return fn.MapOptionZ(cfgMapper, func(mapper MsgMapper[Event]) bool {
return mapper.MapMsg(msg).IsSome()
@ -266,7 +268,7 @@ func (s *StateMachine[Event, Env]) Name() string {
// returned indicating that the message was processed. Otherwise, false is
// returned.
func (s *StateMachine[Event, Env]) SendMessage(ctx context.Context,
msg lnwire.Message) bool {
msg msgmux.PeerMsg) bool {
// If we have no message mapper, then return false as we can't process
// this message.
@ -343,7 +345,7 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
// any preconditions as well as post-send events.
case *SendMsgEvent[Event]:
sendAndCleanUp := func() error {
s.log.DebugS(ctx, "Sending message to target",
s.log.DebugS(ctx, "Sending message:",
btclog.Hex6("target", daemonEvent.TargetPeer.SerializeCompressed()),
"messages", lnutils.SpewLogClosure(daemonEvent.Msgs))
@ -375,9 +377,18 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
})
}
// If this doesn't have a SendWhen predicate, then we can just
// send it off right away.
if !daemonEvent.SendWhen.IsSome() {
canSend := func() bool {
return fn.MapOptionZ(
daemonEvent.SendWhen,
func(pred SendPredicate) bool {
return pred()
},
)
}
// If this doesn't have a SendWhen predicate, or if it's already
// true, then we can just send it off right away.
if !daemonEvent.SendWhen.IsSome() || canSend() {
return sendAndCleanUp()
}
@ -395,14 +406,7 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
for {
select {
case <-predicateTicker.C:
canSend := fn.MapOptionZ(
daemonEvent.SendWhen,
func(pred SendPredicate) bool {
return pred()
},
)
if canSend {
if canSend() {
s.log.InfoS(ctx, "Send active predicate")
err := sendAndCleanUp()
@ -435,7 +439,7 @@ func (s *StateMachine[Event, Env]) executeDaemonEvent(ctx context.Context,
daemonEvent.Tx, daemonEvent.Label,
)
if err != nil {
return fmt.Errorf("unable to broadcast txn: %w", err)
log.Errorf("unable to broadcast txn: %v", err)
}
return nil
@ -597,8 +601,8 @@ func (s *StateMachine[Event, Env]) applyEvents(ctx context.Context,
}
s.log.InfoS(ctx, "State transition",
btclog.Fmt("from_state", "%T", currentState),
btclog.Fmt("to_state", "%T", transition.NextState))
btclog.Fmt("from_state", "%v", currentState),
btclog.Fmt("to_state", "%v", transition.NextState))
// With our events processed, we'll now update our
// internal state.

View file

@ -13,6 +13,7 @@ import (
"github.com/lightningnetwork/lnd/chainntnfs"
"github.com/lightningnetwork/lnd/fn/v2"
"github.com/lightningnetwork/lnd/lnwire"
"github.com/lightningnetwork/lnd/msgmux"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
@ -51,6 +52,10 @@ type dummyStateStart struct {
canSend *atomic.Bool
}
func (d *dummyStateStart) String() string {
return "dummyStateStart"
}
var (
hexDecode = func(keyStr string) []byte {
keyBytes, _ := hex.DecodeString(keyStr)
@ -134,6 +139,10 @@ func (d *dummyStateStart) IsTerminal() bool {
type dummyStateFin struct {
}
func (d *dummyStateFin) String() string {
return "dummyStateFin"
}
func (d *dummyStateFin) ProcessEvent(event dummyEvents, env *dummyEnv,
) (*StateTransition[dummyEvents, *dummyEnv], error) {
@ -397,7 +406,7 @@ type dummyMsgMapper struct {
mock.Mock
}
func (d *dummyMsgMapper) MapMsg(wireMsg lnwire.Message) fn.Option[dummyEvents] {
func (d *dummyMsgMapper) MapMsg(wireMsg msgmux.PeerMsg) fn.Option[dummyEvents] {
args := d.Called(wireMsg)
//nolint:forcetypeassert
@ -421,8 +430,12 @@ func TestStateMachineMsgMapper(t *testing.T) {
// The only thing we know how to map is the error message, which'll
// terminate the state machine.
wireError := &lnwire.Error{}
initMsg := &lnwire.Init{}
wireError := msgmux.PeerMsg{
Message: &lnwire.Error{},
}
initMsg := msgmux.PeerMsg{
Message: &lnwire.Init{},
}
dummyMapper.On("MapMsg", wireError).Return(
fn.Some(dummyEvents(&goToFin{})),
)
@ -448,7 +461,7 @@ func TestStateMachineMsgMapper(t *testing.T) {
// First, we'll verify that the CanHandle method works as expected.
require.True(t, stateMachine.CanHandle(wireError))
require.False(t, stateMachine.CanHandle(&lnwire.Init{}))
require.False(t, stateMachine.CanHandle(initMsg))
// Next, we'll attempt to send the wire message into the state machine.
// We should transition to the final state.

View file

@ -2801,14 +2801,33 @@ func (r *rpcServer) CloseChannel(in *lnrpc.CloseChannelRequest,
}
}
var (
chanInSwitch = true
chanHasRbfCloser = r.server.ChanHasRbfCoopCloser(
channel.IdentityPub, *chanPoint,
)
)
// If the link is not known by the switch, we cannot gracefully close
// the channel.
channelID := lnwire.NewChanIDFromOutPoint(*chanPoint)
if _, err := r.server.htlcSwitch.GetLink(channelID); err != nil {
rpcsLog.Debugf("Trying to non-force close offline channel with "+
"chan_point=%v", chanPoint)
return fmt.Errorf("unable to gracefully close channel while peer "+
"is offline (try force closing it instead): %v", err)
chanInSwitch = false
// The channel isn't in the switch, but if there's an
// active chan closer for the channel, and it's of the
// RBF variant, then we can actually bypass the switch.
// Otherwise, we'll return an error.
if !chanHasRbfCloser {
rpcsLog.Debugf("Trying to non-force close "+
"offline channel with chan_point=%v",
chanPoint)
return fmt.Errorf("unable to gracefully close "+
"channel while peer is offline (try "+
"force closing it instead): %v", err)
}
}
// Keep the old behavior prior to 0.18.0 - when the user
@ -2886,10 +2905,31 @@ func (r *rpcServer) CloseChannel(in *lnrpc.CloseChannelRequest,
feeRate)
}
updateChan, errChan = r.server.htlcSwitch.CloseLink(
chanPoint, contractcourt.CloseRegular, feeRate,
maxFee, deliveryScript,
)
if chanHasRbfCloser && !chanInSwitch {
rpcsLog.Infof("Bypassing Switch to do fee bump "+
"for ChannelPoint(%v)", chanPoint)
closeUpdates, err := r.server.AttemptRBFCloseUpdate(
updateStream.Context(), *chanPoint, feeRate,
deliveryScript,
)
if err != nil {
return fmt.Errorf("unable to do RBF close "+
"update: %w", err)
}
updateChan = closeUpdates.UpdateChan
errChan = closeUpdates.ErrChan
} else {
maxFee := chainfee.SatPerKVByte(
in.MaxFeePerVbyte * 1000,
).FeePerKWeight()
updateChan, errChan = r.server.htlcSwitch.CloseLink(
updateStream.Context(), chanPoint,
contractcourt.CloseRegular, feeRate, maxFee,
deliveryScript,
)
}
}
// If the user doesn't want to wait for the txid to come back then we
@ -2911,6 +2951,7 @@ func (r *rpcServer) CloseChannel(in *lnrpc.CloseChannelRequest,
return err
}
}
out:
for {
select {
@ -2956,6 +2997,7 @@ out:
h, _ := chainhash.NewHash(closeUpdate.ClosingTxid)
rpcsLog.Infof("[closechannel] close completed: "+
"txid(%v)", h)
break out
}
@ -3046,12 +3088,23 @@ func createRPCCloseUpdate(
}, nil
case *peer.PendingUpdate:
upd := &lnrpc.PendingUpdate{
Txid: u.Txid,
OutputIndex: u.OutputIndex,
}
// Potentially set the optional fields that are only set for
// the new RBF close flow.
u.IsLocalCloseTx.WhenSome(func(isLocal bool) {
upd.LocalCloseTx = isLocal
})
u.FeePerVbyte.WhenSome(func(feeRate chainfee.SatPerVByte) {
upd.FeePerVbyte = int64(feeRate)
})
return &lnrpc.CloseStatusUpdate{
Update: &lnrpc.CloseStatusUpdate_ClosePending{
ClosePending: &lnrpc.PendingUpdate{
Txid: u.Txid,
OutputIndex: u.OutputIndex,
},
ClosePending: upd,
},
}, nil
}

View file

@ -1411,6 +1411,9 @@
; Set to disable experimental endorsement signaling.
; protocol.no-experimental-endorsement=false
; Set to disable support for RBF based coop close.
; protocol.rbf-coop-close=false
; Set to handle messages of a particular type that falls outside of the
; custom message number range (i.e. 513 is onion messages). Note that you can
; set this option as many times as you want to support more than one custom

101
server.go
View file

@ -612,6 +612,17 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
"aux controllers")
}
// For now, the RBF coop close flag and the taproot channel type cannot
// be used together.
//
// TODO(roasbeef): fix
if cfg.ProtocolOptions.RbfCoopClose &&
cfg.ProtocolOptions.TaprootChans {
return nil, fmt.Errorf("RBF coop close and taproot " +
"channels cannot be used together")
}
//nolint:ll
featureMgr, err := feature.NewManager(feature.Config{
NoTLVOnion: cfg.ProtocolOptions.LegacyOnion(),
@ -629,6 +640,7 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
NoRouteBlinding: cfg.ProtocolOptions.NoRouteBlinding(),
NoExperimentalEndorsement: cfg.ProtocolOptions.NoExperimentalEndorsement(),
NoQuiescence: cfg.ProtocolOptions.NoQuiescence(),
NoRbfCoopClose: !cfg.ProtocolOptions.RbfCoopClose,
})
if err != nil {
return nil, err
@ -1292,7 +1304,9 @@ func newServer(cfg *Config, listenAddrs []net.Addr,
// Instruct the switch to close the channel. Provide no close out
// delivery script or target fee per kw because user input is not
// available when the remote peer closes the channel.
s.htlcSwitch.CloseLink(chanPoint, closureType, 0, 0, nil)
s.htlcSwitch.CloseLink(
context.Background(), chanPoint, closureType, 0, 0, nil,
)
}
// We will use the following channel to reliably hand off contract
@ -5413,3 +5427,88 @@ func (s *server) getStartingBeat() (*chainio.Beat, error) {
return beat, nil
}
// ChanHasRbfCoopCloser returns true if the channel as identifier by the channel
// point has an active RBF chan closer.
func (s *server) ChanHasRbfCoopCloser(peerPub *btcec.PublicKey,
chanPoint wire.OutPoint) bool {
pubBytes := peerPub.SerializeCompressed()
s.mu.RLock()
targetPeer, ok := s.peersByPub[string(pubBytes)]
s.mu.RUnlock()
if !ok {
return false
}
return targetPeer.ChanHasRbfCoopCloser(chanPoint)
}
// attemptCoopRbfFeeBump attempts to look up the active chan closer for a
// channel given the outpoint. If found, we'll attempt to do a fee bump,
// returning channels used for updates. If the channel isn't currently active
// (p2p connection established), then his function will return an error.
func (s *server) attemptCoopRbfFeeBump(ctx context.Context,
chanPoint wire.OutPoint, feeRate chainfee.SatPerKWeight,
deliveryScript lnwire.DeliveryAddress) (*peer.CoopCloseUpdates, error) {
// First, we'll attempt to look up the channel based on it's
// ChannelPoint.
channel, err := s.chanStateDB.FetchChannel(chanPoint)
if err != nil {
return nil, fmt.Errorf("unable to fetch channel: %w", err)
}
// From the channel, we can now get the pubkey of the peer, then use
// that to eventually get the chan closer.
peerPub := channel.IdentityPub.SerializeCompressed()
// Now that we have the peer pub, we can look up the peer itself.
s.mu.RLock()
targetPeer, ok := s.peersByPub[string(peerPub)]
s.mu.RUnlock()
if !ok {
return nil, fmt.Errorf("peer for ChannelPoint(%v) is "+
"not online", chanPoint)
}
closeUpdates, err := targetPeer.TriggerCoopCloseRbfBump(
ctx, chanPoint, feeRate, deliveryScript,
)
if err != nil {
return nil, fmt.Errorf("unable to trigger coop rbf fee bump: "+
"%w", err)
}
return closeUpdates, nil
}
// AttemptRBFCloseUpdate attempts to trigger a new RBF iteration for a co-op
// close update. This route it to be used only if the target channel in question
// is no longer active in the link. This can happen when we restart while we
// already have done a single RBF co-op close iteration.
func (s *server) AttemptRBFCloseUpdate(ctx context.Context,
chanPoint wire.OutPoint, feeRate chainfee.SatPerKWeight,
deliveryScript lnwire.DeliveryAddress) (*peer.CoopCloseUpdates, error) {
// If the channel is present in the switch, then the request should flow
// through the switch instead.
chanID := lnwire.NewChanIDFromOutPoint(chanPoint)
if _, err := s.htlcSwitch.GetLink(chanID); err == nil {
return nil, fmt.Errorf("ChannelPoint(%v) is active in link, "+
"invalid request", chanPoint)
}
// At this point, we know that the channel isn't present in the link, so
// we'll check to see if we have an entry in the active chan closer map.
updates, err := s.attemptCoopRbfFeeBump(
ctx, chanPoint, feeRate, deliveryScript,
)
if err != nil {
return nil, fmt.Errorf("unable to attempt coop rbf fee bump "+
"ChannelPoint(%v)", chanPoint)
}
return updates, nil
}