Merge pull request #9356 from funyug/add-filters-fwdhistory

lnrpc: add incoming/outgoing channel ids filter to forwarding history request

channeldb/forwarding_log.go

@@ -256,6 +256,16 @@ type ForwardingEventQuery struct {
 
 	// NumMaxEvents is the max number of events to return.
 	NumMaxEvents uint32
+
+	// IncomingChanIds is the list of channels to filter HTLCs being
+	// received from a particular channel.
+	// If the list is empty, then it is ignored.
+	IncomingChanIDs fn.Set[uint64]
+
+	// OutgoingChanIds is the list of channels to filter HTLCs being
+	// forwarded to a particular channel.
+	// If the list is empty, then it is ignored.
+	OutgoingChanIDs fn.Set[uint64]
 }
 
 // ForwardingLogTimeSlice is the response to a forwarding query. It includes
@@ -323,9 +333,13 @@ func (f *ForwardingLog) Query(q ForwardingEventQuery) (ForwardingLogTimeSlice,
 				return nil
 			}
 
-			// If we're not yet past the user defined offset, then
+			// If no incoming or outgoing channel IDs were provided
+			// and we're not yet past the user defined offset, then
 			// we'll continue to seek forward.
-			if recordsToSkip > 0 {
+			if recordsToSkip > 0 &&
+				q.IncomingChanIDs.IsEmpty() &&
+				q.OutgoingChanIDs.IsEmpty() {
+
 				recordsToSkip--
 				continue
 			}
@@ -349,11 +363,41 @@ func (f *ForwardingLog) Query(q ForwardingEventQuery) (ForwardingLogTimeSlice,
 				return err
 			}
 
+			// Check if the incoming channel ID matches the
+			// filter criteria. Either no filtering is
+			// applied (IsEmpty), or the ID is explicitly
+			// included.
+			incomingMatch := q.IncomingChanIDs.IsEmpty() ||
+				q.IncomingChanIDs.Contains(
+					event.IncomingChanID.ToUint64(),
+				)
+
+			// Check if the outgoing channel ID matches the
+			// filter criteria. Either no filtering is
+			// applied (IsEmpty), or  the ID is explicitly
+			// included.
+			outgoingMatch := q.OutgoingChanIDs.IsEmpty() ||
+				q.OutgoingChanIDs.Contains(
+					event.OutgoingChanID.ToUint64(),
+				)
+
+			// Skip this event if it doesn't match the
+			// filters.
+			if !incomingMatch || !outgoingMatch {
+				continue
+			}
+			// If we're not yet past the user defined offset
+			// then we'll continue to seek forward.
+			if recordsToSkip > 0 {
+				recordsToSkip--
+				continue
+			}
+
 			event.Timestamp = currentTime
 			resp.ForwardingEvents = append(
-				resp.ForwardingEvents, event,
+				resp.ForwardingEvents,
+				event,
 			)
-
 			recordOffset++
 		}
 

channeldb/forwarding_log_test.go

@@ -475,3 +475,134 @@ func writeOldFormatEvents(db *DB, events []ForwardingEvent) error {
 		return nil
 	})
 }
