[2/?] Local reputation: persistence and in-flight replay across restarts

config_builder.go

@@ -55,6 +55,7 @@ import (
 	paymentsdb "github.com/lightningnetwork/lnd/payments/db"
 	paymentsmig1 "github.com/lightningnetwork/lnd/payments/db/migration1"
 	paymentsmig1sqlc "github.com/lightningnetwork/lnd/payments/db/migration1/sqlc"
+	"github.com/lightningnetwork/lnd/reputation"
 	"github.com/lightningnetwork/lnd/rpcperms"
 	"github.com/lightningnetwork/lnd/signal"
 	"github.com/lightningnetwork/lnd/sqldb"
@@ -1028,6 +1029,11 @@ type DatabaseInstances struct {
 	// be used for native SQL queries for tables that already support it.
 	// This may be nil if the use-native-sql flag was not set.
 	NativeSQLStore sqldb.DB
+
+	// ReputationStore is the database that stores the local reputation
+	// state of channels. It is only set when the native SQL store is in
+	// use; otherwise reputation is kept in memory only.
+	ReputationStore reputation.Store
 }
 
 // DefaultDatabaseBuilder is a type that builds the default database backends
@@ -1360,6 +1366,16 @@ func (d *DefaultDatabaseBuilder) BuildDatabase(
 
 		dbs.InvoiceDB = sqlInvoiceDB
 
+		// Create the reputation store.
+		reputationExecutor := sqldb.NewTransactionExecutor(
+			baseDB, func(tx *sql.Tx) reputation.SQLQueries {
+				return baseDB.WithTx(tx)
+			},
+		)
+		dbs.ReputationStore = reputation.NewSQLStore(
+			reputationExecutor,
+		)
+
 		// Create the graph store.
 		graphExecutor := sqldb.NewTransactionExecutor(
 			baseDB, func(tx *sql.Tx) graphdb.SQLQueries {

docs/release-notes/unreleased/11266.md

@@ -0,0 +1,20 @@
+# New Features
+
+## Functional Enhancements
+
+* The local reputation subsystem now [persists its channel
+  state](https://github.com/lightningnetwork/lnd/pull/11266) when the native SQL
+  store is in use (`db.use-native-sql`), so peers keep the reputation they have
+  built across a restart, with the downtime decaying it as if the node had
+  stayed online. HTLCs that were in flight during the restart are picked up
+  again from the switch on startup, and closed channels are dropped from the
+  reputation state. Nodes without the native SQL store keep the previous
+  in-memory behaviour.
+
+# Technical and Architectural Updates
+
+## Database
+
+* A new native SQL table, `reputation_channels`, [stores the local reputation
+  state of channels](https://github.com/lightningnetwork/lnd/pull/11266). It is
+  created by schema migration 16 for nodes running with `db.use-native-sql`.

htlcswitch/circuit_map.go

@@ -123,6 +123,10 @@ type CircuitMap interface {
 	// circuits that use the given payment hash.
 	LookupByPaymentHash(hash [32]byte) []*PaymentCircuit
 
+	// ActiveCircuits returns all open circuits, i.e. those whose HTLC has
+	// been forwarded on an outgoing link and is awaiting resolution.
+	ActiveCircuits() []*PaymentCircuit
+
 	// NumPending returns the total number of active circuits added by
 	// CommitCircuits.
 	NumPending() int
@@ -1248,3 +1252,19 @@ func (cm *circuitMap) NumOpen() int {
 
 	return len(cm.opened)
 }
+
+// ActiveCircuits returns a snapshot of all open circuits, i.e. those whose
+// HTLC has been forwarded on an outgoing link and is awaiting a settle or fail
+// from the remote peer. The circuit map never mutates an opened circuit in
+// place, so the returned circuits are safe to read without the lock.
+func (cm *circuitMap) ActiveCircuits() []*PaymentCircuit {
+	cm.mtx.RLock()
+	defer cm.mtx.RUnlock()
+
+	circuits := make([]*PaymentCircuit, 0, len(cm.opened))
+	for _, c := range cm.opened {
+		circuits = append(circuits, c)
+	}
+
+	return circuits
+}

htlcswitch/circuit_test.go

@@ -1359,3 +1359,74 @@ func TestCircuitMapDeleteOpenCircuit(t *testing.T) {
 			circuit2, nil)
 	}
 }
+
+// TestCircuitMapActiveCircuits checks that ActiveCircuits returns exactly the
+// opened circuits, that they survive a restart of the circuit map, and that a
+// circuit is no longer reported once it is closed.
+func TestCircuitMapActiveCircuits(t *testing.T) {
+	t.Parallel()
+
+	cfg, circuitMap := newCircuitMap(t, false)
+
+	// Nothing tracked yet.
+	require.Empty(t, circuitMap.ActiveCircuits())
+
+	// One circuit that is only committed, never opened, and two that are
+	// opened on outgoing channels.
+	pending := htlcswitch.Keystone{
+		InKey: htlcswitch.CircuitKey{
+			ChanID: lnwire.NewShortChanIDFromInt(1), HtlcID: 0,
+		},
+	}
+	openOne := htlcswitch.Keystone{
+		InKey: htlcswitch.CircuitKey{
+			ChanID: lnwire.NewShortChanIDFromInt(1), HtlcID: 1,
+		},
+		OutKey: htlcswitch.CircuitKey{
+			ChanID: lnwire.NewShortChanIDFromInt(2), HtlcID: 0,
+		},
+	}
+	openTwo := htlcswitch.Keystone{
+		InKey: htlcswitch.CircuitKey{
+			ChanID: lnwire.NewShortChanIDFromInt(3), HtlcID: 0,
+		},
+		OutKey: htlcswitch.CircuitKey{
+			ChanID: lnwire.NewShortChanIDFromInt(2), HtlcID: 1,
+		},
+	}
+	for _, ks := range []htlcswitch.Keystone{pending, openOne, openTwo} {
+		require.NoError(t, createTestCircuit(ks, circuitMap))
+	}
+
+	// activeKeys reduces the snapshot to (incoming, outgoing) key pairs
+	// for order independent comparison.
+	activeKeys := func(cm htlcswitch.CircuitMap) map[htlcswitch.CircuitKey]htlcswitch.CircuitKey { //nolint:ll
+		keys := make(map[htlcswitch.CircuitKey]htlcswitch.CircuitKey)
+		for _, c := range cm.ActiveCircuits() {
+			require.NotNil(t, c.Outgoing, "open circuit has no "+
+				"outgoing key")
+			keys[c.Incoming] = *c.Outgoing
+		}
+
+		return keys
+	}
+
+	want := map[htlcswitch.CircuitKey]htlcswitch.CircuitKey{
+		openOne.InKey: openOne.OutKey,
+		openTwo.InKey: openTwo.OutKey,
+	}
+	require.Equal(t, want, activeKeys(circuitMap))
+
+	// Opened circuits are persisted, so they are still active after the
+	// circuit map is rebuilt from disk.
+	_, circuitMap = restartCircuitMap(t, cfg)
+	require.Equal(t, want, activeKeys(circuitMap))
+
+	// Closing one drops it from the snapshot.
+	_, err := circuitMap.FailCircuit(openOne.InKey)
+	require.NoError(t, err)
+	require.NoError(t, circuitMap.DeleteCircuits(openOne.InKey))
+
+	delete(want, openOne.InKey)
+	require.Equal(t, want, activeKeys(circuitMap))
+}

htlcswitch/mock.go

@@ -1150,6 +1150,10 @@ func (m *mockCircuitMap) LookupByPaymentHash(hash [32]byte) []*PaymentCircuit {
 	return nil
 }
 
+func (m *mockCircuitMap) ActiveCircuits() []*PaymentCircuit {
+	return nil
+}
+
 func (m *mockCircuitMap) NumPending() int {
 	return 0
 }

htlcswitch/switch.go

@@ -2624,6 +2624,12 @@ func (s *Switch) CircuitLookup() CircuitLookup {
 	return s.circuits
 }
 
+// ActiveCircuits returns a snapshot of the open circuits tracked by the switch:
+// the HTLCs it has forwarded on an outgoing link that are awaiting resolution.
+func (s *Switch) ActiveCircuits() []*PaymentCircuit {
+	return s.circuits.ActiveCircuits()
+}
+
 // commitCircuits persistently adds a circuit to the switch's circuit map.
 func (s *Switch) commitCircuits(circuits ...*PaymentCircuit) (
 	*CircuitFwdActions, error) {
@@ -3233,10 +3239,16 @@ func (s *Switch) handlePacketAdd(packet *htlcPacket,
 }
 
 // htlcAccountable extracts the experimental accountable signal from an
-// incoming update_add_htlc's custom records (TLV 106823).
+// incoming update_add_htlc's custom records.
 func htlcAccountable(htlc *lnwire.UpdateAddHTLC) bool {
+	return AccountableFromRecords(htlc.CustomRecords)
+}
+
+// AccountableFromRecords reports whether the given custom records carry the
+// experimental accountable signal (TLV 106823).
+func AccountableFromRecords(records lnwire.CustomRecords) bool {
 	key := uint64(lnwire.ExperimentalAccountableType)
-	rec, ok := htlc.CustomRecords[key]
+	rec, ok := records[key]
 
 	return ok && len(rec) > 0 && rec[0] == lnwire.ExperimentalAccountable
 }

itest/list_on_test.go

@@ -14,6 +14,10 @@ var allTestCases = []*lntest.TestCase{
 		Name:     "local reputation log only",
 		TestFunc: testLocalReputationLogOnly,
 	},
+	{
+		Name:     "local reputation persistence",
+		TestFunc: testLocalReputationPersistence,
+	},
 	{
 		Name:     "update channel status",
 		TestFunc: testUpdateChanStatus,