multi: add garbage collection to payments

config.go

@@ -463,6 +463,10 @@ type Config struct {
 
 	GcCanceledInvoicesOnTheFly bool `long:"gc-canceled-invoices-on-the-fly" description:"If true, we'll delete newly canceled invoices on the fly."`
 
+	GcFailedPaymentsOnStartup bool `long:"gc-failed-payments-on-startup" descrition:"If true, we'll attempt to garbage collect failed payments upon start."`
+
+	GcFailedPaymentsOnTheFly bool `long:"gc-failed-payments-on-the-fly" description:"If true, we'll delete newly failed payments on the fly."`
+
 	DustThreshold uint64 `long:"dust-threshold" description:"DEPRECATED: Sets the max fee exposure in satoshis for a channel after which HTLC's will be failed." hidden:"true"`
 
 	MaxFeeExposure uint64 `long:"channel-max-fee-exposure" description:" Limits the maximum fee exposure in satoshis of a channel. This value is enforced for all channels and is independent of the channel initiator."`

routing/control_tower.go

@@ -77,6 +77,11 @@
 	// update with the current state of every inflight payment is always
 	// sent out immediately.
 	SubscribeAllPayments() (ControlTowerSubscriber, error)
+
+	// DeleteFailedPayments deletes all failed payments from the db. We call
+	// this method on startup to clean up the db of any failed payments if
+	// the user has set the GcFailedPaymentsOnStartup flag.
+	DeleteFailedPayments() error
 }
 
 // ControlTowerSubscriber contains the state for a payment update subscriber.
@@ -143,17 +148,26 @@
 	// that no race conditions occur in between updating the database and
 	// sending a notification.
 	paymentsMtx *multimutex.Mutex[lntypes.Hash]
+
+	// gcFailedPaymentsOnTheFly determines if failed payments should be
+	// garbage collected immediately upon failure.
+	gcFailedPaymentsOnTheFly bool
 }
 
 // NewControlTower creates a new instance of the controlTower.
-func NewControlTower(db paymentsdb.DB) ControlTower {
+func NewControlTower(db paymentsdb.DB,
+	gcFailedPaymentsOnTheFly bool) ControlTower {
+
 	return &controlTower{
 		db: db,
 		subscribersAllPayments: make(
 			map[uint64]*controlTowerSubscriberImpl,
 		),
-		subscribers: make(map[lntypes.Hash][]*controlTowerSubscriberImpl),
-		paymentsMtx: multimutex.NewMutex[lntypes.Hash](),
+		subscribers: make(
+			map[lntypes.Hash][]*controlTowerSubscriberImpl,
+		),
+		paymentsMtx:              multimutex.NewMutex[lntypes.Hash](),
+		gcFailedPaymentsOnTheFly: gcFailedPaymentsOnTheFly,
 	}
 }
 
@@ -277,6 +291,17 @@
 	// Notify subscribers of fail event.
 	p.notifySubscribers(paymentHash, payment)
 
+	// If the garbage collection flag is set, we'll delete the failed
+	// payment on the fly. We do this after failing the payment to make
+	// sure the payment and attempt are both marked as failed.
+	if p.gcFailedPaymentsOnTheFly {
+		const failedHtlcsOnly = false
+		err := p.db.DeletePayment(paymentHash, failedHtlcsOnly)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -423,3 +448,13 @@
 		}
 	}
 }
+
+// DeleteFailedPayments deletes all failed payments from the db. We explicitly
+// set failedOnly to true to delete the payment and all its attempts.
+func (p *controlTower) DeleteFailedPayments() error {
+	const failedOnly = true
+	const failedHtlcsOnly = false
+	_, err := p.db.DeletePayments(failedOnly, failedHtlcsOnly)
+
+	return err
+}

routing/control_tower_test.go

