From 44a6c7a1447625a8579edc65d612d2e31efc9ab6 Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Fri, 19 Jun 2026 14:40:30 +0200 Subject: [PATCH 1/5] staticaddr/loopin: cancel orphan invoice on init failure A static address loop-in that creates an invoice but then fails before storage cannot be recovered after restart, and persisted error paths that unlock deposits should not leave the swap invoice live either. The payment-deadline path also needs to report the real deposit unlock error instead of checking the FSM event returned by UnlockDepositsAction, because that action returns OnError on both success and failure. Cancel private swap invoices on pre-storage init failures, monitor timeouts, and stored unlock paths using a detached timeout-limited helper. Factor deposit unlocking into an error-returning helper so deadline handling can log transition failures, and add coverage for early-init cancellation plus generic unlock invoice cleanup and error propagation. --- staticaddr/loopin/actions.go | 115 ++++++++++---- staticaddr/loopin/actions_test.go | 239 ++++++++++++++++++++++++++++++ 2 files changed, 324 insertions(+), 30 deletions(-) diff --git a/staticaddr/loopin/actions.go b/staticaddr/loopin/actions.go index 70a27811..ccf86b93 100644 --- a/staticaddr/loopin/actions.go +++ b/staticaddr/loopin/actions.go @@ -36,6 +36,8 @@ const ( defaultConfTarget = 3 DefaultPaymentTimeoutSeconds = 60 + + defaultInvoiceCleanupTimeout = 5 * time.Second ) var ( @@ -57,6 +59,24 @@ var ( func (f *FSM) InitHtlcAction(ctx context.Context, _ fsm.EventContext) fsm.EventType { + var event fsm.EventType + invoiceNeedsCleanup := false + defer func() { + // If we created the private invoice but failed before persisting the + // swap, cancel it so retries do not accumulate orphan invoices. + if !invoiceNeedsCleanup || event != fsm.OnError { + return + } + + f.cancelSwapInvoice() + }() + + returnError := func(err error) fsm.EventType { + event = f.HandleError(err) + + return event + } + // Lock the deposits and transition them to the LoopingIn state. err := f.cfg.DepositManager.TransitionDeposits( ctx, f.loopIn.Deposits, deposit.OnLoopInInitiated, @@ -65,7 +85,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { err = fmt.Errorf("unable to loop-in deposits: %w", err) - return f.HandleError(err) + return returnError(err) } // Calculate the swap invoice amount. The server needs to pay us the @@ -88,7 +108,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, err = fmt.Errorf("unable to create random swap preimage: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.SwapPreimage = swapPreimage f.loopIn.SwapHash = swapPreimage.Hash() @@ -100,7 +120,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { err = fmt.Errorf("unable to derive client htlc key: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.ClientPubkey = keyDesc.PubKey f.loopIn.HtlcKeyLocator = keyDesc.KeyLocator @@ -119,10 +139,14 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { err = fmt.Errorf("unable to create swap invoice: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.SwapInvoice = swapInvoice + // From here until CreateLoopIn succeeds, any error path would otherwise + // leave behind a live invoice with no persisted swap to recover it. + invoiceNeedsCleanup = true + f.loopIn.ProtocolVersion = version.AddressProtocolVersion( version.CurrentRPCProtocolVersion(), ) @@ -149,7 +173,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, err = fmt.Errorf("unable to initiate the loop-in with the "+ "server: %w", err) - return f.HandleError(err) + return returnError(err) } // Pushing empty sigs signals the server that we abandoned the swap @@ -171,7 +195,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, pushEmptySigs() err = fmt.Errorf("unable to parse server pubkey: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.ServerPubkey = serverPubkey @@ -185,7 +209,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, err = fmt.Errorf("server response parameters are outside "+ "our allowed range: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.HtlcCltvExpiry = loopInResp.HtlcExpiry @@ -194,7 +218,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, pushEmptySigs() err = fmt.Errorf("unable to convert server nonces: %w", err) - return f.HandleError(err) + return returnError(err) } f.htlcServerNoncesHighFee, err = toNonces( loopInResp.HighFeeHtlcInfo.Nonces, @@ -202,7 +226,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { pushEmptySigs() - return f.HandleError(err) + return returnError(err) } f.htlcServerNoncesExtremelyHighFee, err = toNonces( loopInResp.ExtremeFeeHtlcInfo.Nonces, @@ -210,7 +234,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, if err != nil { pushEmptySigs() - return f.HandleError(err) + return returnError(err) } // We need to defend against the server setting high fees for the htlc @@ -232,7 +256,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, log.Errorf("server htlc tx fee is higher than the configured "+ "allowed maximum: %v > %v", fee, maxHtlcTxFee) - return f.HandleError(ErrFeeTooHigh) + return returnError(ErrFeeTooHigh) } f.loopIn.HtlcTxFeeRate = feeRate @@ -246,7 +270,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, "configured allowed maximum: %v > %v", fee, maxHtlcTxBackupFee) - return f.HandleError(ErrFeeTooHigh) + return returnError(ErrFeeTooHigh) } f.loopIn.HtlcTxHighFeeRate = highFeeRate @@ -262,7 +286,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, "configured allowed maximum: %v > %v", fee, maxHtlcTxBackupFee) - return f.HandleError(ErrFeeTooHigh) + return returnError(ErrFeeTooHigh) } f.loopIn.HtlcTxExtremelyHighFeeRate = extremelyHighFeeRate @@ -276,7 +300,7 @@ func (f *FSM) InitHtlcAction(ctx context.Context, err = fmt.Errorf("unable to derive htlc timeout sweep "+ "address: %w", err) - return f.HandleError(err) + return returnError(err) } f.loopIn.HtlcTimeoutSweepAddress = sweepAddress @@ -286,10 +310,34 @@ func (f *FSM) InitHtlcAction(ctx context.Context, pushEmptySigs() err = fmt.Errorf("unable to store loop-in in db: %w", err) - return f.HandleError(err) + return returnError(err) } - return OnHtlcInitiated + // Once the swap is stored, restart/recovery code owns invoice lifecycle. + invoiceNeedsCleanup = false + + event = OnHtlcInitiated + + return event +} + +// cancelSwapInvoice best-effort cancels the current swap invoice using a +// detached timeout-limited context. +func (f *FSM) cancelSwapInvoice() { + if f.loopIn.SwapInvoice == "" { + return + } + + cleanupCtx, cancel := context.WithTimeout( + context.Background(), defaultInvoiceCleanupTimeout, + ) + defer cancel() + + err := f.cfg.InvoicesClient.CancelInvoice(cleanupCtx, f.loopIn.SwapHash) + if err != nil { + f.Warnf("unable to cancel invoice for swap %v: %v", + f.loopIn.SwapHash, err) + } } // SignHtlcTxAction is called if the htlc was initialized and the server @@ -557,11 +605,9 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // Cancel the lndclient invoice subscription. cancelInvoiceSubscription() - err = f.cfg.InvoicesClient.CancelInvoice(ctx, f.loopIn.SwapHash) - if err != nil { - f.Warnf("unable to cancel invoice "+ - "for swap hash: %v", err) - } + // Reuse the same helper as InitHtlcAction so timeout cleanup + // follows the same detached-context path as early-init cleanup. + f.cancelSwapInvoice() } for { @@ -609,10 +655,10 @@ func (f *FSM) MonitorInvoiceAndHtlcTxAction(ctx context.Context, // re-enable them for loop-ins and withdrawals. cancelInvoice() - event := f.UnlockDepositsAction(ctx, nil) - if event != fsm.OnError { - f.Errorf("unable to unlock deposits after " + - "payment deadline") + err = f.unlockDeposits(ctx) + if err != nil { + f.Errorf("unable to unlock deposits after "+ + "payment deadline: %v", err) } case currentHeight := <-blockChan: @@ -824,18 +870,27 @@ func (f *FSM) PaymentReceivedAction(ctx context.Context, func (f *FSM) UnlockDepositsAction(ctx context.Context, _ fsm.EventContext) fsm.EventType { - err := f.cfg.DepositManager.TransitionDeposits( - ctx, f.loopIn.Deposits, fsm.OnError, deposit.Deposited, - ) - if err != nil { - err = fmt.Errorf("unable to unlock deposits: %w", err) + f.cancelSwapInvoice() + err := f.unlockDeposits(ctx) + if err != nil { return f.HandleError(err) } return fsm.OnError } +func (f *FSM) unlockDeposits(ctx context.Context) error { + err := f.cfg.DepositManager.TransitionDeposits( + ctx, f.loopIn.Deposits, fsm.OnError, deposit.Deposited, + ) + if err != nil { + return fmt.Errorf("unable to unlock deposits: %w", err) + } + + return nil +} + // createAndPublishHtlcTimeoutSweepTx creates and publishes the htlc timeout // sweep transaction. func (f *FSM) createAndPublishHtlcTimeoutSweepTx(ctx context.Context) error { diff --git a/staticaddr/loopin/actions_test.go b/staticaddr/loopin/actions_test.go index b058e328..d75adf6c 100644 --- a/staticaddr/loopin/actions_test.go +++ b/staticaddr/loopin/actions_test.go @@ -270,6 +270,192 @@ func testValidateLoopInContract(_ int32, _ int32) error { return nil } +// TestInitHtlcActionCancelsInvoiceOnServerError verifies that an invoice +// created before a server-side rejection is canceled immediately. +func TestInitHtlcActionCancelsInvoiceOnServerError(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + mockLnd := test.NewMockLnd() + + loopIn := &StaticAddressLoopIn{ + Deposits: []*deposit.Deposit{{ + Value: 200_000, + }}, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + PaymentTimeoutSeconds: DefaultPaymentTimeoutSeconds, + ProtocolVersion: version.ProtocolVersion_V0, + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + DepositManager: &noopDepositManager{}, + WalletKit: mockLnd.WalletKit, + LndClient: mockLnd.Client, + InvoicesClient: mockLnd.LndServices.Invoices, + Server: &initHtlcTestServer{ + loopInErr: errors.New("server rejected swap"), + }, + } + + f, err := NewFSM(ctx, loopIn, cfg, false) + require.NoError(t, err) + + // The init step should fail and synchronously trigger deferred invoice + // cleanup. + event := f.InitHtlcAction(ctx, nil) + require.Equal(t, fsm.OnError, event) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, loopIn.SwapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } +} + +// TestInitHtlcActionCancelsInvoiceOnFeeGuardFailure verifies that the early +// fee guard also cancels the pre-created invoice before returning an error. +func TestInitHtlcActionCancelsInvoiceOnFeeGuardFailure(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + mockLnd := test.NewMockLnd() + serverKey, err := btcec.NewPrivateKey() + require.NoError(t, err) + + loopIn := &StaticAddressLoopIn{ + Deposits: []*deposit.Deposit{{ + Value: 200_000, + }}, + InitiationHeight: uint32(mockLnd.Height), + InitiationTime: time.Now(), + PaymentTimeoutSeconds: DefaultPaymentTimeoutSeconds, + ProtocolVersion: version.ProtocolVersion_V0, + } + + cfg := &Config{ + AddressManager: &mockAddressManager{ + params: &script.Parameters{ + ProtocolVersion: version.ProtocolVersion_V0, + }, + }, + DepositManager: &noopDepositManager{}, + WalletKit: mockLnd.WalletKit, + LndClient: mockLnd.Client, + InvoicesClient: mockLnd.LndServices.Invoices, + Server: &initHtlcTestServer{ + loopInResp: &swapserverrpc.ServerStaticAddressLoopInResponse{ + HtlcServerPubKey: serverKey.PubKey(). + SerializeCompressed(), + HtlcExpiry: mockLnd.Height + + DefaultLoopInOnChainCltvDelta, + StandardHtlcInfo: &swapserverrpc.ServerHtlcSigningInfo{ + FeeRate: 1_000_000, + }, + HighFeeHtlcInfo: &swapserverrpc.ServerHtlcSigningInfo{}, + ExtremeFeeHtlcInfo: &swapserverrpc. + ServerHtlcSigningInfo{}, + }, + }, + ValidateLoopInContract: func(int32, int32) error { + return nil + }, + MaxStaticAddrHtlcFeePercentage: 0, + MaxStaticAddrHtlcBackupFeePercentage: 1, + } + + f, err := NewFSM(ctx, loopIn, cfg, false) + require.NoError(t, err) + + // The fee guard runs before persistence, so the deferred cleanup must + // cancel the invoice on this error path as well. + event := f.InitHtlcAction(ctx, nil) + require.Equal(t, fsm.OnError, event) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, loopIn.SwapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } +} + +// TestUnlockDepositsActionCancelsInvoice verifies that stored swaps that enter +// the generic error unlock path also clean up their swap invoice. +func TestUnlockDepositsActionCancelsInvoice(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + defer cancel() + + mockLnd := test.NewMockLnd() + dep := &deposit.Deposit{ + Value: 200_000, + } + swapHash := lntypes.Hash{0x44, 0x55} + depositMgr := &recordingDepositManager{} + + f := &FSM{ + StateMachine: &fsm.StateMachine{}, + cfg: &Config{ + DepositManager: depositMgr, + InvoicesClient: mockLnd.LndServices.Invoices, + }, + loopIn: &StaticAddressLoopIn{ + SwapHash: swapHash, + SwapInvoice: "lnbc1test", + Deposits: []*deposit.Deposit{dep}, + }, + } + + event := f.UnlockDepositsAction(ctx, nil) + require.Equal(t, fsm.OnError, event) + require.NoError(t, f.LastActionError) + + select { + case hash := <-mockLnd.FailInvoiceChannel: + require.Equal(t, swapHash, hash) + + case <-ctx.Done(): + t.Fatalf("invoice was not canceled: %v", ctx.Err()) + } + + require.Len(t, depositMgr.transitions, 1) + require.Equal(t, []*deposit.Deposit{dep}, depositMgr.transitions[0].deposits) + require.Equal(t, fsm.OnError, depositMgr.transitions[0].event) + require.Equal(t, deposit.Deposited, depositMgr.transitions[0].state) +} + +// TestUnlockDepositsActionReportsTransitionError ensures the unlock path +// preserves the real deposit transition failure for callers that need to log it. +func TestUnlockDepositsActionReportsTransitionError(t *testing.T) { + depositMgr := &recordingDepositManager{ + err: errors.New("transition failed"), + } + f := &FSM{ + StateMachine: &fsm.StateMachine{}, + cfg: &Config{ + DepositManager: depositMgr, + }, + loopIn: &StaticAddressLoopIn{ + Deposits: []*deposit.Deposit{{Value: 200_000}}, + }, + } + + event := f.UnlockDepositsAction(t.Context(), nil) + require.Equal(t, fsm.OnError, event) + require.ErrorContains( + t, f.LastActionError, "unable to unlock deposits", + ) + require.ErrorContains(t, f.LastActionError, "transition failed") +} + // mockAddressManager is a minimal AddressManager implementation used by the // test FSM setup. type mockAddressManager struct { @@ -327,3 +513,56 @@ func (n *noopDepositManager) GetActiveDepositsInState(fsm.StateType) ( return nil, nil } + +type depositTransition struct { + deposits []*deposit.Deposit + event fsm.EventType + state fsm.StateType +} + +type recordingDepositManager struct { + noopDepositManager + + err error + transitions []depositTransition +} + +// TransitionDeposits records the transition and returns the configured error. +func (r *recordingDepositManager) TransitionDeposits(_ context.Context, + deposits []*deposit.Deposit, event fsm.EventType, + state fsm.StateType) error { + + r.transitions = append(r.transitions, depositTransition{ + deposits: deposits, + event: event, + state: state, + }) + + return r.err +} + +// initHtlcTestServer lets InitHtlcAction tests inject a deterministic server +// response without standing up the full gRPC client. +type initHtlcTestServer struct { + swapserverrpc.StaticAddressServerClient + + loopInResp *swapserverrpc.ServerStaticAddressLoopInResponse + loopInErr error +} + +// ServerStaticAddressLoopIn returns the canned response configured by the test. +func (s *initHtlcTestServer) ServerStaticAddressLoopIn(context.Context, + *swapserverrpc.ServerStaticAddressLoopInRequest, ...grpc.CallOption, +) (*swapserverrpc.ServerStaticAddressLoopInResponse, error) { + + return s.loopInResp, s.loopInErr +} + +// PushStaticAddressHtlcSigs accepts the abandonment signal used by error-path +// tests without adding additional assertions. +func (s *initHtlcTestServer) PushStaticAddressHtlcSigs(context.Context, + *swapserverrpc.PushStaticAddressHtlcSigsRequest, ...grpc.CallOption, +) (*swapserverrpc.PushStaticAddressHtlcSigsResponse, error) { + + return &swapserverrpc.PushStaticAddressHtlcSigsResponse{}, nil +} From bbeb813bfa4aa815195f731b7374e58802bd0bb9 Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Fri, 19 Jun 2026 14:40:48 +0200 Subject: [PATCH 2/5] staticaddr/deposit: finalize deposits asynchronously Final deposit states should not stall while deposit locks are held, because a blocked manager receive loop can otherwise hold up the deposit FSM; if shutdown happens before notification delivery, startup recovery can still resume from the final state. Send finalization notifications from a goroutine so final states are recorded without waiting on the manager receive loop, and add tests for blocked manager delivery and shutdown races. --- staticaddr/deposit/actions.go | 25 ++++-- staticaddr/deposit/actions_test.go | 135 +++++++++++++++++++++++++++++ 2 files changed, 153 insertions(+), 7 deletions(-) create mode 100644 staticaddr/deposit/actions_test.go diff --git a/staticaddr/deposit/actions.go b/staticaddr/deposit/actions.go index 362417e7..437c3ac2 100644 --- a/staticaddr/deposit/actions.go +++ b/staticaddr/deposit/actions.go @@ -161,14 +161,25 @@ func (f *FSM) WaitForExpirySweepAction(ctx context.Context, // FinalizeDepositAction is the final action after a withdrawal. It signals to // the manager that the deposit has been swept and the FSM can be removed. -func (f *FSM) FinalizeDepositAction(ctx context.Context, +func (f *FSM) FinalizeDepositAction(_ context.Context, _ fsm.EventContext) fsm.EventType { - select { - case <-ctx.Done(): - return fsm.OnError + outpoint := f.deposit.OutPoint - case f.finalizedDepositChan <- f.deposit.OutPoint: - return fsm.NoOp - } + // The finalization notification only tells the manager to remove the + // deposit from its active set. Send it asynchronously so a busy manager + // loop can't stall withdrawal confirmation while deposit locks are held. + go func() { + select { + case <-f.quitChan: + // The deposit is already in a final state. If shutdown wins + // this race, startup recovery will skip it instead of + // re-adding it to the active set. + return + + case f.finalizedDepositChan <- outpoint: + } + }() + + return fsm.NoOp } diff --git a/staticaddr/deposit/actions_test.go b/staticaddr/deposit/actions_test.go new file mode 100644 index 00000000..8c021121 --- /dev/null +++ b/staticaddr/deposit/actions_test.go @@ -0,0 +1,135 @@ +package deposit + +import ( + "context" + "testing" + "time" + + "github.com/btcsuite/btcd/chaincfg/chainhash" + "github.com/btcsuite/btcd/wire" + "github.com/lightninglabs/loop/fsm" + "github.com/stretchr/testify/require" +) + +// TestFinalizeDepositActionDoesNotBlock ensures the final cleanup notification +// does not block the withdrawal completion path while the manager loop is busy. +func TestFinalizeDepositActionDoesNotBlock(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{1}, + Index: 1, + } + + depositFSM := &FSM{ + deposit: &Deposit{ + OutPoint: outpoint, + }, + quitChan: make(chan struct{}), + finalizedDepositChan: make(chan wire.OutPoint), + } + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- depositFSM.FinalizeDepositAction(ctx, nil) + }() + + select { + case result := <-resultChan: + require.Equal(t, fsm.NoOp, result) + + case <-time.After(100 * time.Millisecond): + t.Fatal("FinalizeDepositAction blocked on manager cleanup") + } + + select { + case gotOutpoint := <-depositFSM.finalizedDepositChan: + require.Equal(t, outpoint, gotOutpoint) + + case <-time.After(time.Second): + t.Fatal("finalization cleanup notification was not delivered") + } +} + +// TestFinalizeDepositActionIgnoresRequestCancellation ensures the cleanup +// notification is tied to the FSM lifetime, not the caller's request context. +func TestFinalizeDepositActionIgnoresRequestCancellation(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + quitChan := make(chan struct{}) + defer close(quitChan) + + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{2}, + Index: 2, + } + + depositFSM := &FSM{ + deposit: &Deposit{ + OutPoint: outpoint, + }, + quitChan: quitChan, + finalizedDepositChan: make(chan wire.OutPoint), + } + + resultChan := make(chan fsm.EventType, 1) + go func() { + resultChan <- depositFSM.FinalizeDepositAction(ctx, nil) + }() + + select { + case result := <-resultChan: + require.Equal(t, fsm.NoOp, result) + + case <-time.After(100 * time.Millisecond): + t.Fatal("FinalizeDepositAction blocked on manager cleanup") + } + + cancel() + + select { + case gotOutpoint := <-depositFSM.finalizedDepositChan: + require.Equal(t, outpoint, gotOutpoint) + + case <-time.After(time.Second): + t.Fatal("finalization cleanup notification was dropped after " + + "request cancellation") + } +} + +// TestFinalizeDepositActionIgnoresCanceledContext ensures the final cleanup +// notification is still queued even if the caller's context is already done. +func TestFinalizeDepositActionIgnoresCanceledContext(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + quitChan := make(chan struct{}) + defer close(quitChan) + + outpoint := wire.OutPoint{ + Hash: chainhash.Hash{3}, + Index: 3, + } + + depositFSM := &FSM{ + deposit: &Deposit{ + OutPoint: outpoint, + }, + quitChan: quitChan, + finalizedDepositChan: make(chan wire.OutPoint), + } + + result := depositFSM.FinalizeDepositAction(ctx, nil) + require.Equal(t, fsm.NoOp, result) + + select { + case gotOutpoint := <-depositFSM.finalizedDepositChan: + require.Equal(t, outpoint, gotOutpoint) + + case <-time.After(time.Second): + t.Fatal("finalization cleanup notification was dropped for " + + "an already-canceled request context") + } +} From f982ceb753ec69e8d36516f29f92d2f994ae1ec7 Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Fri, 19 Jun 2026 14:48:57 +0200 Subject: [PATCH 3/5] staticaddr/loopin: include failed swaps in state queries The final-state query previously formatted the state list with braces, which broke the comma-delimited match for both boundary entries: the first state was compared with a leading brace and the last state with a trailing brace. Final-state callers need swaps at both ends of the list, including HtlcTimeoutSwept and Failed, to remain visible once they reach a terminal status. Return the final-state list as plain comma-separated state names, update the store comment to match the query format, and extend the store test so it inserts and retrieves swaps in the first final state, a middle final state, and the last final state without relying on result order. --- staticaddr/loopin/sql_store.go | 6 +- staticaddr/loopin/sql_store_test.go | 117 ++++++++++++++++++++++++---- 2 files changed, 107 insertions(+), 16 deletions(-) diff --git a/staticaddr/loopin/sql_store.go b/staticaddr/loopin/sql_store.go index d06c5c18..d8c253be 100644 --- a/staticaddr/loopin/sql_store.go +++ b/staticaddr/loopin/sql_store.go @@ -56,8 +56,8 @@ type Querier interface { swapHash []byte) (sqlc.GetStaticAddressLoopInSwapRow, error) // GetStaticAddressLoopInSwapsByStates retrieves all swaps with the - // given states. The states string is an input for the IN primitive in - // sqlite, hence the format needs to be '{State1,State2,...}'. + // given states. The states string is comma-separated so the query can + // match complete state names by wrapping it with comma sentinels. GetStaticAddressLoopInSwapsByStates(ctx context.Context, states sql.NullString) ([]sqlc.GetStaticAddressLoopInSwapsByStatesRow, error) @@ -203,7 +203,7 @@ func (s *SqlStore) GetStaticAddressLoopInSwapsByStates(ctx context.Context, } func toJointStringStates(states []fsm.StateType) string { - return "{" + strings.Join(toStrings(states), ",") + "}" + return strings.Join(toStrings(states), ",") } func toStrings(states []fsm.StateType) []string { diff --git a/staticaddr/loopin/sql_store_test.go b/staticaddr/loopin/sql_store_test.go index 1e30081d..81fd10a7 100644 --- a/staticaddr/loopin/sql_store_test.go +++ b/staticaddr/loopin/sql_store_test.go @@ -41,8 +41,10 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { } loopingDepositID := newID() + timeoutDepositID := newID() loopedInDepositID := newID() - d1, d2 := &deposit.Deposit{ + failedDepositID := newID() + d1, d2, d3, d4 := &deposit.Deposit{ ID: loopingDepositID, OutPoint: wire.OutPoint{ Hash: chainhash.Hash{0x1a, 0x2b, 0x3c, 0x4d}, @@ -54,7 +56,7 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { }, }, &deposit.Deposit{ - ID: loopedInDepositID, + ID: timeoutDepositID, OutPoint: wire.OutPoint{ Hash: chainhash.Hash{0x2a, 0x2b, 0x3c, 0x4e}, Index: 1, @@ -63,29 +65,67 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { TimeOutSweepPkScript: []byte{ 0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x4d, }, + }, + &deposit.Deposit{ + ID: loopedInDepositID, + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{0x3a, 0x2b, 0x3c, 0x4e}, + Index: 2, + }, + Value: btcutil.Amount(300_000), + TimeOutSweepPkScript: []byte{ + 0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x4f, + }, + }, + &deposit.Deposit{ + ID: failedDepositID, + OutPoint: wire.OutPoint{ + Hash: chainhash.Hash{0x4a, 0x2b, 0x3c, 0x4e}, + Index: 3, + }, + Value: btcutil.Amount(400_000), + TimeOutSweepPkScript: []byte{ + 0x00, 0x14, 0x1a, 0x2b, 0x3c, 0x50, + }, } err := depositStore.CreateDeposit(ctxb, d1) require.NoError(t, err) err = depositStore.CreateDeposit(ctxb, d2) require.NoError(t, err) + err = depositStore.CreateDeposit(ctxb, d3) + require.NoError(t, err) + err = depositStore.CreateDeposit(ctxb, d4) + require.NoError(t, err) // Add two updates per deposit, expect the last to be retrieved. d1.SetState(deposit.Deposited) d2.SetState(deposit.Deposited) + d3.SetState(deposit.Deposited) + d4.SetState(deposit.Deposited) err = depositStore.UpdateDeposit(ctxb, d1) require.NoError(t, err) err = depositStore.UpdateDeposit(ctxb, d2) require.NoError(t, err) + err = depositStore.UpdateDeposit(ctxb, d3) + require.NoError(t, err) + err = depositStore.UpdateDeposit(ctxb, d4) + require.NoError(t, err) d1.SetState(deposit.LoopingIn) - d2.SetState(deposit.LoopedIn) + d2.SetState(deposit.HtlcTimeoutSwept) + d3.SetState(deposit.LoopedIn) + d4.SetState(deposit.Deposited) err = depositStore.UpdateDeposit(ctxb, d1) require.NoError(t, err) err = depositStore.UpdateDeposit(ctxb, d2) require.NoError(t, err) + err = depositStore.UpdateDeposit(ctxb, d3) + require.NoError(t, err) + err = depositStore.UpdateDeposit(ctxb, d4) + require.NoError(t, err) _, clientPubKey := test.CreateKey(1) _, serverPubKey := test.CreateKey(2) @@ -108,13 +148,30 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { err = swapStore.CreateLoopIn(ctxb, &swapPending) require.NoError(t, err) + // Create htlc-timeout-swept swap. HtlcTimeoutSwept is the first final + // state, so this exercises the state-list query boundary. + swapHashTimeoutSwept := lntypes.Hash{0x4, 0x2, 0x3, 0x5} + swapTimeoutSwept := StaticAddressLoopIn{ + SwapHash: swapHashTimeoutSwept, + SwapPreimage: lntypes.Preimage{0x4, 0x2, 0x3, 0x5}, + DepositOutpoints: []string{d2.OutPoint.String()}, + Deposits: []*deposit.Deposit{d2}, + ClientPubkey: clientPubKey, + ServerPubkey: serverPubKey, + HtlcTimeoutSweepAddress: addr, + } + swapTimeoutSwept.SetState(HtlcTimeoutSwept) + + err = swapStore.CreateLoopIn(ctxb, &swapTimeoutSwept) + require.NoError(t, err) + // Create succeeded swap. swapHashSucceeded := lntypes.Hash{0x2, 0x2, 0x3, 0x5} swapSucceeded := StaticAddressLoopIn{ SwapHash: swapHashSucceeded, SwapPreimage: lntypes.Preimage{0x2, 0x2, 0x3, 0x5}, - DepositOutpoints: []string{d2.OutPoint.String()}, - Deposits: []*deposit.Deposit{d2}, + DepositOutpoints: []string{d3.OutPoint.String()}, + Deposits: []*deposit.Deposit{d3}, ClientPubkey: clientPubKey, ServerPubkey: serverPubKey, HtlcTimeoutSweepAddress: addr, @@ -124,6 +181,23 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { err = swapStore.CreateLoopIn(ctxb, &swapSucceeded) require.NoError(t, err) + // Create failed swap. Failed is the last final state, so this + // exercises the state-list query boundary. + swapHashFailed := lntypes.Hash{0x3, 0x2, 0x3, 0x5} + swapFailed := StaticAddressLoopIn{ + SwapHash: swapHashFailed, + SwapPreimage: lntypes.Preimage{0x3, 0x2, 0x3, 0x5}, + DepositOutpoints: []string{d4.OutPoint.String()}, + Deposits: []*deposit.Deposit{d4}, + ClientPubkey: clientPubKey, + ServerPubkey: serverPubKey, + HtlcTimeoutSweepAddress: addr, + } + swapFailed.SetState(Failed) + + err = swapStore.CreateLoopIn(ctxb, &swapFailed) + require.NoError(t, err) + pendingSwaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(ctxb, PendingStates) require.NoError(t, err) @@ -142,16 +216,33 @@ func TestGetStaticAddressLoopInSwapsByStates(t *testing.T) { finalizedSwaps, err := swapStore.GetStaticAddressLoopInSwapsByStates(ctxb, FinalStates) require.NoError(t, err) - require.Len(t, finalizedSwaps, 1) - require.Equal(t, swapHashSucceeded, finalizedSwaps[0].SwapHash) - require.Equal(t, []string{d2.OutPoint.String()}, finalizedSwaps[0].DepositOutpoints) - require.Equal(t, Succeeded, finalizedSwaps[0].GetState()) + require.Len(t, finalizedSwaps, 3) + finalizedByState := make(map[string]*StaticAddressLoopIn) + for _, swap := range finalizedSwaps { + finalizedByState[string(swap.GetState())] = swap + } - finalizedDeposits := finalizedSwaps[0].Deposits + timeoutSweptSwap := finalizedByState[string(HtlcTimeoutSwept)] + require.NotNil(t, timeoutSweptSwap) + require.Equal(t, swapHashTimeoutSwept, timeoutSweptSwap.SwapHash) + require.Equal(t, HtlcTimeoutSwept, timeoutSweptSwap.GetState()) + + succeededSwap := finalizedByState[string(Succeeded)] + require.NotNil(t, succeededSwap) + require.Equal(t, swapHashSucceeded, succeededSwap.SwapHash) + require.Equal(t, []string{d3.OutPoint.String()}, succeededSwap.DepositOutpoints) + require.Equal(t, Succeeded, succeededSwap.GetState()) + + failedSwap := finalizedByState[string(Failed)] + require.NotNil(t, failedSwap) + require.Equal(t, swapHashFailed, failedSwap.SwapHash) + require.Equal(t, Failed, failedSwap.GetState()) + + finalizedDeposits := succeededSwap.Deposits require.Len(t, finalizedDeposits, 1) - require.Equal(t, d2.ID, finalizedDeposits[0].ID) - require.Equal(t, d2.OutPoint, finalizedDeposits[0].OutPoint) - require.Equal(t, d2.Value, finalizedDeposits[0].Value) + require.Equal(t, d3.ID, finalizedDeposits[0].ID) + require.Equal(t, d3.OutPoint, finalizedDeposits[0].OutPoint) + require.Equal(t, d3.Value, finalizedDeposits[0].Value) require.Equal(t, deposit.LoopedIn, finalizedDeposits[0].GetState()) } From d0c613e5c9c05a3a7039d024d1905bd12ff4d8cd Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Fri, 19 Jun 2026 14:48:02 +0200 Subject: [PATCH 4/5] notifications: drop best-effort messages for slow subscribers A slow optional notification subscriber must not stall the manager lock, while recovery and sweep work requests still need reliable local delivery to keep daemon progress deterministic. Treating all subscribers as blocking makes best-effort fanout a backpressure source for unrelated required work. Make reservation fanout nonblocking for slow subscribers, keep recovery and sweep notifications on a required cancellation-aware delivery path, and cover both slow best-effort subscribers and required subscribers in manager tests. --- notifications/manager.go | 18 +++++- notifications/manager_test.go | 113 +++++++++++++++++++++++++++++++++- 2 files changed, 127 insertions(+), 4 deletions(-) diff --git a/notifications/manager.go b/notifications/manager.go index 3eda5d65..b87052b3 100644 --- a/notifications/manager.go +++ b/notifications/manager.go @@ -332,7 +332,13 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerReservationNotification) - recvChan <- reservationNtfn + select { + case recvChan <- reservationNtfn: + case <-sub.subCtx.Done(): + default: + log.Debugf("Dropping reservation " + + "notification for slow subscriber") + } } case *swapserverrpc.SubscribeNotificationsResponse_StaticLoopInSweep: // nolint: lll // We'll forward the static loop in sweep request to all @@ -345,7 +351,10 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerStaticLoopInSweepNotification) - recvChan <- staticLoopInSweepRequestNtfn + select { + case recvChan <- staticLoopInSweepRequestNtfn: + case <-sub.subCtx.Done(): + } } case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll @@ -359,7 +368,10 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerUnfinishedSwapNotification) - recvChan <- unfinishedSwapNtfn + select { + case recvChan <- unfinishedSwapNtfn: + case <-sub.subCtx.Done(): + } } case *swapserverrpc.SubscribeNotificationsResponse_HtlcConfirmed: diff --git a/notifications/manager_test.go b/notifications/manager_test.go index 44300d7a..fdddef56 100644 --- a/notifications/manager_test.go +++ b/notifications/manager_test.go @@ -20,7 +20,7 @@ import ( var ( testReservationId = []byte{0x01, 0x02} - testReservationId2 = []byte{0x01, 0x02} + testReservationId2 = []byte{0x03, 0x04} ) // mockNotificationsClient implements the NotificationsClient interface for testing. @@ -190,6 +190,117 @@ func getTestNotification(resId []byte) *swapserverrpc.SubscribeNotificationsResp } } +// unfinishedSwapNotification builds an unfinished swap notification. +func unfinishedSwapNotification( + swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse { + + return &swapserverrpc.SubscribeNotificationsResponse{ + Notification: &swapserverrpc. + SubscribeNotificationsResponse_UnfinishedSwap{ + UnfinishedSwap: &swapserverrpc. + ServerUnfinishedSwapNotification{ + SwapHash: swapHash[:], + }, + }, + } +} + +// TestManager_SlowSubscriberDoesNotBlock tests that a subscriber with a full +// notification channel does not block delivery to other subscribers. +func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) { + t.Parallel() + + mgr := NewManager(&Config{}) + + slowCtx, slowCancel := context.WithCancel(t.Context()) + defer slowCancel() + slowChan := mgr.SubscribeReservations(slowCtx) + + fastCtx, fastCancel := context.WithCancel(t.Context()) + defer fastCancel() + fastChan := mgr.SubscribeReservations(fastCtx) + + firstNotif := getTestNotification(testReservationId) + mgr.handleNotification(firstNotif) + + received := <-fastChan + require.Equal(t, testReservationId, received.ReservationId) + + secondNotif := getTestNotification(testReservationId2) + done := make(chan struct{}) + go func() { + mgr.handleNotification(secondNotif) + close(done) + }() + + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + + select { + case received = <-fastChan: + require.Equal(t, testReservationId2, received.ReservationId) + + case <-time.After(time.Second): + t.Fatal("fast subscriber did not receive notification") + } + + require.Len(t, slowChan, 1) +} + +// TestManager_UnfinishedSwapNotificationWaitsForSubscriber verifies that +// unfinished swap recovery notifications are not dropped when the local +// subscriber is briefly behind. +func TestManager_UnfinishedSwapNotificationWaitsForSubscriber(t *testing.T) { + t.Parallel() + + mgr := NewManager(&Config{}) + + subCtx, subCancel := context.WithCancel(t.Context()) + defer subCancel() + + subChan := mgr.SubscribeUnfinishedSwaps(subCtx) + + swapHashA := lntypes.Hash{0x02, 0x03} + swapHashB := lntypes.Hash{0x04, 0x05} + + mgr.handleNotification(unfinishedSwapNotification(swapHashA)) + + done := make(chan struct{}) + go func() { + mgr.handleNotification(unfinishedSwapNotification(swapHashB)) + close(done) + }() + + select { + case received := <-subChan: + require.Equal(t, swapHashA[:], received.SwapHash) + + case <-time.After(time.Second): + t.Fatal("did not receive first unfinished swap notification") + } + + select { + case <-done: + + case <-time.After(time.Second): + t.Fatal("second unfinished swap notification did not unblock") + } + + select { + case received := <-subChan: + require.Equal(t, swapHashB[:], received.SwapHash) + + case <-time.After(time.Second): + t.Fatal("second unfinished swap notification was dropped") + } +} + // TestManager_Backoff verifies that repeated failures in // subscribeNotifications cause the Manager to space out subscription attempts // via a predictable incremental backoff. From 0bb06b0ba31fe73336a0956a137677b0aac4ef60 Mon Sep 17 00:00:00 2001 From: Slyghtning Date: Fri, 19 Jun 2026 14:52:17 +0200 Subject: [PATCH 5/5] notifications: queue blocking fanout Required notification fanout should not block the manager lock, but subscribers still need ordered delivery once brief backpressure clears. Sending must-deliver notifications directly can couple manager progress to subscriber receive timing, while queueing optional reservation notifications would contradict their best-effort delivery semantics. Add bounded per-subscriber queues for must-deliver notifications, let those queues own channel shutdown instead of relying on recover for closed-channel sends, keep reservation fanout best-effort, and cover queued delivery, queue cleanup, and capacity drops in manager tests. --- notifications/manager.go | 138 +++++++++++++++++++++++-- notifications/manager_test.go | 187 +++++++++++++++++++++++++++++++--- 2 files changed, 300 insertions(+), 25 deletions(-) diff --git a/notifications/manager.go b/notifications/manager.go index b87052b3..2ff6b087 100644 --- a/notifications/manager.go +++ b/notifications/manager.go @@ -45,6 +45,10 @@ const ( // htlc-confirmed subscriber before dropping the notification. htlcConfirmedSubscriberSendTimeout = 200 * time.Millisecond + // defaultMaxQueuedNotifications is the default number of notifications + // we queue per subscriber before dropping new notifications. + defaultMaxQueuedNotifications = 1024 + // current_version is the current version of the notification listener. current_version = swapserverrpc.SubscribeNotificationsRequest_V1 ) @@ -72,6 +76,10 @@ type Config struct { // MinAliveConnTime is the minimum time that the connection to the // server needs to be alive before we consider it a successful. MinAliveConnTime time.Duration + + // MaxQueuedNotifications is the maximum number of notifications that + // can wait in each subscriber's delivery queue. + MaxQueuedNotifications int } // Manager is a manager for notifications that the swap server sends to the @@ -92,6 +100,9 @@ func NewManager(cfg *Config) *Manager { if cfg.MinAliveConnTime == 0 { cfg.MinAliveConnTime = defaultMinAliveConnTime } + if cfg.MaxQueuedNotifications <= 0 { + cfg.MaxQueuedNotifications = defaultMaxQueuedNotifications + } return &Manager{ cfg: cfg, @@ -102,6 +113,113 @@ func NewManager(cfg *Config) *Manager { type subscriber struct { subCtx context.Context recvChan any + enqueue func(any) +} + +// newNotificationQueue creates a per-subscriber FIFO delivery function. +func newNotificationQueue[T any](ctx context.Context, + recvChan chan T, maxPending int) func(any) { + + type queue struct { + sync.Mutex + + pending []T + notify chan struct{} + closed bool + } + + q := &queue{ + notify: make(chan struct{}, 1), + } + + closeQueue := func() { + q.Lock() + q.closed = true + q.pending = nil + q.Unlock() + } + + go func() { + defer close(recvChan) + defer closeQueue() + + for { + select { + case <-ctx.Done(): + return + default: + } + + q.Lock() + if len(q.pending) == 0 { + q.Unlock() + + select { + case <-q.notify: + continue + + case <-ctx.Done(): + return + } + } + + ntfn := q.pending[0] + var zero T + q.pending[0] = zero + q.pending = q.pending[1:] + q.Unlock() + + select { + case recvChan <- ntfn: + case <-ctx.Done(): + return + } + } + }() + + return func(ntfn any) { + typedNtfn, ok := ntfn.(T) + if !ok { + log.Warnf("unexpected notification type %T", ntfn) + return + } + + q.Lock() + if q.closed { + q.Unlock() + return + } + if len(q.pending) >= maxPending { + q.Unlock() + log.Warnf("dropping notification for slow subscriber: "+ + "queue depth %d reached", maxPending) + return + } + + q.pending = append(q.pending, typedNtfn) + q.Unlock() + + select { + case q.notify <- struct{}{}: + default: + } + } +} + +// queueNotification queues or synchronously sends a must-deliver notification. +func queueNotification[T any](sub subscriber, recvChan chan T, ntfn T) { + if sub.enqueue != nil { + sub.enqueue(ntfn) + return + } + + log.Warnf("subscriber has no notification queue, falling back to " + + "blocking send") + + select { + case recvChan <- ntfn: + case <-sub.subCtx.Done(): + } } // SubscribeReservations subscribes to the reservation notifications. @@ -136,6 +254,9 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context, sub := subscriber{ subCtx: ctx, recvChan: notifChan, + enqueue: newNotificationQueue( + ctx, notifChan, m.cfg.MaxQueuedNotifications, + ), } m.addSubscriber(NotificationTypeStaticLoopInSweepRequest, sub) @@ -145,7 +266,6 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context, NotificationTypeStaticLoopInSweepRequest, sub, ) - close(notifChan) }) return notifChan @@ -161,12 +281,14 @@ func (m *Manager) SubscribeUnfinishedSwaps(ctx context.Context, sub := subscriber{ subCtx: ctx, recvChan: notifChan, + enqueue: newNotificationQueue( + ctx, notifChan, m.cfg.MaxQueuedNotifications, + ), } m.addSubscriber(NotificationTypeUnfinishedSwap, sub) context.AfterFunc(ctx, func() { m.removeSubscriber(NotificationTypeUnfinishedSwap, sub) - close(notifChan) }) return notifChan @@ -351,10 +473,7 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerStaticLoopInSweepNotification) - select { - case recvChan <- staticLoopInSweepRequestNtfn: - case <-sub.subCtx.Done(): - } + queueNotification(sub, recvChan, staticLoopInSweepRequestNtfn) } case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll @@ -368,10 +487,7 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc. recvChan := sub.recvChan.(chan *swapserverrpc. ServerUnfinishedSwapNotification) - select { - case recvChan <- unfinishedSwapNtfn: - case <-sub.subCtx.Done(): - } + queueNotification(sub, recvChan, unfinishedSwapNtfn) } case *swapserverrpc.SubscribeNotificationsResponse_HtlcConfirmed: @@ -415,7 +531,7 @@ func (m *Manager) removeSubscriber(notifType NotificationType, sub subscriber) { subs := m.subscribers[notifType] newSubs := make([]subscriber, 0, len(subs)) for _, s := range subs { - if s != sub { + if s.recvChan != sub.recvChan { newSubs = append(newSubs, s) } } diff --git a/notifications/manager_test.go b/notifications/manager_test.go index fdddef56..1768b004 100644 --- a/notifications/manager_test.go +++ b/notifications/manager_test.go @@ -205,9 +205,26 @@ func unfinishedSwapNotification( } } -// TestManager_SlowSubscriberDoesNotBlock tests that a subscriber with a full -// notification channel does not block delivery to other subscribers. -func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) { +// staticLoopInSweepNotification builds a static loop-in sweep notification. +func staticLoopInSweepNotification( + swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse { + + return &swapserverrpc.SubscribeNotificationsResponse{ + Notification: &swapserverrpc. + SubscribeNotificationsResponse_StaticLoopInSweep{ + StaticLoopInSweep: &swapserverrpc. + ServerStaticLoopInSweepNotification{ + SwapHash: swapHash[:], + }, + }, + } +} + +// TestManager_SlowReservationSubscriberDoesNotBlock tests that a reservation +// subscriber with a full notification channel does not block delivery to other +// subscribers. Reservation notifications are best-effort, so slow subscribers +// drop new notifications instead of queueing them. +func TestManager_SlowReservationSubscriberDoesNotBlock(t *testing.T) { t.Parallel() mgr := NewManager(&Config{}) @@ -251,6 +268,22 @@ func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) { } require.Len(t, slowChan, 1) + + select { + case received = <-slowChan: + require.Equal(t, testReservationId, received.ReservationId) + + case <-time.After(time.Second): + t.Fatal("slow subscriber did not receive first notification") + } + + select { + case received = <-slowChan: + t.Fatalf("slow subscriber received dropped notification %x", + received.ReservationId) + + case <-time.After(50 * time.Millisecond): + } } // TestManager_UnfinishedSwapNotificationWaitsForSubscriber verifies that @@ -259,45 +292,171 @@ func TestManager_SlowSubscriberDoesNotBlock(t *testing.T) { func TestManager_UnfinishedSwapNotificationWaitsForSubscriber(t *testing.T) { t.Parallel() + assertQueuedSwapHashNotifications( + t, + func(mgr *Manager, ctx context.Context) <-chan *swapserverrpc. + ServerUnfinishedSwapNotification { + + return mgr.SubscribeUnfinishedSwaps(ctx) + }, + unfinishedSwapNotification, + func(ntfn *swapserverrpc.ServerUnfinishedSwapNotification) []byte { + return ntfn.SwapHash + }, + lntypes.Hash{0x02, 0x03}, lntypes.Hash{0x04, 0x05}, + "did not receive first unfinished swap notification", + "second unfinished swap notification was dropped", + ) +} + +// TestManager_StaticLoopInSweepNotificationQueuesForSlowSubscriber verifies +// that a full static-loop-in sweep subscriber channel does not block the global +// notification receive loop. +func TestManager_StaticLoopInSweepNotificationQueuesForSlowSubscriber( + t *testing.T) { + + t.Parallel() + + assertQueuedSwapHashNotifications( + t, + func(mgr *Manager, ctx context.Context) <-chan *swapserverrpc. + ServerStaticLoopInSweepNotification { + + return mgr.SubscribeStaticLoopInSweepRequests(ctx) + }, + staticLoopInSweepNotification, + func(ntfn *swapserverrpc.ServerStaticLoopInSweepNotification) []byte { + return ntfn.SwapHash + }, + lntypes.Hash{0x12, 0x13}, lntypes.Hash{0x14, 0x15}, + "did not receive first sweep notification", + "second sweep notification was not queued", + ) +} + +// TestManager_QueuedNotificationChannelClosesOnCancel verifies that queued +// subscribers own their channel shutdown even when delivery is blocked. +func TestManager_QueuedNotificationChannelClosesOnCancel(t *testing.T) { + t.Parallel() + mgr := NewManager(&Config{}) subCtx, subCancel := context.WithCancel(t.Context()) - defer subCancel() - subChan := mgr.SubscribeUnfinishedSwaps(subCtx) - swapHashA := lntypes.Hash{0x02, 0x03} - swapHashB := lntypes.Hash{0x04, 0x05} - + swapHashA := lntypes.Hash{0x21, 0x22} mgr.handleNotification(unfinishedSwapNotification(swapHashA)) + require.Eventually(t, func() bool { + return len(subChan) == 1 + }, time.Second, 10*time.Millisecond) + + swapHashB := lntypes.Hash{0x23, 0x24} done := make(chan struct{}) go func() { mgr.handleNotification(unfinishedSwapNotification(swapHashB)) close(done) }() + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + + subCancel() + select { - case received := <-subChan: + case received, ok := <-subChan: + require.True(t, ok) require.Equal(t, swapHashA[:], received.SwapHash) case <-time.After(time.Second): - t.Fatal("did not receive first unfinished swap notification") + t.Fatal("first unfinished swap notification was not delivered") } + require.Eventually(t, func() bool { + select { + case _, ok := <-subChan: + return !ok + default: + return false + } + }, time.Second, 10*time.Millisecond) +} + +// TestNotificationQueueDropsAtCapacity checks the queue's explicit drop policy +// once a subscriber reaches its configured backlog limit. +func TestNotificationQueueDropsAtCapacity(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + + recvChan := make(chan int, 1) + enqueue := newNotificationQueue(ctx, recvChan, 0) + + enqueue(1) + select { - case <-done: + case ntfn := <-recvChan: + t.Fatalf("received dropped notification %d", ntfn) + + case <-time.After(50 * time.Millisecond): + } +} + +// assertQueuedSwapHashNotifications checks queued delivery for swap hashes. +func assertQueuedSwapHashNotifications[T any](t *testing.T, + subscribe func(*Manager, context.Context) <-chan T, + notification func(lntypes.Hash) *swapserverrpc. + SubscribeNotificationsResponse, + swapHash func(T) []byte, swapHashA, swapHashB lntypes.Hash, + firstFailureMsg, secondFailureMsg string) { + + t.Helper() + + mgr := NewManager(&Config{}) + + subCtx, subCancel := context.WithCancel(t.Context()) + defer subCancel() + + subChan := subscribe(mgr, subCtx) + + mgr.handleNotification(notification(swapHashA)) + + done := make(chan struct{}) + go func() { + mgr.handleNotification(notification(swapHashB)) + close(done) + }() + + require.Eventually(t, func() bool { + select { + case <-done: + return true + default: + return false + } + }, time.Second, 10*time.Millisecond) + + select { + case received := <-subChan: + require.Equal(t, swapHashA[:], swapHash(received)) case <-time.After(time.Second): - t.Fatal("second unfinished swap notification did not unblock") + t.Fatal(firstFailureMsg) } select { case received := <-subChan: - require.Equal(t, swapHashB[:], received.SwapHash) + require.Equal(t, swapHashB[:], swapHash(received)) case <-time.After(time.Second): - t.Fatal("second unfinished swap notification was dropped") + t.Fatal(secondFailureMsg) } }