From e3e6205a4b579d8cf6f4cf052154db7e2f29bc8d Mon Sep 17 00:00:00 2001 From: Gustavo Stingelin Date: Wed, 8 Jul 2026 23:39:05 -0300 Subject: [PATCH] staticaddr/loopin: publish status updates after persistence --- loopd/daemon.go | 13 +++++- loopd/static_loopin_status_updater.go | 47 ++++++++++++++++++++ loopd/swapclient_server_test.go | 20 +++++++++ staticaddr/loopin/actions.go | 5 +++ staticaddr/loopin/actions_test.go | 62 +++++++++++++++++++++++++++ staticaddr/loopin/fsm.go | 16 +++++++ staticaddr/loopin/manager.go | 3 ++ staticaddr/loopin/manager_test.go | 42 +++++++++++++++++- 8 files changed, 206 insertions(+), 2 deletions(-) create mode 100644 loopd/static_loopin_status_updater.go diff --git a/loopd/daemon.go b/loopd/daemon.go index 68b76141..f625aa27 100644 --- a/loopd/daemon.go +++ b/loopd/daemon.go @@ -110,6 +110,10 @@ type Daemon struct { macaroonService *lndclient.MacaroonService } +// staticLoopInStatusChanBuffer keeps the shared swap-status fanout from +// stalling static loop-in FSM progress during transient subscriber gaps. +const staticLoopInStatusChanBuffer = 20 + // New creates a new instance of the loop client daemon. func New(config *Config, lisCfg *ListenerCfg) *Daemon { return &Daemon{ @@ -681,6 +685,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error { LightningClient: d.lnd.Client, } openChannelManager = openchannel.NewManager(openChannelCfg) + statusChan := make(chan loop.SwapInfo, staticLoopInStatusChanBuffer) // Run the deposit swap hash migration. err = loopin.MigrateDepositSwapHash( @@ -701,6 +706,11 @@ func (d *Daemon) initialize(withMacaroonService bool) error { return err } + statusUpdater := &staticLoopInStatusUpdater{ + statusChan: statusChan, + mainCtx: d.mainCtx, + chainParams: d.lnd.ChainParams, + } staticLoopInManager, err = loopin.NewManager(&loopin.Config{ Server: staticAddressClient, @@ -718,6 +728,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error { ChainParams: d.lnd.ChainParams, Signer: d.lnd.Signer, ValidateLoopInContract: loop.ValidateLoopInContract, + SendUpdate: statusUpdater.sendUpdate, MaxStaticAddrHtlcFeePercentage: d.cfg.MaxStaticAddrHtlcFeePercentage, MaxStaticAddrHtlcBackupFeePercentage: d.cfg.MaxStaticAddrHtlcBackupFeePercentage, }, blockHeight) @@ -783,7 +794,7 @@ func (d *Daemon) initialize(withMacaroonService bool) error { lnd: &d.lnd.LndServices, swaps: make(map[lntypes.Hash]loop.SwapInfo), subscribers: make(map[int]chan<- any), - statusChan: make(chan loop.SwapInfo), + statusChan: statusChan, mainCtx: d.mainCtx, reservationManager: reservationManager, instantOutManager: instantOutManager, diff --git a/loopd/static_loopin_status_updater.go b/loopd/static_loopin_status_updater.go new file mode 100644 index 00000000..aa0592cb --- /dev/null +++ b/loopd/static_loopin_status_updater.go @@ -0,0 +1,47 @@ +package loopd + +import ( + "context" + "fmt" + + "github.com/btcsuite/btcd/chaincfg" + "github.com/lightninglabs/loop" + "github.com/lightninglabs/loop/staticaddr/loopin" +) + +// staticLoopInStatusUpdater publishes static-address loop-in status updates to +// the client-facing swap stream. +type staticLoopInStatusUpdater struct { + // statusChan sends updates to the client-facing swap stream. + statusChan chan<- loop.SwapInfo + + // mainCtx is canceled when loopd is shutting down. + mainCtx context.Context + + // chainParams are used to reconstruct the static loop-in HTLC address. + chainParams *chaincfg.Params +} + +// sendUpdate converts the persisted static-address loop-in into swap info and +// forwards it to the status stream unless either context is canceled. +func (u *staticLoopInStatusUpdater) sendUpdate(ctx context.Context, + swp *loopin.StaticAddressLoopIn) error { + + info, err := staticAddressLoopInSwapInfoWithChainParams( + swp, u.chainParams, + ) + if err != nil { + return fmt.Errorf("unable to notify static loop-in update: %w", err) + } + + select { + case u.statusChan <- *info: + return nil + + case <-ctx.Done(): + return ctx.Err() + + case <-u.mainCtx.Done(): + return u.mainCtx.Err() + } +} diff --git a/loopd/swapclient_server_test.go b/loopd/swapclient_server_test.go index 9d1acf3c..03f3d95a 100644 --- a/loopd/swapclient_server_test.go +++ b/loopd/swapclient_server_test.go @@ -661,6 +661,26 @@ func TestMonitorSnapshotIncludesFinalStaticAddressLoopIns(t *testing.T) { require.Len(t, completedSwaps, 1) } +// TestStaticLoopInStatusUpdaterUsesSwapHtlcAddress protects the live-update +// invariant that static loop-in status events derive the HTLC address from the +// swap, not from reusable static address parameters. +func TestStaticLoopInStatusUpdaterUsesSwapHtlcAddress(t *testing.T) { + ctx := t.Context() + _, staticLoopIn := newGenericStaticLoopInServer(t) + staticLoopIn.AddressParams = nil + statusChan := make(chan loop.SwapInfo, 1) + updater := &staticLoopInStatusUpdater{ + statusChan: statusChan, + mainCtx: ctx, + chainParams: &chaincfg.TestNet3Params, + } + + err := updater.sendUpdate(ctx, staticLoopIn) + require.NoError(t, err) + swapInfo := <-statusChan + assertStaticLoopInUsesSwapHtlcAddress(t, staticLoopIn, swapInfo) +} + // TestMonitorSuppressesStaticAddressLoopInSnapshotLiveDuplicate protects the // monitor race invariant that live static loop-in updates arriving during the // initial snapshot are deduplicated without dropping newer progress. diff --git a/staticaddr/loopin/actions.go b/staticaddr/loopin/actions.go index 1c432e6f..3dc1c029 100644 --- a/staticaddr/loopin/actions.go +++ b/staticaddr/loopin/actions.go @@ -341,6 +341,11 @@ func (f *FSM) InitHtlcAction(ctx context.Context, // Once the swap is stored, restart/recovery code owns invoice lifecycle. invoiceNeedsCleanup = false + err = f.sendUpdate(ctx) + if err != nil { + f.Errorf("Error sending loop-in update: %v", err) + } + event = OnHtlcInitiated return event diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index 20b86251..afc0085d 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -878,6 +878,68 @@ func TestCheckDepositsAvailableRejectsDivergentDepositOutpoints( require.Empty(t, checker.outpoints) } +// TestInitHtlcActionIgnoresSendUpdateErrorAfterPersistence protects the +// persistence-first invariant: once the loop-in is stored, a later status +// update failure must not roll back the action or state transition. +func TestInitHtlcActionIgnoresSendUpdateErrorAfterPersistence(t *testing.T) { + mockLnd := test.NewMockLnd() + _, serverKey := test.CreateKey(22) + + server := &mockStaticAddressServer{ + response: testStaticAddressLoopInResponse( + serverKey.SerializeCompressed(), + ), + } + + dep := &deposit.Deposit{ + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{2}, + Index: 0, + }, + Value: 500_000, + } + + loopIn := &StaticAddressLoopIn{ + Deposits: []*deposit.Deposit{dep}, + DepositOutpoints: []string{dep.OutPoint.String()}, + SelectedAmount: dep.Value, + QuotedSwapFee: 1_000, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + PaymentTimeoutSeconds: 3_600, + } + + sendUpdateErr := errors.New("status channel blocked") + sendUpdateCalled := false + f := &FSM{ + StateMachine: &fsm.StateMachine{}, + cfg: &Config{ + Server: server, + DepositManager: &noopDepositManager{}, + LndClient: mockLnd.Client, + WalletKit: mockLnd.WalletKit, + ChainParams: mockLnd.ChainParams, + Store: &mockStore{}, + ValidateLoopInContract: testValidateLoopInContract, + MaxStaticAddrHtlcFeePercentage: 1, + MaxStaticAddrHtlcBackupFeePercentage: 1, + SendUpdate: func(context.Context, + *StaticAddressLoopIn) error { + + sendUpdateCalled = true + + return sendUpdateErr + }, + }, + loopIn: loopIn, + } + + event := f.InitHtlcAction(t.Context(), nil) + require.Equal(t, OnHtlcInitiated, event) + require.Nil(t, f.LastActionError) + require.True(t, sendUpdateCalled) +} + // mockStaticAddressServer captures static-address loop-in requests in tests. type mockStaticAddressServer struct { swapserverrpc.StaticAddressServerClient diff --git a/staticaddr/loopin/fsm.go b/staticaddr/loopin/fsm.go index ea6e610f..eb53dada 100644 --- a/staticaddr/loopin/fsm.go +++ b/staticaddr/loopin/fsm.go @@ -273,6 +273,22 @@ func (f *FSM) updateLoopIn(ctx context.Context, notification fsm.Notification) { return } + + err = f.sendUpdate(ctx) + if err != nil { + f.Errorf("Error sending loop-in update: %v", err) + } +} + +// sendUpdate publishes the latest loop-in state after it has been persisted. +// The callback must remain lightweight because it runs synchronously with FSM +// state transitions. +func (f *FSM) sendUpdate(ctx context.Context) error { + if f.cfg.SendUpdate == nil { + return nil + } + + return f.cfg.SendUpdate(ctx, f.loopIn) } // isUpdateSkipped returns true if the loop-in should not be updated for the diff --git a/staticaddr/loopin/manager.go b/staticaddr/loopin/manager.go index 92c4a9f1..47a447fc 100644 --- a/staticaddr/loopin/manager.go +++ b/staticaddr/loopin/manager.go @@ -98,6 +98,9 @@ type Config struct { // request. ValidateLoopInContract ValidateLoopInContract + // SendUpdate publishes a loop-in status update after it is persisted. + SendUpdate func(context.Context, *StaticAddressLoopIn) error + // MaxStaticAddrHtlcFeePercentage is the percentage of the swap amount // that we allow the server to charge for the htlc transaction. // Although highly unlikely, this is a defense against the server diff --git a/staticaddr/loopin/manager_test.go b/staticaddr/loopin/manager_test.go index 564429c6..9fa8587c 100644 --- a/staticaddr/loopin/manager_test.go +++ b/staticaddr/loopin/manager_test.go @@ -245,6 +245,45 @@ func TestInitiateLoopInAllowsReservedAutoloopLabel(t *testing.T) { require.Equal(t, selectedDeposit.Value, quoteGetter.amount) } +// TestUpdateLoopInSendsUpdateAfterSuccessfulStoreUpdate protects the +// notification contract: after a successful database update, the manager must +// publish the stored loop-in state to listeners. +func TestUpdateLoopInSendsUpdateAfterSuccessfulStoreUpdate(t *testing.T) { + ctx := t.Context() + swapHash := lntypes.Hash{1, 2, 3} + updates := make(chan *StaticAddressLoopIn, 1) + loopIn := &StaticAddressLoopIn{SwapHash: swapHash} + loopIn.SetState(SignHtlcTx) + + loopInFsm := &FSM{ + cfg: &Config{ + Store: &mockStore{stored: true}, + SendUpdate: func(_ context.Context, + updated *StaticAddressLoopIn) error { + + updates <- updated + + return nil + }, + }, + loopIn: loopIn, + } + + loopInFsm.updateLoopIn(ctx, fsm.Notification{ + PreviousState: SignHtlcTx, + NextState: MonitorInvoiceAndHtlcTx, + }) + + select { + case updated := <-updates: + require.Equal(t, swapHash, updated.SwapHash) + require.Equal(t, MonitorInvoiceAndHtlcTx, updated.GetState()) + + case <-ctx.Done(): + t.Fatal(ctx.Err()) + } +} + // TestHandleLoopInSweepReqRejectsInvalidServerNonce ensures that a malformed // MuSig2 nonce returned by the server is rejected before it reaches the signer. func TestHandleLoopInSweepReqRejectsInvalidServerNonce(t *testing.T) { @@ -501,6 +540,7 @@ type mockStore struct { swaps []*StaticAddressLoopIn loopIns map[lntypes.Hash]*StaticAddressLoopIn mapIDs map[lntypes.Hash][]deposit.ID + stored bool } func (s *mockStore) CreateLoopIn(_ context.Context, @@ -521,7 +561,7 @@ func (s *mockStore) GetStaticAddressLoopInSwapsByStates(_ context.Context, return s.swaps, nil } func (s *mockStore) IsStored(_ context.Context, _ lntypes.Hash) (bool, error) { - return false, nil + return s.stored, nil } // RecordStaticAddressRiskDecision implements Store for manager tests.