@@ -56,7 +56,7 @@ func TestControlTowerSubscribeUnknown(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	pControl := NewControlTower(paymentDB)
+	pControl := NewControlTower(paymentDB, false)
 
 	// Subscription should fail when the payment is not known.
 	_, err = pControl.SubscribePayment(lntypes.Hash{1})
@@ -73,7 +73,7 @@ func TestControlTowerSubscribeSuccess(t *testing.T) {
 	paymentDB, err := paymentsdb.NewKVStore(db)
 	require.NoError(t, err)
 
-	pControl := NewControlTower(paymentDB)
+	pControl := NewControlTower(paymentDB, false)
 
 	// Initiate a payment.
 	info, attempt, preimg, err := genInfo()
@@ -206,7 +206,7 @@ func TestKVStoreSubscribeAllSuccess(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	pControl := NewControlTower(paymentDB)
+	pControl := NewControlTower(paymentDB, false)
 
 	// Initiate a payment.
 	info1, attempt1, preimg1, err := genInfo()
@@ -331,7 +331,7 @@ func TestKVStoreSubscribeAllImmediate(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	pControl := NewControlTower(paymentDB)
+	pControl := NewControlTower(paymentDB, false)
 
 	// Initiate a payment.
 	info, attempt, _, err := genInfo()
@@ -380,7 +380,7 @@ func TestKVStoreUnsubscribeSuccess(t *testing.T) {
 	)
 	require.NoError(t, err)
 
-	pControl := NewControlTower(paymentDB)
+	pControl := NewControlTower(paymentDB, false)
 
 	subscription1, err := pControl.SubscribeAllPayments()
 	require.NoError(t, err, "expected subscribe to succeed, but got: %v")
@@ -457,7 +457,7 @@ func testKVStoreSubscribeFail(t *testing.T, registerAttempt,
 	)
 	require.NoError(t, err)
 
-	pControl := NewControlTower(paymentDB)
+	pControl := NewControlTower(paymentDB, false)
 
 	// Initiate a payment.
 	info, attempt, _, err := genInfo()

routing/mock_test.go

@@ -588,6 +588,10 @@ func (m *mockControlTowerOld) SubscribeAllPayments() (
 	return nil, errors.New("not implemented")
 }
 
+func (m *mockControlTowerOld) DeleteFailedPayments() error {
+	return nil
+}
+
 type mockPaymentAttemptDispatcher struct {
 	mock.Mock
 }
@@ -821,6 +825,11 @@ func (m *mockControlTower) SubscribeAllPayments() (
 	return args.Get(0).(ControlTowerSubscriber), args.Error(1)
 }
 
+func (m *mockControlTower) DeleteFailedPayments() error {
+	args := m.Called()
+	return args.Error(0)
+}
+
 type mockMPPayment struct {
 	mock.Mock
 }

routing/router.go

@@ -295,6 +295,10 @@ type Config struct {
 	// TrafficShaper is an optional traffic shaper that can be used to
 	// control the outgoing channel of a payment.
 	TrafficShaper fn.Option[htlcswitch.AuxTrafficShaper]
+
+	// GcFailedPaymentsOnStartup is a flag that indicates whether to
+	// garbage collect failed payments on startup.
+	GcFailedPaymentsOnStartup bool
 }
 
 // EdgeLocator is a struct used to identify a specific edge.
@@ -354,6 +358,15 @@ func (r *ChannelRouter) Start() error {
 
 	log.Info("Channel Router starting")
 
+	// If the garbage collection flag is set, we'll delete the failed
+	// payments on startup.
+	if r.cfg.GcFailedPaymentsOnStartup {
+		if err := r.cfg.Control.DeleteFailedPayments(); err != nil {
+			log.Error("Failed to delete failed payments on startup")
+			return err
+		}
+	}
+
 	// If any payments are still in flight, we resume, to make sure their
 	// results are properly handled.
 	if err := r.resumePayments(); err != nil {

sample-lnd.conf

@@ -548,6 +548,12 @@
 ; If true, we'll delete newly canceled invoices on the fly.
 ; gc-canceled-invoices-on-the-fly=false
 
+; If true, we'll attempt to garbage collect failed payments upon start.
+; gc-failed-payments-on-startup=false
+
+; If true, we'll delete newly failed payments on the fly.
+; gc-failed-payments-on-the-fly=false
+
 ; If true, our node will allow htlc forwards that arrive and depart on the same
 ; channel.
 ; allow-circular-route=false

server.go

@@ -1005,7 +1005,9 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
 		PathFindingConfig:   pathFindingConfig,
 	}
 
-	s.controlTower = routing.NewControlTower(dbs.PaymentsDB)
+	s.controlTower = routing.NewControlTower(
+		dbs.PaymentsDB, cfg.GcFailedPaymentsOnTheFly,
+	)
 
 	strictPruning := cfg.Bitcoin.Node == "neutrino" ||
 		cfg.Routing.StrictZombiePruning
@@ -1028,20 +1030,21 @@ func newServer(ctx context.Context, cfg *Config, listenAddrs []net.Addr,
 	}
 
 	s.chanRouter, err = routing.New(routing.Config{
-		SelfNode:           nodePubKey,
-		RoutingGraph:       dbs.GraphDB,
-		Chain:              cc.ChainIO,
-		Payer:              s.htlcSwitch,
-		Control:            s.controlTower,
-		MissionControl:     s.defaultMC,
-		SessionSource:      paymentSessionSource,
-		GetLink:            s.htlcSwitch.GetLinkByShortID,
-		NextPaymentID:      sequencer.NextID,
-		PathFindingConfig:  pathFindingConfig,
-		Clock:              clock.NewDefaultClock(),
-		ApplyChannelUpdate: s.graphBuilder.ApplyChannelUpdate,
-		ClosedSCIDs:        s.fetchClosedChannelSCIDs(),
-		TrafficShaper:      implCfg.TrafficShaper,
+		SelfNode:                  nodePubKey,
+		RoutingGraph:              dbs.GraphDB,
+		Chain:                     cc.ChainIO,
+		Payer:                     s.htlcSwitch,
+		Control:                   s.controlTower,
+		MissionControl:            s.defaultMC,
+		SessionSource:             paymentSessionSource,
+		GetLink:                   s.htlcSwitch.GetLinkByShortID,
+		NextPaymentID:             sequencer.NextID,
+		PathFindingConfig:         pathFindingConfig,
+		Clock:                     clock.NewDefaultClock(),
+		ApplyChannelUpdate:        s.graphBuilder.ApplyChannelUpdate,
+		ClosedSCIDs:               s.fetchClosedChannelSCIDs(),
+		TrafficShaper:             implCfg.TrafficShaper,
+		GcFailedPaymentsOnStartup: cfg.GcFailedPaymentsOnStartup,
 	})
 	if err != nil {
 		return nil, fmt.Errorf("can't create router: %w", err)