diff --git a/chanevents/analyzer.go b/chanevents/analyzer.go index 5ca78a2..19812a5 100644 --- a/chanevents/analyzer.go +++ b/chanevents/analyzer.go @@ -181,32 +181,58 @@ func (a *ForwardingAnalyzer) EffectiveUptime(ctx context.Context, startTime, ) } -// getForwardingData returns successful forwards and channels from lnd's -// forwarding history over [startTime, endTime), indexed by peer pair. Unknown -// channels are skipped. +// getForwardingData queries lnd's forwarding history sequentially in paginated +// batches to retrieve successful forwarding events within the specified time +// range, indexing the results by peer pair. func (a *ForwardingAnalyzer) getForwardingData(ctx context.Context, startTime, endTime time.Time, scidToPeer map[uint64]string) ( map[PeerPair][]btcutil.Amount, map[uint64]string, error) { - fwds, err := a.lnd.Client.ForwardingHistory( - ctx, lndclient.ForwardingHistoryRequest{ - StartTime: startTime, - EndTime: endTime, - }, - ) - if err != nil { - return nil, nil, err + var events []lndclient.ForwardingEvent + var offset uint32 + const forwardingPageSize = 1000 + + for { + fwds, err := a.lnd.Client.ForwardingHistory( + ctx, lndclient.ForwardingHistoryRequest{ + StartTime: startTime, + EndTime: endTime, + Offset: offset, + MaxEvents: forwardingPageSize, + }, + ) + if err != nil { + return nil, nil, err + } + + if len(fwds.Events) == 0 { + break + } + + events = append(events, fwds.Events...) + if len(fwds.Events) < forwardingPageSize { + break + } + + // Guard against a non-advancing offset: if lnd does not move + // LastIndexOffset past the cursor we already queried, stop + // rather than re-fetch the same page forever. + if fwds.LastIndexOffset <= offset { + break + } + offset = fwds.LastIndexOffset } + log.DebugS( ctx, "Found forwarding events", slog.Int( - "count", len(fwds.Events), + "count", len(events), ), ) channelPeersConsidered := make(map[uint64]string) successfulForwards := make(map[PeerPair][]btcutil.Amount) - for _, fwd := range fwds.Events { + for _, fwd := range events { inPeer, ok := scidToPeer[fwd.ChannelIn] if !ok { log.WarnS( diff --git a/chanevents/analyzer_test.go b/chanevents/analyzer_test.go index b49e313..c501f2c 100644 --- a/chanevents/analyzer_test.go +++ b/chanevents/analyzer_test.go @@ -836,13 +836,17 @@ func (s *stubLndChannelClient) ClosedChannels(_ context.Context) ( } func (s *stubLndChannelClient) ForwardingHistory(_ context.Context, - _ lndclient.ForwardingHistoryRequest) ( + req lndclient.ForwardingHistoryRequest) ( *lndclient.ForwardingHistoryResponse, error) { if s.forwardingHistory == nil { return &lndclient.ForwardingHistoryResponse{}, nil } + if req.Offset > 0 { + return &lndclient.ForwardingHistoryResponse{}, nil + } + return s.forwardingHistory, nil }