+
+// TestForwardingLogQueryChanIDs tests that querying the forwarding log with
+// various combinations of incoming and/or outgoing channel IDs returns the
+// correct subset of forwarding events.
+func TestForwardingLogQueryChanIDs(t *testing.T) {
+	t.Parallel()
+
+	db, err := MakeTestDB(t)
+	require.NoError(t, err, "unable to make test db")
+
+	log := ForwardingLog{db: db}
+
+	initialTime := time.Unix(1234, 0)
+	endTime := initialTime
+
+	numEvents := 10
+	incomingChanIDs := []lnwire.ShortChannelID{
+		lnwire.NewShortChanIDFromInt(2001),
+		lnwire.NewShortChanIDFromInt(2002),
+		lnwire.NewShortChanIDFromInt(2003),
+	}
+	outgoingChanIDs := []lnwire.ShortChannelID{
+		lnwire.NewShortChanIDFromInt(3001),
+		lnwire.NewShortChanIDFromInt(3002),
+		lnwire.NewShortChanIDFromInt(3003),
+	}
+
+	events := make([]ForwardingEvent, numEvents)
+	for i := 0; i < numEvents; i++ {
+		events[i] = ForwardingEvent{
+			Timestamp:      endTime,
+			IncomingChanID: incomingChanIDs[i%len(incomingChanIDs)],
+			OutgoingChanID: outgoingChanIDs[i%len(outgoingChanIDs)],
+			AmtIn:          lnwire.MilliSatoshi(rand.Int63()),
+			AmtOut:         lnwire.MilliSatoshi(rand.Int63()),
+			IncomingHtlcID: fn.Some(uint64(i)),
+			OutgoingHtlcID: fn.Some(uint64(i)),
+		}
+		endTime = endTime.Add(10 * time.Minute)
+	}
+
+	require.NoError(
+		t,
+		log.AddForwardingEvents(events),
+		"unable to add events",
+	)
+
+	tests := []struct {
+		name     string
+		query    ForwardingEventQuery
+		expected func(e ForwardingEvent) bool
+	}{
+		{
+			name: "only incomingChanIDs filter",
+			query: ForwardingEventQuery{
+				StartTime: initialTime,
+				EndTime:   endTime,
+				IncomingChanIDs: fn.NewSet(
+					incomingChanIDs[0].ToUint64(),
+					incomingChanIDs[1].ToUint64(),
+				),
+				IndexOffset:  0,
+				NumMaxEvents: 10,
+			},
+			expected: func(e ForwardingEvent) bool {
+				return e.IncomingChanID == incomingChanIDs[0] ||
+					e.IncomingChanID == incomingChanIDs[1]
+			},
+		},
+		{
+			name: "only outgoingChanIDs filter",
+			query: ForwardingEventQuery{
+				StartTime: initialTime,
+				EndTime:   endTime,
+				OutgoingChanIDs: fn.NewSet(
+					outgoingChanIDs[0].ToUint64(),
+					outgoingChanIDs[1].ToUint64(),
+				),
+				IndexOffset:  0,
+				NumMaxEvents: 10,
+			},
+			expected: func(e ForwardingEvent) bool {
+				return e.OutgoingChanID == outgoingChanIDs[0] ||
+					e.OutgoingChanID == outgoingChanIDs[1]
+			},
+		},
+		{
+			name: "incoming and outgoingChanIDs filter",
+			query: ForwardingEventQuery{
+				StartTime: initialTime,
+				EndTime:   endTime,
+				IncomingChanIDs: fn.NewSet(
+					incomingChanIDs[0].ToUint64(),
+					incomingChanIDs[1].ToUint64(),
+				),
+				OutgoingChanIDs: fn.NewSet(
+					outgoingChanIDs[0].ToUint64(),
+					outgoingChanIDs[1].ToUint64(),
+				),
+				IndexOffset:  0,
+				NumMaxEvents: 10,
+			},
+			expected: func(e ForwardingEvent) bool {
+				return e.IncomingChanID ==
+					incomingChanIDs[0] ||
+					e.IncomingChanID ==
+						incomingChanIDs[1] ||
+					e.OutgoingChanID ==
+						outgoingChanIDs[0] ||
+					e.OutgoingChanID ==
+						outgoingChanIDs[1]
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			result, err := log.Query(tc.query)
+			require.NoError(t, err, "query failed")
+
+			expected := make([]ForwardingEvent, 0)
+			for _, e := range events {
+				if tc.expected(e) {
+					expected = append(expected, e)
+				}
+			}
+
+			require.Equal(t, expected, result.ForwardingEvents)
+		})
+	}
+}

cmd/commands/cmd_payments.go

