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

notifications: persist static loop-in risk decisions

Browse source 
Persist static loop-in confirmation-risk decisions before fanout when a
persistence callback is configured. Keep unpersisted decisions cached
for replay so notification delivery is not lost if the swap row is not
available yet.
This commit is contained in:
Slyghtning 2026-07-08 13:57:15 +02:00
parent 491fddbc34
commit e53bb67e73
No known key found for this signature in database
GPG key ID: F82D456EA023C9BF
2 changed files with 351 additions and 129 deletions
Download patch file Download diff file Expand all files Collapse all files

288
notifications/manager.go
View file

@ -88,6 +88,13 @@ type Config struct {
// MaxQueuedNotifications is the maximum number of notifications that
// can wait in each subscriber's delivery queue.
MaxQueuedNotifications int
// PersistStaticLoopInRiskDecision durably records static loop-in
// confirmation-risk decisions. If this fails, the notification is still
// cached and forwarded so a later subscriber can process it after the swap
// row exists.
PersistStaticLoopInRiskDecision func(context.Context, lntypes.Hash,
bool) error
}
// Manager is a manager for notifications that the swap server sends to the
@ -99,13 +106,26 @@ type Manager struct {
hasL402 bool
// subscribers holds active notification subscribers by notification
// type. It is guarded by the Manager mutex.
subscribers map[NotificationType][]subscriber
// staticLoopInRiskAccepted caches accepted risk decisions by swap hash
// so a later matching subscriber can receive a previously delivered
// server decision.
staticLoopInRiskAccepted map[lntypes.Hash]*swapserverrpc.
ServerStaticLoopInRiskAcceptedNotification
// staticLoopInRiskRejected caches rejected risk decisions by swap hash
// so a later matching subscriber can receive a previously delivered
// server decision.
staticLoopInRiskRejected map[lntypes.Hash]*swapserverrpc.
ServerStaticLoopInRiskRejectedNotification
// staticLoopInRiskPersisted records whether the cached risk decision for
// a swap hash was durably persisted. Unpersisted decisions remain cached
// after subscriber cancellation so they can be replayed.
staticLoopInRiskPersisted map[lntypes.Hash]bool
}
// NewManager creates a new notification manager.
@ -129,14 +149,15 @@ func NewManager(cfg *Config) *Manager {
map[lntypes.Hash]*swapserverrpc.
ServerStaticLoopInRiskRejectedNotification,
),
staticLoopInRiskPersisted: make(map[lntypes.Hash]bool),
}
}
type subscriber struct {
subCtx context.Context
recvChan any
enqueue func(any)
swapHash *lntypes.Hash
enqueue func(any)
}
// newNotificationQueue creates a per-subscriber FIFO delivery function.
@ -245,6 +266,19 @@ func queueNotification[T any](sub subscriber, recvChan chan T, ntfn T) {
}
}
// dropNotification sends a best-effort notification to a subscriber.
func dropNotification[T any](sub subscriber, recvChan chan T, ntfn T,
description string) {
select {
case recvChan <- ntfn:
case <-sub.subCtx.Done():
default:
log.Debugf("Dropping %s notification for slow subscriber",
description)
}
}
// SubscribeReservations subscribes to the reservation notifications.
func (m *Manager) SubscribeReservations(ctx context.Context,
) <-chan *swapserverrpc.ServerReservationNotification {
@ -294,16 +328,11 @@ func (m *Manager) SubscribeStaticLoopInSweepRequests(ctx context.Context,
return notifChan
}
// SubscribeStaticLoopInRiskAccepted subscribes to static loop in risk accepted
// notifications.
func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context,
swapHash lntypes.Hash,
) <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification {
notifChan := make(
chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification, 1,
)
func subscribeStaticLoopInRiskDecision[T any](m *Manager, ctx context.Context,
swapHash lntypes.Hash, notifType NotificationType,
notifications map[lntypes.Hash]T) <-chan T {
notifChan := make(chan T, 1)
sub := subscriber{
subCtx: ctx,
recvChan: notifChan,
@ -311,19 +340,25 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context,
}
m.Lock()
m.subscribers[NotificationTypeStaticLoopInRiskAccepted] = append(
m.subscribers[NotificationTypeStaticLoopInRiskAccepted], sub,
)
if ntfn, ok := m.staticLoopInRiskAccepted[swapHash]; ok {
m.subscribers[notifType] = append(m.subscribers[notifType], sub)
if ntfn, ok := notifications[swapHash]; ok {
notifChan <- ntfn
delete(m.staticLoopInRiskAccepted, swapHash)
if m.staticLoopInRiskPersisted[swapHash] {
delete(notifications, swapHash)
delete(m.staticLoopInRiskPersisted, swapHash)
}
}
m.Unlock()
context.AfterFunc(ctx, func() {
m.removeSubscriber(NotificationTypeStaticLoopInRiskAccepted, sub)
m.removeSubscriber(notifType, sub)
m.Lock()
delete(m.staticLoopInRiskAccepted, swapHash)
if _, ok := notifications[swapHash]; ok &&
m.staticLoopInRiskPersisted[swapHash] {
delete(notifications, swapHash)
delete(m.staticLoopInRiskPersisted, swapHash)
}
m.Unlock()
close(notifChan)
})
@ -331,41 +366,28 @@ func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context,
return notifChan
}
// SubscribeStaticLoopInRiskAccepted subscribes to static loop in risk accepted
// notifications.
func (m *Manager) SubscribeStaticLoopInRiskAccepted(ctx context.Context,
swapHash lntypes.Hash,
) <-chan *swapserverrpc.ServerStaticLoopInRiskAcceptedNotification {
return subscribeStaticLoopInRiskDecision(
m, ctx, swapHash, NotificationTypeStaticLoopInRiskAccepted,
m.staticLoopInRiskAccepted,
)
}
// SubscribeStaticLoopInRiskRejected subscribes to static loop in risk rejected
// notifications.
func (m *Manager) SubscribeStaticLoopInRiskRejected(ctx context.Context,
swapHash lntypes.Hash,
) <-chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification {
notifChan := make(
chan *swapserverrpc.ServerStaticLoopInRiskRejectedNotification, 1,
return subscribeStaticLoopInRiskDecision(
m, ctx, swapHash, NotificationTypeStaticLoopInRiskRejected,
m.staticLoopInRiskRejected,
)
sub := subscriber{
subCtx: ctx,
recvChan: notifChan,
swapHash: &swapHash,
}
m.Lock()
m.subscribers[NotificationTypeStaticLoopInRiskRejected] = append(
m.subscribers[NotificationTypeStaticLoopInRiskRejected], sub,
)
if ntfn, ok := m.staticLoopInRiskRejected[swapHash]; ok {
notifChan <- ntfn
delete(m.staticLoopInRiskRejected, swapHash)
}
m.Unlock()
context.AfterFunc(ctx, func() {
m.removeSubscriber(NotificationTypeStaticLoopInRiskRejected, sub)
m.Lock()
delete(m.staticLoopInRiskRejected, swapHash)
m.Unlock()
close(notifChan)
})
return notifChan
}
// SubscribeUnfinishedSwaps subscribes to the unfinished swap notifications.
@ -525,7 +547,7 @@ func (m *Manager) subscribeNotifications(ctx context.Context) error {
notification, err := notifStream.Recv()
if err == nil && notification != nil {
log.Tracef("Received notification: %v", notification)
m.handleNotification(notification)
m.handleNotification(ctx, notification)
continue
}
@ -535,9 +557,73 @@ func (m *Manager) subscribeNotifications(ctx context.Context) error {
}
}
// staticLoopInRiskDecisionName returns the log label for a risk decision.
func staticLoopInRiskDecisionName(accepted bool) string {
if accepted {
return "accepted"
}
return "rejected"
}
// handleStaticLoopInRiskDecision persists, caches, and forwards a risk
// decision notification to the matching subscriber.
func (m *Manager) handleStaticLoopInRiskDecision(ctx context.Context,
swapHashBytes []byte, accepted bool, notifType NotificationType,
cacheDecision func(lntypes.Hash, bool),
notifySubscriber func(subscriber)) {
decision := staticLoopInRiskDecisionName(accepted)
persisted := m.cfg.PersistStaticLoopInRiskDecision == nil
var (
swapHash lntypes.Hash
hasSwapHash bool
)
if swapHashBytes != nil {
hash, err := lntypes.MakeHash(swapHashBytes)
if err != nil {
log.Warnf("Received invalid static loop in risk "+
"%s notification: %v", decision, err)
} else {
swapHash = hash
hasSwapHash = true
}
}
if hasSwapHash && m.cfg.PersistStaticLoopInRiskDecision != nil {
err := m.cfg.PersistStaticLoopInRiskDecision(
ctx, swapHash, accepted,
)
if err != nil {
log.Errorf("Unable to persist static loop in risk "+
"%s notification: %v", decision, err)
} else {
persisted = true
}
}
m.Lock()
defer m.Unlock()
if hasSwapHash {
cacheDecision(swapHash, persisted)
}
for _, sub := range m.subscribers[notifType] {
if !hasSwapHash || sub.swapHash == nil ||
*sub.swapHash != swapHash {
continue
}
notifySubscriber(sub)
}
}
// handleNotification handles an incoming notification from the server,
// forwarding it to the appropriate subscribers.
func (m *Manager) handleNotification(ntfn *swapserverrpc.
func (m *Manager) handleNotification(ctx context.Context, ntfn *swapserverrpc.
SubscribeNotificationsResponse) {
switch ntfn.Notification.(type) {
@ -577,89 +663,57 @@ func (m *Manager) handleNotification(ntfn *swapserverrpc.
// We'll forward the static loop in risk accepted notification to the
// subscriber for the matching swap.
riskAcceptedNtfn := ntfn.GetStaticLoopInRiskAccepted()
m.Lock()
defer m.Unlock()
var (
swapHash lntypes.Hash
hasSwapHash bool
)
var swapHashBytes []byte
if riskAcceptedNtfn != nil {
hash, err := lntypes.MakeHash(riskAcceptedNtfn.SwapHash)
if err != nil {
log.Warnf("Received invalid static loop in risk "+
"accepted notification: %v", err)
} else {
swapHash = hash
hasSwapHash = true
m.staticLoopInRiskAccepted[hash] =
swapHashBytes = riskAcceptedNtfn.SwapHash
}
m.handleStaticLoopInRiskDecision(
ctx, swapHashBytes, true,
NotificationTypeStaticLoopInRiskAccepted,
func(swapHash lntypes.Hash, persisted bool) {
m.staticLoopInRiskAccepted[swapHash] =
riskAcceptedNtfn
delete(m.staticLoopInRiskRejected, hash)
}
}
for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskAccepted] { // nolint: lll
if !hasSwapHash || sub.swapHash == nil ||
*sub.swapHash != swapHash {
continue
}
m.staticLoopInRiskPersisted[swapHash] = persisted
delete(m.staticLoopInRiskRejected, swapHash)
},
func(sub subscriber) {
recvChan := sub.recvChan.(chan *swapserverrpc.
ServerStaticLoopInRiskAcceptedNotification)
select {
case recvChan <- riskAcceptedNtfn:
case <-sub.subCtx.Done():
default:
log.Debugf("Dropping static loop in risk " +
"accepted notification for slow subscriber")
}
}
dropNotification(
sub, recvChan, riskAcceptedNtfn,
"static loop in risk accepted",
)
},
)
case *swapserverrpc.SubscribeNotificationsResponse_StaticLoopInRiskRejected: // nolint: lll
// We'll forward the static loop in risk rejected notification to the
// subscriber for the matching swap.
riskRejectedNtfn := ntfn.GetStaticLoopInRiskRejected()
m.Lock()
defer m.Unlock()
var (
swapHash lntypes.Hash
hasSwapHash bool
)
var swapHashBytes []byte
if riskRejectedNtfn != nil {
hash, err := lntypes.MakeHash(riskRejectedNtfn.SwapHash)
if err != nil {
log.Warnf("Received invalid static loop in risk "+
"rejected notification: %v", err)
} else {
swapHash = hash
hasSwapHash = true
m.staticLoopInRiskRejected[hash] =
swapHashBytes = riskRejectedNtfn.SwapHash
}
m.handleStaticLoopInRiskDecision(
ctx, swapHashBytes, false,
NotificationTypeStaticLoopInRiskRejected,
func(swapHash lntypes.Hash, persisted bool) {
m.staticLoopInRiskRejected[swapHash] =
riskRejectedNtfn
delete(m.staticLoopInRiskAccepted, hash)
}
}
for _, sub := range m.subscribers[NotificationTypeStaticLoopInRiskRejected] { // nolint: lll
if !hasSwapHash || sub.swapHash == nil ||
*sub.swapHash != swapHash {
continue
}
m.staticLoopInRiskPersisted[swapHash] = persisted
delete(m.staticLoopInRiskAccepted, swapHash)
},
func(sub subscriber) {
recvChan := sub.recvChan.(chan *swapserverrpc.
ServerStaticLoopInRiskRejectedNotification)
select {
case recvChan <- riskRejectedNtfn:
case <-sub.subCtx.Done():
default:
log.Debugf("Dropping static loop in risk " +
"rejected notification for slow subscriber")
}
}
dropNotification(
sub, recvChan, riskRejectedNtfn,
"static loop in risk rejected",
)
},
)
case *swapserverrpc.SubscribeNotificationsResponse_UnfinishedSwap: // nolint: lll
// We'll forward the unfinished swap notification to all

184
notifications/manager_test.go
View file

@ -220,6 +220,7 @@ func staticLoopInSweepNotification(
}
}
// staticLoopInRiskAcceptedNotification builds a risk accepted notification.
func staticLoopInRiskAcceptedNotification(
swapHash lntypes.Hash) *swapserverrpc.SubscribeNotificationsResponse {
@ -271,7 +272,7 @@ func assertStaticLoopInRiskNotificationSwapScoped[
subChanA := subscribe(mgr, subCtx, swapHashA)
subChanB := subscribe(mgr, subCtx, swapHashB)
mgr.handleNotification(notification(swapHashA))
mgr.handleNotification(t.Context(), notification(swapHashA))
select {
case received := <-subChanA:
@ -290,7 +291,7 @@ func assertStaticLoopInRiskNotificationSwapScoped[
default:
}
mgr.handleNotification(notification(swapHashB))
mgr.handleNotification(t.Context(), notification(swapHashB))
select {
case received := <-subChanB:
@ -320,7 +321,7 @@ func TestManager_SlowReservationSubscriberDoesNotBlock(t *testing.T) {
fastChan := mgr.SubscribeReservations(fastCtx)
firstNotif := getTestNotification(testReservationId)
mgr.handleNotification(firstNotif)
mgr.handleNotification(t.Context(), firstNotif)
received := <-fastChan
require.Equal(t, testReservationId, received.ReservationId)
@ -328,7 +329,7 @@ func TestManager_SlowReservationSubscriberDoesNotBlock(t *testing.T) {
secondNotif := getTestNotification(testReservationId2)
done := make(chan struct{})
go func() {
mgr.handleNotification(secondNotif)
mgr.handleNotification(t.Context(), secondNotif)
close(done)
}()
@ -427,7 +428,7 @@ func TestManager_QueuedNotificationChannelClosesOnCancel(t *testing.T) {
subChan := mgr.SubscribeUnfinishedSwaps(subCtx)
swapHashA := lntypes.Hash{0x21, 0x22}
mgr.handleNotification(unfinishedSwapNotification(swapHashA))
mgr.handleNotification(t.Context(), unfinishedSwapNotification(swapHashA))
require.Eventually(t, func() bool {
return len(subChan) == 1
@ -436,7 +437,7 @@ func TestManager_QueuedNotificationChannelClosesOnCancel(t *testing.T) {
swapHashB := lntypes.Hash{0x23, 0x24}
done := make(chan struct{})
go func() {
mgr.handleNotification(unfinishedSwapNotification(swapHashB))
mgr.handleNotification(t.Context(), unfinishedSwapNotification(swapHashB))
close(done)
}()
@ -508,11 +509,11 @@ func assertQueuedSwapHashNotifications[T any](t *testing.T,
subChan := subscribe(mgr, subCtx)
mgr.handleNotification(notification(swapHashA))
mgr.handleNotification(t.Context(), notification(swapHashA))
done := make(chan struct{})
go func() {
mgr.handleNotification(notification(swapHashB))
mgr.handleNotification(t.Context(), notification(swapHashB))
close(done)
}()
@ -557,6 +558,7 @@ func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) {
subChan := mgr.SubscribeStaticLoopInRiskAccepted(subCtx, swapHash)
mgr.handleNotification(
t.Context(),
&swapserverrpc.SubscribeNotificationsResponse{
Notification: &swapserverrpc.
SubscribeNotificationsResponse_StaticLoopInRiskAccepted{
@ -577,6 +579,169 @@ func TestManager_StaticLoopInRiskAcceptedNotification(t *testing.T) {
}
}
// TestManager_StaticLoopInRiskDecisionPersists verifies that risk decisions are
// handed to the durable callback before they are treated as delivered.
func TestManager_StaticLoopInRiskDecisionPersists(t *testing.T) {
t.Parallel()
type persistedDecision struct {
swapHash lntypes.Hash
accepted bool
}
persisted := make(chan persistedDecision, 2)
mgr := NewManager(&Config{
PersistStaticLoopInRiskDecision: func(_ context.Context,
swapHash lntypes.Hash, accepted bool) error {
persisted <- persistedDecision{
swapHash: swapHash,
accepted: accepted,
}
return nil
},
})
acceptedHash := lntypes.Hash{0x16, 0x17}
rejectedHash := lntypes.Hash{0x18, 0x19}
mgr.handleNotification(
t.Context(), staticLoopInRiskAcceptedNotification(acceptedHash),
)
mgr.handleNotification(
t.Context(), staticLoopInRiskRejectedNotification(rejectedHash),
)
select {
case decision := <-persisted:
require.Equal(t, acceptedHash, decision.swapHash)
require.True(t, decision.accepted)
case <-time.After(time.Second):
t.Fatal("accepted risk decision was not persisted")
}
select {
case decision := <-persisted:
require.Equal(t, rejectedHash, decision.swapHash)
require.False(t, decision.accepted)
case <-time.After(time.Second):
t.Fatal("rejected risk decision was not persisted")
}
}
// TestManager_StaticLoopInRiskDecisionReplayOnPersistFailure verifies that an
// early risk notification is still cached if the swap row does not exist yet.
func TestManager_StaticLoopInRiskDecisionReplayOnPersistFailure(t *testing.T) {
t.Parallel()
swapHash := lntypes.Hash{0x1a, 0x1b}
mgr := NewManager(&Config{
PersistStaticLoopInRiskDecision: func(_ context.Context,
_ lntypes.Hash, _ bool) error {
return errors.New("swap not stored yet")
},
})
mgr.handleNotification(
t.Context(), staticLoopInRiskAcceptedNotification(swapHash),
)
subCtx, subCancel := context.WithCancel(t.Context())
defer subCancel()
subChan := mgr.SubscribeStaticLoopInRiskAccepted(subCtx, swapHash)
select {
case received := <-subChan:
require.Equal(t, swapHash[:], received.SwapHash)
case <-time.After(time.Second):
t.Fatal("did not replay risk notification after persist failure")
}
}
// TestManager_StaticLoopInRiskDecisionReplaysAfterSubscriberCancel verifies that
// a non-persisted risk decision remains replayable if the subscriber is canceled
// before the FSM has a chance to process it.
func TestManager_StaticLoopInRiskDecisionReplaysAfterSubscriberCancel(
t *testing.T) {
t.Parallel()
assertStaticLoopInRiskDecisionReplaysAfterSubscriberCancel(
t,
(*Manager).SubscribeStaticLoopInRiskAccepted,
staticLoopInRiskAcceptedNotification,
)
assertStaticLoopInRiskDecisionReplaysAfterSubscriberCancel(
t,
(*Manager).SubscribeStaticLoopInRiskRejected,
staticLoopInRiskRejectedNotification,
)
}
func assertStaticLoopInRiskDecisionReplaysAfterSubscriberCancel[
T staticLoopInRiskNotification](t *testing.T,
subscribe func(*Manager, context.Context, lntypes.Hash) <-chan T,
notification func(lntypes.Hash) *swapserverrpc.
SubscribeNotificationsResponse) {
t.Helper()
swapHash := lntypes.Hash{0x2a, 0x2b}
mgr := NewManager(&Config{
PersistStaticLoopInRiskDecision: func(_ context.Context,
_ lntypes.Hash, _ bool) error {
return errors.New("swap not stored yet")
},
})
subCtx, subCancel := context.WithCancel(t.Context())
subChan := subscribe(mgr, subCtx, swapHash)
mgr.handleNotification(t.Context(), notification(swapHash))
require.Eventually(t, func() bool {
return len(subChan) == 1
}, time.Second, 10*time.Millisecond)
subCancel()
select {
case <-subChan:
case <-time.After(time.Second):
t.Fatal("risk decision notification was not delivered before " +
"cancel")
}
select {
case _, ok := <-subChan:
require.False(t, ok)
case <-time.After(time.Second):
t.Fatal("risk decision subscription did not close after cancel")
}
replayCtx, replayCancel := context.WithCancel(t.Context())
defer replayCancel()
replayChan := subscribe(mgr, replayCtx, swapHash)
select {
case received := <-replayChan:
require.Equal(t, swapHash[:], received.GetSwapHash())
case <-time.After(time.Second):
t.Fatal("cached risk decision was lost after subscriber " +
"cancellation")
}
}
// TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped verifies that a
// notification for one swap does not occupy another swap's subscriber channel.
func TestManager_StaticLoopInRiskAcceptedNotificationSwapScoped(t *testing.T) {
@ -603,6 +768,7 @@ func TestManager_StaticLoopInRiskAcceptedNotificationReplay(t *testing.T) {
swapHash := lntypes.Hash{0x06, 0x07}
mgr.handleNotification(
t.Context(),
&swapserverrpc.SubscribeNotificationsResponse{
Notification: &swapserverrpc.
SubscribeNotificationsResponse_StaticLoopInRiskAccepted{
@ -643,6 +809,7 @@ func TestManager_StaticLoopInRiskRejectedNotification(t *testing.T) {
subChan := mgr.SubscribeStaticLoopInRiskRejected(subCtx, swapHash)
mgr.handleNotification(
t.Context(),
&swapserverrpc.SubscribeNotificationsResponse{
Notification: &swapserverrpc.
SubscribeNotificationsResponse_StaticLoopInRiskRejected{
@ -689,6 +856,7 @@ func TestManager_StaticLoopInRiskRejectedNotificationReplay(t *testing.T) {
swapHash := lntypes.Hash{0x0a, 0x0b}
mgr.handleNotification(
t.Context(),
&swapserverrpc.SubscribeNotificationsResponse{
Notification: &swapserverrpc.
SubscribeNotificationsResponse_StaticLoopInRiskRejected{