staticaddr/loopin: publish status updates after persistence

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,

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()
+	}
+}

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.

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

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

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

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

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.