@@ -1521,10 +1521,11 @@ func listPayments(ctx *cli.Context) error {
 }
 
 var forwardingHistoryCommand = cli.Command{
-	Name:      "fwdinghistory",
-	Category:  "Payments",
-	Usage:     "Query the history of all forwarded HTLCs.",
-	ArgsUsage: "start_time [end_time] [index_offset] [max_events]",
+	Name:     "fwdinghistory",
+	Category: "Payments",
+	Usage:    "Query the history of all forwarded HTLCs.",
+	ArgsUsage: "start_time [end_time] [index_offset] [max_events]" +
+		"[--incoming_channel_ids] [--outgoing_channel_ids]",
 	Description: `
 	Query the HTLC switch's internal forwarding log for all completed
 	payment circuits (HTLCs) over a particular time range (--start_time and
@@ -1539,6 +1540,9 @@ var forwardingHistoryCommand = cli.Command{
 	The max number of events returned is 50k. The default number is 100,
 	callers can use the --max_events param to modify this value.
 
+	Incoming and outgoing channel IDs can be provided to further filter
+	the events. If not provided, all events will be returned.
+
 	Finally, callers can skip a series of events using the --index_offset
 	parameter. Each response will contain the offset index of the last
 	entry. Using this callers can manually paginate within a time slice.
@@ -1567,6 +1571,18 @@ var forwardingHistoryCommand = cli.Command{
 			Usage: "skip the peer alias lookup per forwarding " +
 				"event in order to improve performance",
 		},
+		cli.Int64SliceFlag{
+			Name: "incoming_chan_ids",
+			Usage: "the short channel id of the incoming " +
+				"channel to filter events by; can be " +
+				"specified multiple times in the same command",
+		},
+		cli.Int64SliceFlag{
+			Name: "outgoing_chan_ids",
+			Usage: "the short channel id of the outgoing " +
+				"channel to filter events by; can be " +
+				"specified multiple times in the same command",
+		},
 	},
 	Action: actionDecorator(forwardingHistory),
 }
@@ -1647,6 +1663,21 @@ func forwardingHistory(ctx *cli.Context) error {
 		NumMaxEvents:    maxEvents,
 		PeerAliasLookup: lookupPeerAlias,
 	}
+	outgoingChannelIDs := ctx.Int64Slice("outgoing_chan_ids")
+	if len(outgoingChannelIDs) != 0 {
+		req.OutgoingChanIds = make([]uint64, len(outgoingChannelIDs))
+		for i, c := range outgoingChannelIDs {
+			req.OutgoingChanIds[i] = uint64(c)
+		}
+	}
+
+	incomingChannelIDs := ctx.Int64Slice("incoming_chan_ids")
+	if len(incomingChannelIDs) != 0 {
+		req.IncomingChanIds = make([]uint64, len(incomingChannelIDs))
+		for i, c := range incomingChannelIDs {
+			req.IncomingChanIds[i] = uint64(c)
+		}
+	}
 	resp, err := client.ForwardingHistory(ctxc, req)
 	if err != nil {
 		return err

docs/release-notes/release-notes-0.20.0.md

@@ -37,12 +37,22 @@ circuit. The indices are only available for forwarding events saved after v0.20.
   the option to specify multiple channels this control can be extended to
   multiple hops leading to the node.
 
+
+* The `lnrpc.ForwardingHistory` RPC method now supports filtering by 
+  [`incoming_chan_ids` and `outgoing_chan_ids`](https://github.com/lightningnetwork/lnd/pull/9356). 
+  This allows to retrieve forwarding events for specific channels.
+
 ## lncli Additions
 
 * [`lncli sendpayment` and `lncli queryroutes` now support the
   `--route_hints` flag](https://github.com/lightningnetwork/lnd/pull/9721) to
   support routing through private channels.
 
+
+* The `lncli fwdinghistory` command now supports two new flags:
+  [`--incoming_chan_ids` and `--outgoing_chan_ids`](https://github.com/lightningnetwork/lnd/pull/9356).
+  These filters allows to query forwarding events for specific channels.
+
 # Improvements
 ## Functional Updates
 
@@ -108,4 +118,5 @@ circuit. The indices are only available for forwarding events saved after v0.20.
 
 * Abdulkbk
 * Elle Mouton
+* Funyug
 * Pins

lnrpc/lightning.proto

@@ -4671,6 +4671,14 @@ message ForwardingHistoryRequest {
     // Informs the server if the peer alias should be looked up for each
     // forwarding event.
     bool peer_alias_lookup = 5;
+
+    // List of incoming channel ids to filter htlcs received from a
+    // particular channel
+    repeated uint64 incoming_chan_ids = 6;
+
+    // List of outgoing channel ids to filter htlcs being forwarded to a
+    // particular channel
+    repeated uint64 outgoing_chan_ids = 7;
 }
 message ForwardingEvent {
     // Timestamp is the time (unix epoch offset) that this circuit was

lnrpc/lightning.swagger.json

@@ -5114,6 +5114,22 @@
         "peer_alias_lookup": {
           "type": "boolean",
           "description": "Informs the server if the peer alias should be looked up for each\nforwarding event."
+        },
+        "incoming_chan_ids": {
+          "type": "array",
+          "items": {
+            "type": "string",
+            "format": "uint64"
+          },
+          "title": "List of incoming channel ids to filter htlcs received from a\nparticular channel"
+        },
+        "outgoing_chan_ids": {
+          "type": "array",
+          "items": {
+            "type": "string",
+            "format": "uint64"
+          },
+          "title": "List of outgoing channel ids to filter htlcs being forwarded to a\nparticular channel"
         }
       }
     },

rpcserver.go

@@ -7990,13 +7990,20 @@ func (r *rpcServer) ForwardingHistory(ctx context.Context,
 		numEvents = 100
 	}
 
+	// Create sets of incoming and outgoing channel IDs from the request
+	// for faster lookups for filtering.
+	incomingChanIDs := fn.NewSet(req.IncomingChanIds...)
+	outgoingChanIDs := fn.NewSet(req.OutgoingChanIds...)
+
 	// Next, we'll map the proto request into a format that is understood by
 	// the forwarding log.
 	eventQuery := channeldb.ForwardingEventQuery{
-		StartTime:    startTime,
-		EndTime:      endTime,
-		IndexOffset:  req.IndexOffset,
-		NumMaxEvents: numEvents,
+		StartTime:       startTime,
+		EndTime:         endTime,
+		IndexOffset:     req.IndexOffset,
+		NumMaxEvents:    numEvents,
+		IncomingChanIDs: incomingChanIDs,
+		OutgoingChanIDs: outgoingChanIDs,
 	}
 	timeSlice, err := r.server.miscDB.ForwardingLog().Query(eventQuery)
 	if err != nil {