GlcoinCore/loop
1
0
Fork 0
mirror of https://github.com/lightninglabs/loop.git synced 2026-08-18 13:08:28 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

staticaddr: track unconfirmed deposits

Browse source 
Surface static-address deposits as soon as they appear in the wallet instead
of waiting for the old six-confirmation readiness threshold.

Reconcile the wallet view on startup, on each block, and on the polling ticker
so mempool deposits are created immediately. Backfill the first confirmation
height once those outputs confirm, protect unconfirmed deposits from expiry,
and mark vanished unconfirmed outpoints as Replaced so RBFed-away deposits stop
showing up in RPCs.

Expose the new state through static-address RPCs by deriving availability and
summary totals from stored deposit state, reporting sensible expiry data for
unconfirmed outputs, and hiding Replaced records from normal listings.
This commit is contained in:
Slyghtning 2026-04-27 11:23:06 +02:00
parent 5818b986ae
commit 3df71f9a19
No known key found for this signature in database
GPG key ID: F82D456EA023C9BF
13 changed files with 1265 additions and 117 deletions
Download patch file Download diff file Expand all files Collapse all files

1
loopd/daemon.go
View file

@ -622,6 +622,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error {
depositStore := deposit.NewSqlStore(baseDb)
depoCfg := &deposit.ManagerConfig{
AddressManager: staticAddressManager,
ChainKit: d.lnd.ChainKit,
Store: depositStore,
WalletKit: d.lnd.WalletKit,
ChainNotifier: d.lnd.ChainNotifier,

148
loopd/swapclient_server.go
View file

@ -976,15 +976,24 @@ func (s *swapClientServer) GetLoopInQuote(ctx context.Context,
return nil, fmt.Errorf("expected %d deposits, got %d",
len(req.DepositOutpoints),
len(depositList.FilteredDeposits))
} else {
numDeposits = len(depositList.FilteredDeposits)
}
numDeposits = len(depositList.FilteredDeposits)
// In case we quote for deposits, we send the server both the
// selected value and the number of deposits. This is so the
// server can probe the selected value and calculate the per
// input fee.
for _, deposit := range depositList.FilteredDeposits {
// ListStaticAddressDeposits only filters out deposits that are no
// longer visible to the user, such as Replaced records. For a manual
// quote we additionally require the current state to be Deposited so a
// stale client-side outpoint selection fails early instead of making it
// to swap initiation.
if deposit.State != looprpc.DepositState_DEPOSITED {
return nil, fmt.Errorf("deposit %s is not "+
"currently available", deposit.Outpoint)
}
totalDepositAmount += btcutil.Amount(
deposit.Value,
)
@ -1696,54 +1705,43 @@ func (s *swapClientServer) ListUnspentDeposits(ctx context.Context,
// not spendable because they already have been used but not yet spent
// by the server. We filter out such deposits here.
var (
outpoints []string
isUnspent = make(map[wire.OutPoint]struct{})
outpoints []string
isUnspent = make(map[wire.OutPoint]struct{})
knownUtxos = make(map[wire.OutPoint]struct{})
)
// Keep track of confirmed outpoints that we need to check against our
// database.
confirmedToCheck := make(map[wire.OutPoint]struct{})
for _, utxo := range utxos {
if utxo.Confirmations < deposit.MinConfs {
// Unconfirmed deposits are always available.
isUnspent[utxo.OutPoint] = struct{}{}
} else {
// Confirmed deposits need to be checked.
outpoints = append(outpoints, utxo.OutPoint.String())
confirmedToCheck[utxo.OutPoint] = struct{}{}
}
outpoints = append(outpoints, utxo.OutPoint.String())
knownUtxos[utxo.OutPoint] = struct{}{}
}
// Check the spent status of the deposits by looking at their states.
ignoreUnknownOutpoints := false
ignoreUnknownOutpoints := true
deposits, err := s.depositManager.DepositsForOutpoints(
ctx, outpoints, ignoreUnknownOutpoints,
)
if err != nil {
return nil, err
}
knownDeposits := make(map[wire.OutPoint]struct{}, len(deposits))
for _, d := range deposits {
// A nil deposit means we don't have a record for it. We'll
// handle this case after the loop.
if d == nil {
continue
}
// If the deposit is in the "Deposited" state, it's available.
knownDeposits[d.OutPoint] = struct{}{}
if d.IsInState(deposit.Deposited) {
isUnspent[d.OutPoint] = struct{}{}
}
// We have a record for this deposit, so we no longer need to
// check it.
delete(confirmedToCheck, d.OutPoint)
}
// Any remaining outpoints in confirmedToCheck are ones that lnd knows
// about but we don't. These are new, unspent deposits.
for op := range confirmedToCheck {
isUnspent[op] = struct{}{}
// Any wallet outpoints that are unknown to the deposit store are new
// deposits and therefore still available.
for op := range knownUtxos {
if _, ok := knownDeposits[op]; !ok {
isUnspent[op] = struct{}{}
}
}
// Prepare the list of unspent deposits for the rpc response.
@ -1814,6 +1812,22 @@ func (s *swapClientServer) WithdrawDeposits(ctx context.Context,
}, err
}
// confirmedDeposits filters the given deposits and returns only those that have
// a positive confirmation height, i.e. deposits that have been confirmed
// on-chain.
func confirmedDeposits(deposits []*deposit.Deposit) []*deposit.Deposit {
confirmed := make([]*deposit.Deposit, 0, len(deposits))
for _, d := range deposits {
if d.ConfirmationHeight <= 0 {
continue
}
confirmed = append(confirmed, d)
}
return confirmed
}
// ListStaticAddressDeposits returns a list of all sufficiently confirmed
// deposits behind the static address and displays properties like value,
// state or blocks til expiry.
@ -1838,7 +1852,8 @@ func (s *swapClientServer) ListStaticAddressDeposits(ctx context.Context,
var filteredDeposits []*looprpc.Deposit
if len(outpoints) > 0 {
f := func(d *deposit.Deposit) bool {
return slices.Contains(outpoints, d.OutPoint.String())
return isVisibleDeposit(d) &&
slices.Contains(outpoints, d.OutPoint.String())
}
filteredDeposits = filter(allDeposits, f)
@ -1848,6 +1863,10 @@ func (s *swapClientServer) ListStaticAddressDeposits(ctx context.Context,
}
} else {
f := func(d *deposit.Deposit) bool {
if !isVisibleDeposit(d) {
return false
}
if req.StateFilter == looprpc.DepositState_UNKNOWN_STATE {
// Per default, we return deposits in all
// states.
@ -1982,9 +2001,10 @@ func (s *swapClientServer) ListStaticAddressSwaps(ctx context.Context,
protoDeposits = make([]*looprpc.Deposit, 0, len(ds))
for _, d := range ds {
state := toClientDepositState(d.GetState())
blocksUntilExpiry := d.ConfirmationHeight +
int64(addrParams.Expiry) -
int64(lndInfo.BlockHeight)
blocksUntilExpiry := depositBlocksUntilExpiry(
d.ConfirmationHeight, addrParams.Expiry,
int64(lndInfo.BlockHeight),
)
pd := &looprpc.Deposit{
Id: d.ID[:],
@ -2033,6 +2053,7 @@ func (s *swapClientServer) GetStaticAddressSummary(ctx context.Context,
if err != nil {
return nil, err
}
allDeposits = filterDeposits(allDeposits, isVisibleDeposit)
var (
totalNumDeposits = len(allDeposits)
@ -2045,23 +2066,16 @@ func (s *swapClientServer) GetStaticAddressSummary(ctx context.Context,
htlcTimeoutSwept int64
)
// Value unconfirmed.
utxos, err := s.staticAddressManager.ListUnspent(
ctx, 0, deposit.MinConfs-1,
)
if err != nil {
return nil, err
}
for _, u := range utxos {
valueUnconfirmed += int64(u.Value)
}
// Confirmed total values by category.
// Total values by category.
for _, d := range allDeposits {
value := int64(d.Value)
switch d.GetState() {
case deposit.Deposited:
valueDeposited += value
if d.ConfirmationHeight <= 0 {
valueUnconfirmed += value
} else {
valueDeposited += value
}
case deposit.Expired:
valueExpired += value
@ -2211,13 +2225,27 @@ func (s *swapClientServer) populateBlocksUntilExpiry(ctx context.Context,
return err
}
for i := range len(deposits) {
deposits[i].BlocksUntilExpiry =
deposits[i].ConfirmationHeight +
int64(params.Expiry) - bestBlockHeight
deposits[i].BlocksUntilExpiry = depositBlocksUntilExpiry(
deposits[i].ConfirmationHeight, params.Expiry,
bestBlockHeight,
)
}
return nil
}
// depositBlocksUntilExpiry returns the remaining blocks until a deposit
// expires. Unconfirmed deposits return the full CSV value because the timeout
// has not started yet.
func depositBlocksUntilExpiry(confirmationHeight int64, expiry uint32,
bestBlockHeight int64) int64 {
if confirmationHeight <= 0 {
return int64(expiry)
}
return confirmationHeight + int64(expiry) - bestBlockHeight
}
// StaticOpenChannel initiates an open channel request using static address
// deposits.
func (s *swapClientServer) StaticOpenChannel(ctx context.Context,
@ -2247,6 +2275,30 @@ func (s *swapClientServer) StaticOpenChannel(ctx context.Context,
type filterFunc func(deposits *deposit.Deposit) bool
// filterDeposits returns all deposits accepted by the given predicate.
func filterDeposits(deposits []*deposit.Deposit,
f filterFunc) []*deposit.Deposit {
filtered := make([]*deposit.Deposit, 0, len(deposits))
for _, deposit := range deposits {
if !f(deposit) {
continue
}
filtered = append(filtered, deposit)
}
return filtered
}
// isVisibleDeposit returns true if a deposit should appear in normal listings.
func isVisibleDeposit(d *deposit.Deposit) bool {
// Replaced deposits are kept in the DB as history, but they should disappear
// from normal deposit listings and summary totals because the underlying
// outpoint is no longer present in the wallet and cannot be spent.
return d.GetState() != deposit.Replaced
}
func filter(deposits []*deposit.Deposit, f filterFunc) []*looprpc.Deposit {
var clientDeposits []*looprpc.Deposit
for _, d := range deposits {

21
loopd/swapclient_server_deposit_test.go Normal file
View file

@ -0,0 +1,21 @@
package loopd
import "testing"
// TestDepositBlocksUntilExpiry checks blocks-until-expiry handling for
// confirmed and unconfirmed deposits.
func TestDepositBlocksUntilExpiry(t *testing.T) {
t.Run("unconfirmed", func(t *testing.T) {
if blocks := depositBlocksUntilExpiry(0, 144, 500); blocks != 144 {
t.Fatalf("expected 144 blocks for unconfirmed deposit, got %d",
blocks)
}
})
t.Run("confirmed", func(t *testing.T) {
if blocks := depositBlocksUntilExpiry(450, 144, 500); blocks != 94 {
t.Fatalf("expected 94 blocks until expiry, got %d",
blocks)
}
})
}

225
loopd/swapclient_server_staticaddr_test.go Normal file
View file

@ -0,0 +1,225 @@
package loopd
import (
"context"
"testing"
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/btcsuite/btclog/v2"
"github.com/lightninglabs/loop/looprpc"
"github.com/lightninglabs/loop/staticaddr/address"
"github.com/lightninglabs/loop/staticaddr/deposit"
"github.com/lightninglabs/loop/staticaddr/script"
mock_lnd "github.com/lightninglabs/loop/test"
"github.com/stretchr/testify/require"
)
type staticAddrDepositStore struct {
allDeposits []*deposit.Deposit
byOutpoint map[string]*deposit.Deposit
}
// CreateDeposit implements deposit.Store for static address server tests.
func (s *staticAddrDepositStore) CreateDeposit(context.Context,
*deposit.Deposit) error {
return nil
}
// UpdateDeposit implements deposit.Store for static address server tests.
func (s *staticAddrDepositStore) UpdateDeposit(context.Context,
*deposit.Deposit) error {
return nil
}
// GetDeposit implements deposit.Store for static address server tests.
func (s *staticAddrDepositStore) GetDeposit(context.Context,
deposit.ID) (*deposit.Deposit, error) {
return nil, nil
}
// DepositForOutpoint returns the deposit for the requested outpoint.
func (s *staticAddrDepositStore) DepositForOutpoint(_ context.Context,
outpoint string) (*deposit.Deposit, error) {
if deposit, ok := s.byOutpoint[outpoint]; ok {
return deposit, nil
}
return nil, deposit.ErrDepositNotFound
}
// AllDeposits returns all deposits seeded into the test store.
func (s *staticAddrDepositStore) AllDeposits(context.Context) (
[]*deposit.Deposit, error) {
return s.allDeposits, nil
}
// newTestDepositManager creates a deposit manager backed by seeded deposits.
func newTestDepositManager(
deposits ...*deposit.Deposit) *deposit.Manager {
byOutpoint := make(map[string]*deposit.Deposit, len(deposits))
for _, deposit := range deposits {
byOutpoint[deposit.OutPoint.String()] = deposit
}
return deposit.NewManager(&deposit.ManagerConfig{
Store: &staticAddrDepositStore{
allDeposits: deposits,
byOutpoint: byOutpoint,
},
})
}
// newTestStaticAddressContext creates static address test dependencies.
func newTestStaticAddressContext(t *testing.T) (*address.Manager,
*mock_lnd.LndMockServices) {
t.Helper()
mock := mock_lnd.NewMockLnd()
_, client := mock_lnd.CreateKey(1)
_, server := mock_lnd.CreateKey(2)
addrStore := &mockAddressStore{
params: []*script.Parameters{{
ClientPubkey: client,
ServerPubkey: server,
Expiry: 10,
PkScript: []byte("pkscript"),
}},
}
addrMgr, err := address.NewManager(&address.ManagerConfig{
Store: addrStore,
WalletKit: mock.WalletKit,
ChainParams: mock.ChainParams,
}, 1)
require.NoError(t, err)
return addrMgr, mock
}
// TestListStaticAddressDepositsHidesReplaced verifies replaced deposits are
// hidden from normal deposit listings.
func TestListStaticAddressDepositsHidesReplaced(t *testing.T) {
t.Parallel()
replaced := &deposit.Deposit{
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{1},
Index: 1,
},
}
replaced.SetState(deposit.Replaced)
available := &deposit.Deposit{
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{2},
Index: 2,
},
}
available.SetState(deposit.Deposited)
addrMgr, lnd := newTestStaticAddressContext(t)
server := &swapClientServer{
depositManager: newTestDepositManager(replaced, available),
staticAddressManager: addrMgr,
lnd: &lnd.LndServices,
}
resp, err := server.ListStaticAddressDeposits(
context.Background(), &looprpc.ListStaticAddressDepositsRequest{},
)
require.NoError(t, err)
require.Len(t, resp.FilteredDeposits, 1)
require.Equal(
t, available.OutPoint.String(),
resp.FilteredDeposits[0].Outpoint,
)
}
// TestGetStaticAddressSummaryIgnoresReplaced verifies replaced deposits are
// excluded from static address summary totals.
func TestGetStaticAddressSummaryIgnoresReplaced(t *testing.T) {
t.Parallel()
replaced := &deposit.Deposit{
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{3},
Index: 3,
},
Value: btcutil.Amount(1_000),
}
replaced.SetState(deposit.Replaced)
unconfirmed := &deposit.Deposit{
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{4},
Index: 4,
},
Value: btcutil.Amount(2_000),
ConfirmationHeight: 0,
}
unconfirmed.SetState(deposit.Deposited)
confirmed := &deposit.Deposit{
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{5},
Index: 5,
},
Value: btcutil.Amount(3_000),
ConfirmationHeight: 123,
}
confirmed.SetState(deposit.Deposited)
addrMgr, _ := newTestStaticAddressContext(t)
server := &swapClientServer{
depositManager: newTestDepositManager(
replaced, unconfirmed, confirmed,
),
staticAddressManager: addrMgr,
}
resp, err := server.GetStaticAddressSummary(
context.Background(), &looprpc.StaticAddressSummaryRequest{},
)
require.NoError(t, err)
require.EqualValues(t, 2, resp.TotalNumDeposits)
require.EqualValues(t, 2_000, resp.ValueUnconfirmedSatoshis)
require.EqualValues(t, 3_000, resp.ValueDepositedSatoshis)
}
// TestGetLoopInQuoteRejectsUnavailableSelectedDeposit verifies manual quote
// requests fail for selected deposits that are no longer available.
func TestGetLoopInQuoteRejectsUnavailableSelectedDeposit(t *testing.T) {
t.Parallel()
setLogger(btclog.Disabled)
locked := &deposit.Deposit{
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{6},
Index: 6,
},
Value: btcutil.Amount(5_000),
}
locked.SetState(deposit.LoopingIn)
addrMgr, lnd := newTestStaticAddressContext(t)
server := &swapClientServer{
depositManager: newTestDepositManager(locked),
staticAddressManager: addrMgr,
lnd: &lnd.LndServices,
}
_, err := server.GetLoopInQuote(context.Background(), &looprpc.QuoteRequest{
DepositOutpoints: []string{locked.OutPoint.String()},
})
require.ErrorContains(t, err, "is not currently available")
}

55
loopd/swapclient_server_test.go
View file

@ -1071,7 +1071,7 @@ func (s *mockDepositStore) DepositForOutpoint(_ context.Context,
if d, ok := s.byOutpoint[outpoint]; ok {
return d, nil
}
return nil, nil
return nil, deposit.ErrDepositNotFound
}
func (s *mockDepositStore) AllDeposits(_ context.Context) ([]*deposit.Deposit,
error) {
@ -1120,11 +1120,11 @@ func TestListUnspentDeposits(t *testing.T) {
}
}
minConfs := int64(deposit.MinConfs)
utxoBelow := makeUtxo(0, minConfs-1) // always included
utxoAt := makeUtxo(1, minConfs) // included only if Deposited
utxoAbove1 := makeUtxo(2, minConfs+1)
utxoAbove2 := makeUtxo(3, minConfs+2)
utxoUnknown := makeUtxo(0, 0)
utxoDeposited := makeUtxo(1, 1)
utxoWithdrawn := makeUtxo(2, 2)
utxoLoopingIn := makeUtxo(3, 5)
utxoConfirmedUnknown := makeUtxo(4, 3)
// Helper to build the deposit manager with specific states.
buildDepositMgr := func(
@ -1142,17 +1142,19 @@ func TestListUnspentDeposits(t *testing.T) {
return deposit.NewManager(&deposit.ManagerConfig{Store: store})
}
// Include below-min-conf and >=min with Deposited; exclude others.
t.Run("below min conf always, Deposited included, others excluded",
// Unknown deposits are available, Deposited is available and known
// non-Deposited states are excluded.
t.Run("unknown and Deposited included, locked states excluded",
func(t *testing.T) {
mock.SetListUnspent([]*lnwallet.Utxo{
utxoBelow, utxoAt, utxoAbove1, utxoAbove2,
utxoUnknown, utxoDeposited, utxoWithdrawn,
utxoLoopingIn,
})
depMgr := buildDepositMgr(map[wire.OutPoint]fsm.StateType{
utxoAt.OutPoint: deposit.Deposited,
utxoAbove1.OutPoint: deposit.Withdrawn,
utxoAbove2.OutPoint: deposit.LoopingIn,
utxoDeposited.OutPoint: deposit.Deposited,
utxoWithdrawn.OutPoint: deposit.Withdrawn,
utxoLoopingIn.OutPoint: deposit.LoopingIn,
})
server := &swapClientServer{
@ -1165,7 +1167,7 @@ func TestListUnspentDeposits(t *testing.T) {
)
require.NoError(t, err)
// Expect utxoBelow and utxoAt only.
// Expect the unknown utxo and the Deposited utxo only.
require.Len(t, resp.Utxos, 2)
got := map[string]struct{}{}
for _, u := range resp.Utxos {
@ -1174,25 +1176,25 @@ func TestListUnspentDeposits(t *testing.T) {
// same across utxos.
require.NotEmpty(t, u.StaticAddress)
}
_, ok1 := got[utxoBelow.OutPoint.String()]
_, ok2 := got[utxoAt.OutPoint.String()]
_, ok1 := got[utxoUnknown.OutPoint.String()]
_, ok2 := got[utxoDeposited.OutPoint.String()]
require.True(t, ok1)
require.True(t, ok2)
})
// Swap states, now include utxoBelow and utxoAbove1.
t.Run("Deposited on >=min included; non-Deposited excluded",
// Confirmation depth no longer changes availability; state does.
t.Run("availability ignores conf depth once deposit state is known",
func(t *testing.T) {
mock.SetListUnspent(
[]*lnwallet.Utxo{
utxoBelow, utxoAt, utxoAbove1,
utxoAbove2,
utxoUnknown, utxoDeposited,
utxoWithdrawn, utxoLoopingIn,
})
depMgr := buildDepositMgr(map[wire.OutPoint]fsm.StateType{
utxoAt.OutPoint: deposit.Withdrawn,
utxoAbove1.OutPoint: deposit.Deposited,
utxoAbove2.OutPoint: deposit.Withdrawn,
utxoDeposited.OutPoint: deposit.Deposited,
utxoWithdrawn.OutPoint: deposit.Withdrawn,
utxoLoopingIn.OutPoint: deposit.LoopingIn,
})
server := &swapClientServer{
@ -1210,8 +1212,8 @@ func TestListUnspentDeposits(t *testing.T) {
for _, u := range resp.Utxos {
got[u.Outpoint] = struct{}{}
}
_, ok1 := got[utxoBelow.OutPoint.String()]
_, ok2 := got[utxoAbove1.OutPoint.String()]
_, ok1 := got[utxoUnknown.OutPoint.String()]
_, ok2 := got[utxoDeposited.OutPoint.String()]
require.True(t, ok1)
require.True(t, ok2)
})
@ -1220,7 +1222,7 @@ func TestListUnspentDeposits(t *testing.T) {
t.Run("confirmed utxo not in store is included", func(t *testing.T) {
// Only return a confirmed UTXO from lnd and make sure the
// deposit manager/store doesn't know about it.
mock.SetListUnspent([]*lnwallet.Utxo{utxoAbove2})
mock.SetListUnspent([]*lnwallet.Utxo{utxoConfirmedUnknown})
// Empty store (no states for any outpoint).
depMgr := buildDepositMgr(map[wire.OutPoint]fsm.StateType{})
@ -1239,7 +1241,8 @@ func TestListUnspentDeposits(t *testing.T) {
// doesn't exist in the store yet.
require.Len(t, resp.Utxos, 1)
require.Equal(
t, utxoAbove2.OutPoint.String(), resp.Utxos[0].Outpoint,
t, utxoConfirmedUnknown.OutPoint.String(),
resp.Utxos[0].Outpoint,
)
require.NotEmpty(t, resp.Utxos[0].StaticAddress)
})

4
loopdb/sqlc/migrations/000010_static_address_deposits.up.sql
View file

@ -16,7 +16,7 @@ CREATE TABLE IF NOT EXISTS deposits (
amount BIGINT NOT NULL,
-- confirmation_height is the absolute height at which the deposit was
-- confirmed.
-- confirmed. A value of 0 means the deposit is still unconfirmed.
confirmation_height BIGINT NOT NULL,
-- timeout_sweep_pk_script is the public key script that will be used to
@ -45,4 +45,4 @@ CREATE TABLE IF NOT EXISTS deposit_updates (
-- update_timestamp is the timestamp of the update.
update_timestamp TIMESTAMP NOT NULL
);
);

16
staticaddr/deposit/deposit.go
View file

@ -29,6 +29,10 @@ func (r *ID) FromByteSlice(b []byte) error {
// Deposit bundles an utxo at a static address together with manager-relevant
// data.
//
// Lock order: if both Manager.mu and a Deposit lock are needed, acquire
// Manager.mu before Deposit.Lock. Never acquire Manager.mu while holding a
// Deposit lock.
type Deposit struct {
sync.Mutex
@ -45,7 +49,8 @@ type Deposit struct {
Value btcutil.Amount
// ConfirmationHeight is the absolute height at which the deposit was
// first confirmed.
// first confirmed. A value of zero means the deposit is still
// unconfirmed.
ConfirmationHeight int64
// TimeOutSweepPkScript is the pk script that is used to sweep the
@ -69,15 +74,22 @@ func (d *Deposit) IsInFinalState() bool {
d.Lock()
defer d.Unlock()
// Replaced is inactive from the deposit FSM's point of view. The manager may
// still revive the same record if lnd reports the exact outpoint again after
// a transient wallet-view miss.
return d.state == Expired || d.state == Withdrawn ||
d.state == LoopedIn || d.state == HtlcTimeoutSwept ||
d.state == ChannelPublished
d.state == ChannelPublished || d.state == Replaced
}
func (d *Deposit) IsExpired(currentHeight, expiry uint32) bool {
d.Lock()
defer d.Unlock()
if d.ConfirmationHeight <= 0 {
return false
}
return currentHeight >= uint32(d.ConfirmationHeight)+expiry
}

15
staticaddr/deposit/deposit_test.go Normal file
View file

@ -0,0 +1,15 @@
package deposit
import "testing"
// TestIsExpiredUnconfirmed checks that unconfirmed deposits don't start their
// expiry timer.
func TestIsExpiredUnconfirmed(t *testing.T) {
deposit := &Deposit{
ConfirmationHeight: 0,
}
if deposit.IsExpired(500, 100) {
t.Fatal("unconfirmed deposit should not be expired")
}
}

45
staticaddr/deposit/fsm.go
View file

@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"sync"
"github.com/btcsuite/btcd/txscript"
"github.com/btcsuite/btcd/wire"
@ -41,10 +42,26 @@ var (
// States.
var (
// Deposited signals that funds at a static address have reached the
// confirmation height.
// Deposited signals that funds at a static address have been detected
// and are available to the client.
Deposited = fsm.StateType("Deposited")
// Replaced signals that a deposit disappeared from the wallet view and
// can no longer be spent.
//
// The concrete case we need to handle is mempool replacement: a user can
// receive to the static address, we persist that unconfirmed outpoint, and
// then the funding transaction can be replaced or otherwise evicted before
// confirmation. Once that happens lnd stops returning the old outpoint from
// ListUnspent, but our DB would otherwise keep presenting it as selectable.
// Replaced lets us retain the historic record while making it clear that the
// original outpoint is no longer a live deposit.
//
// This state is managed directly by the deposit manager rather than via
// DepositStatesV0 because it reflects wallet visibility changes such as
// mempool replacement or deep reorgs, not an FSM-driven spend path.
Replaced = fsm.StateType("Replaced")
// Withdrawing signals that the withdrawal transaction has been
// broadcast, awaiting sufficient confirmations.
Withdrawing = fsm.StateType("Withdrawing")
@ -92,8 +109,8 @@ var (
// Events.
var (
// OnStart is sent to the fsm once the deposit outpoint has been
// sufficiently confirmed. It transitions the fsm into the Deposited
// state from where we can trigger a withdrawal, a loopin or an expiry.
// detected. It transitions the fsm into the Deposited state from where
// we can trigger a withdrawal, a loopin or an expiry.
OnStart = fsm.EventType("OnStart")
// OnWithdrawInitiated is sent to the fsm when a withdrawal has been
@ -160,12 +177,17 @@ type FSM struct {
blockNtfnChan chan uint32
// stopChan requests shutdown of the block notification loop.
stopChan chan struct{}
// quitChan stops after the FSM stops consuming blockNtfnChan.
quitChan chan struct{}
// finalizedDepositChan is used to signal that the deposit has been
// finalized and the FSM can be removed from the manager's memory.
finalizedDepositChan chan wire.OutPoint
stopOnce sync.Once
}
// NewFSM creates a new state machine that can action on all static address
@ -191,6 +213,7 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
params: params,
address: address,
blockNtfnChan: make(chan uint32),
stopChan: make(chan struct{}),
quitChan: make(chan struct{}),
finalizedDepositChan: finalizedDepositChan,
}
@ -226,6 +249,9 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
ctx, currentHeight,
)
case <-fsm.stopChan:
return
case <-ctx.Done():
return
}
@ -235,6 +261,17 @@ func NewFSM(ctx context.Context, deposit *Deposit, cfg *ManagerConfig,
return depoFsm, nil
}
// Stop requests shutdown of the FSM's block notification loop.
func (f *FSM) Stop() {
if f == nil || f.stopChan == nil {
return
}
f.stopOnce.Do(func() {
close(f.stopChan)
})
}
// handleBlockNotification inspects the current block height and sends the
// OnExpiry event to publish the expiry sweep transaction if the deposit timed
// out, or it republishes the expiry sweep transaction if it was not yet swept.

419
staticaddr/deposit/manager.go
View file

@ -6,6 +6,7 @@ import (
"fmt"
"sort"
"sync"
"sync/atomic"
"time"
"github.com/btcsuite/btcd/txscript"
@ -33,6 +34,15 @@ const (
// PollInterval is the interval in which we poll for new deposits to our
// static address.
PollInterval = 10 * time.Second
// vanishedDepositThreshold is the number of consecutive wallet
// observations in which a Deposited outpoint must be missing before we
// mark it replaced.
//
// A single miss can happen during a transient wallet-view gap while lnd is
// processing a replacement or reorg. Requiring two misses keeps that narrow
// race recoverable without leaving vanished deposits selectable forever.
vanishedDepositThreshold = 2
)
// ManagerConfig holds the configuration for the address manager.
@ -41,6 +51,10 @@ type ManagerConfig struct {
// address parameters.
AddressManager AddressManager
// ChainKit is used to query the best known chain tip when deriving
// confirmation heights from wallet UTXOs.
ChainKit lndclient.ChainKitClient
// Store is the database store that is used to store static address
// related records.
Store Store
@ -58,15 +72,27 @@ type ManagerConfig struct {
}
// Manager manages the address state machines.
//
// Lock order: if both Manager.mu and a Deposit lock are needed, acquire
// Manager.mu before Deposit.Lock. Never acquire Manager.mu while holding a
// Deposit lock.
type Manager struct {
cfg *ManagerConfig
// mu guards access to the activeDeposits map.
mu sync.Mutex
// reconcileMu serializes deposit reconciliation so new deposits are
// discovered and retained exactly once per outpoint.
reconcileMu sync.Mutex
// activeDeposits contains all the active static address outputs.
activeDeposits map[wire.OutPoint]*FSM
// missingDeposits counts consecutive wallet observations in which a
// Deposited outpoint was missing from the wallet view.
missingDeposits map[wire.OutPoint]uint8
// deposits contain all the deposits that have ever been made to the
// static address. This field is used to store and recover deposits. It
// also serves as a basis for reconciliation of newly detected deposits
@ -77,6 +103,9 @@ type Manager struct {
// been finalized. The manager will adjust its internal state and flush
// finalized deposits from its memory.
finalizedDepositChan chan wire.OutPoint
// currentHeight stores the currently best known block height.
currentHeight atomic.Uint32
}
// NewManager creates a new deposit manager.
@ -84,6 +113,7 @@ func NewManager(cfg *ManagerConfig) *Manager {
return &Manager{
cfg: cfg,
activeDeposits: make(map[wire.OutPoint]*FSM),
missingDeposits: make(map[wire.OutPoint]uint8),
deposits: make(map[wire.OutPoint]*Deposit),
finalizedDepositChan: make(chan wire.OutPoint),
}
@ -98,6 +128,17 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
return err
}
select {
case height := <-newBlockChan:
m.currentHeight.Store(uint32(height))
case err = <-newBlockErrChan:
return err
case <-ctx.Done():
return ctx.Err()
}
// Recover previous deposits and static address parameters from the DB.
err = m.recoverDeposits(ctx)
if err != nil {
@ -123,6 +164,13 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
for {
select {
case height := <-newBlockChan:
m.currentHeight.Store(uint32(height))
err := m.reconcileDeposits(ctx)
if err != nil {
log.Errorf("unable to reconcile deposits: %v", err)
}
// Inform all active deposits about a new block arrival.
m.mu.Lock()
activeDeposits := make([]*FSM, 0, len(m.activeDeposits))
@ -146,9 +194,7 @@ func (m *Manager) Run(ctx context.Context, initChan chan struct{}) error {
case outpoint := <-m.finalizedDepositChan:
// If deposits notify us about their finalization, flush
// the finalized deposit from memory.
m.mu.Lock()
delete(m.activeDeposits, outpoint)
m.mu.Unlock()
m.removeActiveDeposit(outpoint)
case err = <-newBlockErrChan:
return err
@ -207,8 +253,10 @@ func (m *Manager) recoverDeposits(ctx context.Context) error {
return nil
}
// pollDeposits polls new deposits to our static address and notifies the
// manager's event loop about them.
// pollDeposits periodically polls for new deposits to our static address. This
// complements the block-driven reconciliation in the main event loop: while new
// blocks trigger reconcileDeposits to promptly detect confirmations, the ticker
// here catches deposits that appear in the mempool between blocks.
func (m *Manager) pollDeposits(ctx context.Context) {
log.Debugf("Waiting for new static address deposits...")
@ -236,13 +284,36 @@ func (m *Manager) pollDeposits(ctx context.Context) {
// far. It picks the newly identified deposits and starts a state machine per
// deposit to track its progress.
func (m *Manager) reconcileDeposits(ctx context.Context) error {
m.reconcileMu.Lock()
defer m.reconcileMu.Unlock()
log.Tracef("Reconciling new deposits...")
utxos, err := m.cfg.AddressManager.ListUnspent(
ctx, MinConfs, MaxConfs,
)
utxos, bestHeight, err := m.listUnspentWithBestHeight(ctx)
if err != nil {
return fmt.Errorf("unable to list new deposits: %w", err)
return err
}
err = m.updateDepositConfirmations(ctx, utxos, bestHeight)
if err != nil {
return fmt.Errorf("unable to update deposit "+
"confirmations: %w", err)
}
// If the same outpoint reappeared after a transient wallet-view miss,
// reactivate the existing record before we consider it new or vanished.
err = m.reviveReappearedDeposits(ctx, utxos, bestHeight)
if err != nil {
return fmt.Errorf("unable to revive reappeared deposits: %w",
err)
}
// After handling reappearances, only still-missing outpoints contribute
// towards replacement detection.
err = m.invalidateVanishedDeposits(ctx, utxos)
if err != nil {
return fmt.Errorf("unable to invalidate vanished "+
"deposits: %w", err)
}
newDeposits := m.filterNewDeposits(utxos)
@ -252,7 +323,7 @@ func (m *Manager) reconcileDeposits(ctx context.Context) error {
}
for _, utxo := range newDeposits {
deposit, err := m.createNewDeposit(ctx, utxo)
deposit, err := m.createNewDeposit(ctx, utxo, bestHeight)
if err != nil {
return fmt.Errorf("unable to retain new deposit: %w",
err)
@ -269,12 +340,70 @@ func (m *Manager) reconcileDeposits(ctx context.Context) error {
return nil
}
// listUnspentWithBestHeight returns the wallet's current static-address UTXOs
// together with a stable chain tip height for any confirmed outputs.
func (m *Manager) listUnspentWithBestHeight(ctx context.Context) (
[]*lnwallet.Utxo, int32, error) {
utxos, err := m.cfg.AddressManager.ListUnspent(ctx, 0, MaxConfs)
if err != nil {
return nil, 0, fmt.Errorf("unable to list new deposits: %w", err)
}
needsBestHeight := false
for _, utxo := range utxos {
if utxo.Confirmations > 0 {
needsBestHeight = true
break
}
}
if !needsBestHeight {
return utxos, 0, nil
}
if m.cfg.ChainKit == nil {
return nil, 0, errors.New("chain kit client required for " +
"confirmed deposits")
}
const maxAttempts = 3
for range maxAttempts {
_, beforeHeight, err := m.cfg.ChainKit.GetBestBlock(ctx)
if err != nil {
return nil, 0, fmt.Errorf("unable to get best block "+
"before listing deposits: %w", err)
}
utxos, err = m.cfg.AddressManager.ListUnspent(ctx, 0, MaxConfs)
if err != nil {
return nil, 0, fmt.Errorf("unable to list new deposits: %w",
err)
}
_, afterHeight, err := m.cfg.ChainKit.GetBestBlock(ctx)
if err != nil {
return nil, 0, fmt.Errorf("unable to get best block "+
"after listing deposits: %w", err)
}
if beforeHeight == afterHeight {
m.currentHeight.Store(uint32(afterHeight))
return utxos, afterHeight, nil
}
}
return nil, 0, errors.New("unable to get stable best block while " +
"listing deposits")
}
// createNewDeposit transforms the wallet utxo into a deposit struct and stores
// it in our database and manager memory.
func (m *Manager) createNewDeposit(ctx context.Context,
utxo *lnwallet.Utxo) (*Deposit, error) {
utxo *lnwallet.Utxo, bestHeight int32) (*Deposit, error) {
blockHeight, err := m.getBlockHeight(ctx, utxo)
confirmationHeight, err := confirmationHeightForUtxo(
bestHeight, utxo,
)
if err != nil {
return nil, err
}
@ -302,7 +431,7 @@ func (m *Manager) createNewDeposit(ctx context.Context,
state: Deposited,
OutPoint: utxo.OutPoint,
Value: utxo.Value,
ConfirmationHeight: int64(blockHeight),
ConfirmationHeight: confirmationHeight,
TimeOutSweepPkScript: timeoutSweepPkScript,
}
@ -318,37 +447,243 @@ func (m *Manager) createNewDeposit(ctx context.Context,
return deposit, nil
}
// getBlockHeight retrieves the block height of a given utxo.
func (m *Manager) getBlockHeight(ctx context.Context,
utxo *lnwallet.Utxo) (uint32, error) {
// confirmationHeightForUtxo derives the first confirmation height of a wallet
// UTXO from a stable best-known chain tip. Unconfirmed UTXOs return 0.
func confirmationHeightForUtxo(bestHeight int32,
utxo *lnwallet.Utxo) (int64, error) {
addressParams, err := m.cfg.AddressManager.GetStaticAddressParameters(
ctx,
)
if err != nil {
return 0, fmt.Errorf("couldn't get confirmation height for "+
"deposit, %w", err)
if utxo.Confirmations <= 0 {
return 0, nil
}
notifChan, errChan, err :=
m.cfg.ChainNotifier.RegisterConfirmationsNtfn(
ctx, &utxo.OutPoint.Hash, addressParams.PkScript,
MinConfs, addressParams.InitiationHeight,
if bestHeight <= 0 {
return 0, fmt.Errorf("invalid best height %d", bestHeight)
}
firstConfirmationHeight := int64(bestHeight) - utxo.Confirmations + 1
if firstConfirmationHeight <= 0 {
return 0, fmt.Errorf("invalid confirmation height %d for %v "+
"with best height %d and %d confirmations",
firstConfirmationHeight, utxo.OutPoint, bestHeight,
utxo.Confirmations)
}
return firstConfirmationHeight, nil
}
// updateDepositConfirmations backfills first confirmation heights for deposits
// that were previously detected unconfirmed.
func (m *Manager) updateDepositConfirmations(ctx context.Context,
utxos []*lnwallet.Utxo, bestHeight int32) error {
for _, utxo := range utxos {
if utxo.Confirmations <= 0 {
continue
}
m.mu.Lock()
deposit, ok := m.deposits[utxo.OutPoint]
m.mu.Unlock()
if !ok {
continue
}
deposit.Lock()
if deposit.ConfirmationHeight > 0 {
deposit.Unlock()
continue
}
deposit.Unlock()
confirmationHeight, err := confirmationHeightForUtxo(
bestHeight, utxo,
)
if err != nil {
return 0, err
if err != nil {
return err
}
deposit.Lock()
if deposit.ConfirmationHeight > 0 {
deposit.Unlock()
continue
}
previousConfirmationHeight := deposit.ConfirmationHeight
deposit.ConfirmationHeight = confirmationHeight
err = m.cfg.Store.UpdateDeposit(ctx, deposit)
if err != nil {
deposit.ConfirmationHeight = previousConfirmationHeight
deposit.Unlock()
return err
}
deposit.Unlock()
}
select {
case tx := <-notifChan:
return tx.BlockHeight, nil
return nil
}
case err := <-errChan:
return 0, err
// reviveReappearedDeposits reactivates deposits that were previously marked as
// replaced if the exact same outpoint reappears in the wallet view.
//
// This is the inverse of invalidateVanishedDeposits: it lets us
// recover from a transient ListUnspent gap without inventing a second record
// for the same outpoint.
func (m *Manager) reviveReappearedDeposits(ctx context.Context,
utxos []*lnwallet.Utxo, bestHeight int32) error {
case <-ctx.Done():
return 0, ctx.Err()
type reviveCandidate struct {
deposit *Deposit
utxo *lnwallet.Utxo
}
var candidates []reviveCandidate
m.mu.Lock()
for _, utxo := range utxos {
delete(m.missingDeposits, utxo.OutPoint)
deposit, ok := m.deposits[utxo.OutPoint]
if !ok {
continue
}
if _, active := m.activeDeposits[utxo.OutPoint]; active {
continue
}
deposit.Lock()
isReplaced := deposit.IsInStateNoLock(Replaced)
deposit.Unlock()
if !isReplaced {
continue
}
candidates = append(candidates, reviveCandidate{
deposit: deposit,
utxo: utxo,
})
}
m.mu.Unlock()
for _, candidate := range candidates {
confirmationHeight, err := confirmationHeightForUtxo(
bestHeight, candidate.utxo,
)
if err != nil {
return err
}
deposit := candidate.deposit
deposit.Lock()
if !deposit.IsInStateNoLock(Replaced) {
deposit.Unlock()
continue
}
previousState := deposit.state
previousConfirmationHeight := deposit.ConfirmationHeight
deposit.ConfirmationHeight = confirmationHeight
deposit.SetStateNoLock(Deposited)
err = m.cfg.Store.UpdateDeposit(ctx, deposit)
if err != nil {
deposit.ConfirmationHeight = previousConfirmationHeight
deposit.SetStateNoLock(previousState)
deposit.Unlock()
return err
}
deposit.Unlock()
log.Infof("Reactivated deposit %v after it reappeared in "+
"wallet view", deposit.OutPoint)
err = m.startDepositFsm(ctx, deposit)
if err != nil {
return err
}
}
return nil
}
// invalidateVanishedDeposits marks Deposited outputs as replaced once lnd no
// longer reports the outpoint in multiple consecutive wallet observations.
//
// This closes the gap between wallet state and our DB state when a persisted
// deposit later disappears from the wallet view, for example because an
// unconfirmed funding transaction was replaced or because a previously
// confirmed transaction was evicted by a deep reorg. We only invalidate
// deposits that are still in the plain Deposited state.
//
// That keeps the scope narrow: in-flight states like LoopingIn already have
// their own recovery/error handling.
func (m *Manager) invalidateVanishedDeposits(ctx context.Context,
utxos []*lnwallet.Utxo) error {
currentUtxos := make(map[wire.OutPoint]struct{}, len(utxos))
for _, utxo := range utxos {
currentUtxos[utxo.OutPoint] = struct{}{}
}
m.mu.Lock()
candidates := make([]*Deposit, 0, len(m.deposits))
for outpoint, deposit := range m.deposits {
if _, ok := currentUtxos[outpoint]; ok {
delete(m.missingDeposits, outpoint)
continue
}
deposit.Lock()
isVanishedDeposit := deposit.IsInStateNoLock(Deposited)
deposit.Unlock()
if !isVanishedDeposit {
delete(m.missingDeposits, outpoint)
continue
}
m.missingDeposits[outpoint]++
if m.missingDeposits[outpoint] < vanishedDepositThreshold {
log.Debugf("Waiting for another wallet observation before "+
"marking deposit %v replaced", outpoint)
continue
}
delete(m.missingDeposits, outpoint)
candidates = append(candidates, deposit)
}
m.mu.Unlock()
for _, deposit := range candidates {
deposit.Lock()
if !deposit.IsInStateNoLock(Deposited) {
deposit.Unlock()
continue
}
// Persist the replacement marker before removing the deposit from the
// active set so restarted clients and RPC consumers see the same outcome.
previousState := deposit.state
deposit.SetStateNoLock(Replaced)
err := m.cfg.Store.UpdateDeposit(ctx, deposit)
if err != nil {
deposit.SetStateNoLock(previousState)
deposit.Unlock()
return err
}
deposit.Unlock()
m.removeActiveDeposit(deposit.OutPoint)
log.Infof("Marked vanished deposit %v as replaced",
deposit.OutPoint)
}
return nil
}
// filterNewDeposits filters the given utxos for new deposits that we haven't
@ -537,6 +872,20 @@ func unlockDeposits(deposits []*Deposit) {
}
}
// removeActiveDeposit removes and stops the FSM for an active outpoint.
func (m *Manager) removeActiveDeposit(outpoint wire.OutPoint) {
m.mu.Lock()
fsm, ok := m.activeDeposits[outpoint]
if ok {
delete(m.activeDeposits, outpoint)
}
m.mu.Unlock()
if ok {
fsm.Stop()
}
}
// GetAllDeposits returns all active deposits.
func (m *Manager) GetAllDeposits(ctx context.Context) ([]*Deposit, error) {
return m.cfg.Store.AllDeposits(ctx)

40
staticaddr/deposit/manager_height_test.go Normal file
View file

@ -0,0 +1,40 @@
package deposit
import (
"testing"
"github.com/btcsuite/btcd/wire"
"github.com/lightningnetwork/lnd/lnwallet"
"github.com/stretchr/testify/require"
)
// TestConfirmationHeightForUtxo verifies first-confirmation height lookup for
// wallet UTXOs.
func TestConfirmationHeightForUtxo(t *testing.T) {
t.Run("unconfirmed", func(t *testing.T) {
height, err := confirmationHeightForUtxo(0, &lnwallet.Utxo{})
require.NoError(t, err)
require.Zero(t, height)
})
t.Run("confirmed uses best height arithmetic", func(t *testing.T) {
height, err := confirmationHeightForUtxo(101, &lnwallet.Utxo{
Confirmations: 1,
OutPoint: wire.OutPoint{
Index: 1,
},
})
require.NoError(t, err)
require.EqualValues(t, 101, height)
})
t.Run("rejects impossible height", func(t *testing.T) {
_, err := confirmationHeightForUtxo(2, &lnwallet.Utxo{
Confirmations: 4,
OutPoint: wire.OutPoint{
Index: 3,
},
})
require.ErrorContains(t, err, "invalid confirmation height")
})
}

350
staticaddr/deposit/manager_reconcile_test.go Normal file
View file

@ -0,0 +1,350 @@
package deposit
import (
"context"
"errors"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/btcsuite/btcd/btcutil"
"github.com/btcsuite/btcd/chaincfg/chainhash"
"github.com/btcsuite/btcd/wire"
"github.com/lightninglabs/loop/staticaddr/script"
"github.com/lightninglabs/loop/staticaddr/version"
"github.com/lightninglabs/loop/test"
"github.com/lightningnetwork/lnd/lnwallet"
"github.com/stretchr/testify/mock"
"github.com/stretchr/testify/require"
)
// expectStableBestBlock configures two stable best-block lookups.
func expectStableBestBlock(mockChainKit *MockChainKit, height int32) {
mockChainKit.On(
"GetBestBlock", mock.Anything,
).Return(chainhash.Hash{}, height, nil).Twice()
}
// TestReconcileDepositsSerialized verifies reconciliation is serialized across
// concurrent callers.
func TestReconcileDepositsSerialized(t *testing.T) {
ctx := context.Background()
mockLnd := test.NewMockLnd()
utxo := &lnwallet.Utxo{
AddressType: lnwallet.TaprootPubkey,
Value: btcutil.Amount(100_000),
Confirmations: 0,
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{1},
Index: 1,
},
}
mockAddressManager := new(mockAddressManager)
mockAddressManager.On(
"ListUnspent", mock.Anything, int32(0), int32(MaxConfs),
).Return([]*lnwallet.Utxo{utxo}, nil)
mockAddressManager.On(
"GetStaticAddressParameters", mock.Anything,
).Return((*script.Parameters)(nil), errors.New("fsm init failed"))
mockStore := new(mockStore)
var createCalls atomic.Int32
createEntered := make(chan struct{})
releaseCreate := make(chan struct{})
mockStore.On(
"CreateDeposit", mock.Anything, mock.Anything,
).Return(nil).Run(func(mock.Arguments) {
if createCalls.Add(1) == 1 {
close(createEntered)
}
<-releaseCreate
})
manager := NewManager(&ManagerConfig{
AddressManager: mockAddressManager,
Store: mockStore,
WalletKit: mockLnd.WalletKit,
Signer: mockLnd.Signer,
})
var wg sync.WaitGroup
wg.Add(2)
errs := make(chan error, 2)
go func() {
defer wg.Done()
errs <- manager.reconcileDeposits(ctx)
}()
<-createEntered
go func() {
defer wg.Done()
errs <- manager.reconcileDeposits(ctx)
}()
time.Sleep(100 * time.Millisecond)
close(releaseCreate)
wg.Wait()
close(errs)
var gotErrs []error
for err := range errs {
gotErrs = append(gotErrs, err)
}
require.EqualValues(t, 1, createCalls.Load())
require.Len(t, manager.deposits, 1)
require.Empty(t, manager.activeDeposits)
require.Len(t, gotErrs, 2)
var errCount int
for _, err := range gotErrs {
if err == nil {
continue
}
errCount++
require.ErrorContains(t, err, "unable to start new deposit FSM")
}
require.Equal(t, 1, errCount)
}
// TestReconcileConfirmedDepositUsesBestBlockHeight verifies confirmation
// heights are derived from a stable chain tip.
func TestReconcileConfirmedDepositUsesBestBlockHeight(t *testing.T) {
ctx := context.Background()
mockLnd := test.NewMockLnd()
utxo := &lnwallet.Utxo{
AddressType: lnwallet.TaprootPubkey,
Value: btcutil.Amount(100_000),
Confirmations: 3,
OutPoint: wire.OutPoint{
Hash: chainhash.Hash{8},
Index: 1,
},
}
mockAddressManager := new(mockAddressManager)
mockAddressManager.On(
"ListUnspent", mock.Anything, int32(0), int32(MaxConfs),
).Return([]*lnwallet.Utxo{utxo}, nil)
mockAddressManager.On(
"GetStaticAddressParameters", mock.Anything,
).Return((*script.Parameters)(nil), errors.New("fsm init failed"))
mockChainKit := new(MockChainKit)
expectStableBestBlock(mockChainKit, 100)
mockStore := new(mockStore)
mockStore.On(
"CreateDeposit", mock.Anything, mock.Anything,
).Return(nil).Run(func(args mock.Arguments) {
createdDeposit := args.Get(1).(*Deposit)
require.EqualValues(t, 98, createdDeposit.ConfirmationHeight)
})
manager := NewManager(&ManagerConfig{
AddressManager: mockAddressManager,
ChainKit: mockChainKit,
Store: mockStore,
WalletKit: mockLnd.WalletKit,
Signer: mockLnd.Signer,
})
err := manager.reconcileDeposits(ctx)
require.ErrorContains(t, err, "unable to start new deposit FSM")
}
// TestReconcileDepositsInvalidatesVanishedUnconfirmedDeposit verifies that a
// single missing ListUnspent observation is reversible, but repeated misses
// still mark the deposit as replaced.
func TestReconcileDepositsInvalidatesVanishedUnconfirmedDeposit(t *testing.T) {
ctx := context.Background()
outpoint := wire.OutPoint{
Hash: chainhash.Hash{2},
Index: 7,
}
deposit := &Deposit{
OutPoint: outpoint,
}
deposit.SetState(Deposited)
mockAddressManager := new(mockAddressManager)
mockAddressManager.On(
"ListUnspent", mock.Anything, int32(0), int32(MaxConfs),
).Return([]*lnwallet.Utxo{}, nil)
mockStore := new(mockStore)
var updateCalls atomic.Int32
mockStore.On(
"UpdateDeposit", mock.Anything, mock.Anything,
).Return(nil).Run(func(args mock.Arguments) {
updateCalls.Add(1)
updatedDeposit := args.Get(1).(*Deposit)
require.True(t, updatedDeposit.IsInStateNoLock(Replaced))
})
manager := NewManager(&ManagerConfig{
AddressManager: mockAddressManager,
Store: mockStore,
})
manager.deposits[outpoint] = deposit
fsm := &FSM{
stopChan: make(chan struct{}),
quitChan: make(chan struct{}),
}
go func() {
<-fsm.stopChan
close(fsm.quitChan)
}()
manager.activeDeposits[outpoint] = fsm
// The first miss only increments the consecutive-miss counter.
require.NoError(t, manager.reconcileDeposits(ctx))
require.EqualValues(t, 0, updateCalls.Load())
require.Equal(t, Deposited, deposit.GetState())
require.Len(t, manager.activeDeposits, 1)
// The second consecutive miss is strong enough evidence to finalize the
// record as replaced.
require.NoError(t, manager.reconcileDeposits(ctx))
require.EqualValues(t, 1, updateCalls.Load())
require.Equal(t, Replaced, deposit.GetState())
require.Empty(t, manager.activeDeposits)
select {
case <-fsm.quitChan:
case <-time.After(time.Second):
t.Fatal("fsm did not stop after deposit was replaced")
}
}
// TestReconcileDepositsInvalidatesVanishedConfirmedDeposit verifies that a
// previously confirmed deposit is also marked replaced if it vanishes from the
// wallet view for multiple consecutive observations.
func TestReconcileDepositsInvalidatesVanishedConfirmedDeposit(t *testing.T) {
ctx := context.Background()
outpoint := wire.OutPoint{
Hash: chainhash.Hash{9},
Index: 4,
}
deposit := &Deposit{
OutPoint: outpoint,
ConfirmationHeight: 123,
}
deposit.SetState(Deposited)
mockAddressManager := new(mockAddressManager)
mockAddressManager.On(
"ListUnspent", mock.Anything, int32(0), int32(MaxConfs),
).Return([]*lnwallet.Utxo{}, nil)
mockStore := new(mockStore)
var updateCalls atomic.Int32
mockStore.On(
"UpdateDeposit", mock.Anything, mock.Anything,
).Return(nil).Run(func(args mock.Arguments) {
updateCalls.Add(1)
updatedDeposit := args.Get(1).(*Deposit)
require.True(t, updatedDeposit.IsInStateNoLock(Replaced))
require.EqualValues(
t, 123, updatedDeposit.ConfirmationHeight,
)
})
manager := NewManager(&ManagerConfig{
AddressManager: mockAddressManager,
Store: mockStore,
})
manager.deposits[outpoint] = deposit
fsm := &FSM{
stopChan: make(chan struct{}),
quitChan: make(chan struct{}),
}
go func() {
<-fsm.stopChan
close(fsm.quitChan)
}()
manager.activeDeposits[outpoint] = fsm
require.NoError(t, manager.reconcileDeposits(ctx))
require.EqualValues(t, 0, updateCalls.Load())
require.Equal(t, Deposited, deposit.GetState())
require.Len(t, manager.activeDeposits, 1)
require.NoError(t, manager.reconcileDeposits(ctx))
require.EqualValues(t, 1, updateCalls.Load())
require.Equal(t, Replaced, deposit.GetState())
require.Empty(t, manager.activeDeposits)
select {
case <-fsm.quitChan:
case <-time.After(time.Second):
t.Fatal("fsm did not stop after confirmed deposit was replaced")
}
}
// TestReconcileDepositsReactivatesReappearedReplacedDeposit verifies that the
// same outpoint can be revived if lnd reports it again after being marked
// replaced.
func TestReconcileDepositsReactivatesReappearedReplacedDeposit(t *testing.T) {
ctx := context.Background()
outpoint := wire.OutPoint{
Hash: chainhash.Hash{3},
Index: 5,
}
deposit := &Deposit{
OutPoint: outpoint,
Value: btcutil.Amount(100_000),
ConfirmationHeight: 77,
}
deposit.SetState(Replaced)
utxo := &lnwallet.Utxo{
OutPoint: outpoint,
Value: deposit.Value,
Confirmations: 0,
}
mockAddressManager := new(mockAddressManager)
mockAddressManager.On(
"ListUnspent", mock.Anything, int32(0), int32(MaxConfs),
).Return([]*lnwallet.Utxo{utxo}, nil)
mockAddressManager.On(
"GetStaticAddressParameters", mock.Anything,
).Return(&script.Parameters{
ProtocolVersion: version.ProtocolVersion_V0,
}, nil)
mockAddressManager.On(
"GetStaticAddress", mock.Anything,
).Return((*script.StaticAddress)(nil), nil)
mockStore := new(mockStore)
mockStore.On(
"UpdateDeposit", mock.Anything, mock.Anything,
).Return(nil).Run(func(args mock.Arguments) {
updatedDeposit := args.Get(1).(*Deposit)
require.True(t, updatedDeposit.IsInStateNoLock(Deposited))
require.Zero(t, updatedDeposit.ConfirmationHeight)
})
manager := NewManager(&ManagerConfig{
AddressManager: mockAddressManager,
Store: mockStore,
})
manager.deposits[outpoint] = deposit
// Reconciliation should revive the existing record instead of creating a
// second deposit entry for the same outpoint.
require.NoError(t, manager.reconcileDeposits(ctx))
require.Equal(t, Deposited, deposit.GetState())
require.Zero(t, deposit.ConfirmationHeight)
require.Len(t, manager.activeDeposits, 1)
}

43
staticaddr/deposit/manager_test.go
View file

@ -215,6 +215,49 @@ func (m *MockChainNotifier) RegisterSpendNtfn(ctx context.Context,
args.Get(1).(chan error), args.Error(2)
}
type MockChainKit struct {
mock.Mock
}
// RawClientWithMacAuth implements lndclient.ChainKitClient for tests.
func (m *MockChainKit) RawClientWithMacAuth(
ctx context.Context) (context.Context, time.Duration,
chainrpc.ChainKitClient) {
return ctx, 0, nil
}
// GetBlock implements lndclient.ChainKitClient for tests.
func (m *MockChainKit) GetBlock(context.Context, chainhash.Hash) (
*wire.MsgBlock, error) {
panic("unexpected GetBlock call")
}
// GetBlockHeader implements lndclient.ChainKitClient for tests.
func (m *MockChainKit) GetBlockHeader(context.Context, chainhash.Hash) (
*wire.BlockHeader, error) {
panic("unexpected GetBlockHeader call")
}
// GetBestBlock returns the configured best-block mock response.
func (m *MockChainKit) GetBestBlock(ctx context.Context) (
chainhash.Hash, int32, error) {
args := m.Called(ctx)
return args.Get(0).(chainhash.Hash), args.Get(1).(int32),
args.Error(2)
}
// GetBlockHash implements lndclient.ChainKitClient for tests.
func (m *MockChainKit) GetBlockHash(context.Context, int64) (
chainhash.Hash, error) {
panic("unexpected GetBlockHash call")
}
// TestManager checks that the manager processes the right channel notifications
// while a deposit is expiring.
func TestManager(t *testing.T) {