itest: add forwarding ability integration test

Open a channel, seed forwarding events with payments, and assert the
ForwardingAbility RPC returns a decodable response whose window matches the
request. Note that we can only simulate forwards back to the same node
because we only have two lnd nodes available in tests.

itest/channel_events_test.go

@@ -11,6 +11,8 @@ import (
 	"github.com/lightningnetwork/lnd/lnrpc"
 	"github.com/lightningnetwork/lnd/lnwire"
 	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 )
 
 // TestGetChannelEvents pins the GetChannelEvents RPC contract: a regtest
@@ -146,3 +148,215 @@ func TestGetChannelEvents(t *testing.T) {
 			"event order mismatch at index %d", i)
 	}
 }
+
+// TestForwardingAbility integration test opens a channel, sends payments to
+// seed events, and verifies that calling the ForwardingAbility RPC returns
+// the peer pair analytics successfully and can be decoded.
+func TestForwardingAbility(t *testing.T) {
+	c := newTestContext(t)
+	defer c.stop()
+
+	ctx := context.Background()
+
+	// Connect nodes and open a channel from alice to bob.
+	var aliceChannelAmt = btcutil.Amount(500000)
+
+	err := c.aliceClient.Client.Connect(
+		ctx, c.bobPubkey, "localhost:10012", true,
+	)
+	require.NoError(c.t, err, "could not connect nodes")
+
+	_, _ = c.openChannel(
+		c.aliceClient.Client, c.bobPubkey, aliceChannelAmt,
+	)
+
+	// Wait until alice can route a payment to bob.
+	var paymentAmount lnwire.MilliSatoshi = 20000000
+	c.eventuallyf(func() bool {
+		return c.channelRoutable(c.bobPubkey, paymentAmount)
+	}, "channel did not become routable")
+
+	// Send a payment from alice to bob.
+	hash, payreq := c.addInvoice(c.bobClient.Client, paymentAmount)
+	c.makePayment(
+		c.aliceClient.LndServices, c.bobClient.LndServices,
+		lndclient.SendPaymentRequest{
+			Invoice:     payreq,
+			PaymentHash: &hash,
+			Timeout:     paymentTimeout,
+		}, lnrpc.Payment_SUCCEEDED,
+	)
+
+	// The alice->bob payment moved liquidity onto bob's side of the only
+	// channel, so from this point on the bob self-pair holds at least the
+	// requested floor. Measuring over a window that starts now keeps the
+	// pair's uptime fraction high, which both clears the uptime threshold
+	// and keeps the node-down guard satisfied. A future end time extends
+	// the window over the still-funded state.
+	bobHex := c.bobPubkey.String()
+	windowStart := time.Now()
+
+	// The events store ingests channel updates asynchronously, so retry
+	// until the bob self-pair surfaces.
+	var ability frdrpc.ForwardingAbility
+	c.eventuallyf(func() bool {
+		endTime := time.Now().Add(2 * time.Second).Unix()
+		resp, err := c.faradayClient.ForwardingAbility(
+			ctx, &frdrpc.ForwardingAbilityRequest{
+				StartTime:         uint64(windowStart.Unix()),
+				EndTime:           uint64(endTime),
+				LiquidityFloorSat: 1000,
+				UptimeThreshold:   0.1,
+			},
+		)
+		if err != nil {
+			return false
+		}
+
+		decoded, err := frdrpc.DecodeForwardingAbility(resp)
+		if err != nil {
+			return false
+		}
+
+		a, ok := decoded[bobHex][bobHex]
+		if !ok {
+			return false
+		}
+		ability = a
+
+		return true
+	}, "expected bob self-pair in forwarding ability")
+
+	// The bob self-pair was up but never forwarded through itself, so it
+	// surfaces via the up-but-idle bitmask: non-zero effective uptime and
+	// zero forwarded volume.
+	require.Greater(c.t, ability.EffectiveUptimeS, int64(0))
+	require.Zero(c.t, ability.ForwardedSat)
+}
+
+// TestForwardingDowntime exercises the offline/online plumbing end to end. It
+// disconnects the only channel peer to take the channel offline, asserts that
+// faraday records the resulting offline event, then reconnects the peer and
+// asserts the recovering online event lands and the bob self-pair surfaces in
+// ForwardingAbility again. This proves downtime and recovery flow through to
+// the analyzer; the exact per-second uptime math is covered deterministically
+// by the analyzer unit tests.
+func TestForwardingDowntime(t *testing.T) {
+	c := newTestContext(t)
+	defer c.stop()
+
+	ctx := context.Background()
+
+	// Connect nodes and open a channel from alice to bob.
+	var aliceChannelAmt = btcutil.Amount(500000)
+
+	err := c.aliceClient.Client.Connect(
+		ctx, c.bobPubkey, "localhost:10012", true,
+	)
+	require.NoError(c.t, err, "could not connect nodes")
+
+	aliceChannel, _ := c.openChannel(
+		c.aliceClient.Client, c.bobPubkey, aliceChannelAmt,
+	)
+
+	// Wait until alice can route a payment to bob.
+	var paymentAmount lnwire.MilliSatoshi = 20000000
+	c.eventuallyf(func() bool {
+		return c.channelRoutable(c.bobPubkey, paymentAmount)
+	}, "channel did not become routable")
+
+	// Move liquidity onto bob's side so the bob self-pair clears the
+	// liquidity floor while the channel is up.
+	hash, payreq := c.addInvoice(c.bobClient.Client, paymentAmount)
+	c.makePayment(
+		c.aliceClient.LndServices, c.bobClient.LndServices,
+		lndclient.SendPaymentRequest{
+			Invoice:     payreq,
+			PaymentHash: &hash,
+			Timeout:     paymentTimeout,
+		}, lnrpc.Payment_SUCCEEDED,
+	)
+
+	chanPoint := aliceChannel.String()
+	bobHex := c.bobPubkey.String()
+
+	// Snapshot the event counts before the disconnect so we can detect the
+	// new offline and online events the disconnect and recovery produce.
+	onlineBefore, offlineBefore := c.channelEventCounts(chanPoint)
+
+	// Disconnect bob to take the only channel offline.
+	c.disconnectPeer(c.aliceClient, c.bobPubkey)
+
+	// faraday should ingest the resulting offline event: this is the
+	// downtime signal that the channel went inactive.
+	c.eventuallyf(func() bool {
+		_, offline := c.channelEventCounts(chanPoint)
+		return offline > offlineBefore
+	}, "expected an offline event after disconnect")
+
+	// An explicit DisconnectPeer is sticky: lnd does not auto-reconnect, so
+	// the channel stays offline until we reconnect. A window that sits
+	// entirely in this offline period leaves no pair clearing the uptime
+	// threshold, so the node-down guard rejects the request with
+	// FailedPrecondition rather than returning an empty response. The
+	// threshold is irrelevant here since the only pair has zero uptime.
+	c.eventuallyf(func() bool {
+		now := time.Now()
+		_, err := c.faradayClient.ForwardingAbility(
+			ctx, &frdrpc.ForwardingAbilityRequest{
+				StartTime: uint64(now.Unix()),
+				EndTime: uint64(
+					now.Add(2 * time.Second).Unix(),
+				),
+				LiquidityFloorSat: 1000,
+				UptimeThreshold:   0.9,
+			},
+		)
+
+		return status.Code(err) == codes.FailedPrecondition
+	}, "expected node-down guard while bob is disconnected")
+
+	// An explicit DisconnectPeer drops lnd's persistent connection, so the
+	// channel only comes back up once we reconnect. Reconnect bob to bring
+	// the channel active again.
+	err = c.aliceClient.Client.Connect(
+		ctx, c.bobPubkey, "localhost:10012", true,
+	)
+	require.NoError(c.t, err, "could not reconnect nodes")
+
+	// The channel goes active again on reconnect, which faraday records as
+	// an online event.
+	c.eventuallyf(func() bool {
+		online, _ := c.channelEventCounts(chanPoint)
+		return online > onlineBefore
+	}, "expected an online event after reconnect")
+
+	// With the channel back up and liquidity still on bob's side, the bob
+	// self-pair surfaces in ForwardingAbility again over a fresh window
+	// that opens after recovery.
+	c.eventuallyf(func() bool {
+		now := time.Now()
+		resp, err := c.faradayClient.ForwardingAbility(
+			ctx, &frdrpc.ForwardingAbilityRequest{
+				StartTime: uint64(now.Unix()),
+				EndTime: uint64(
+					now.Add(2 * time.Second).Unix(),
+				),
+				LiquidityFloorSat: 1000,
+				UptimeThreshold:   0.1,
+			},
+		)
+		if err != nil {
+			return false
+		}
+
+		decoded, err := frdrpc.DecodeForwardingAbility(resp)
+		if err != nil {
+			return false
+		}
+
+		_, ok := decoded[bobHex][bobHex]
+
+		return ok
+	}, "expected bob self-pair after reconnect")
+}

itest/test_context.go

@@ -466,6 +466,58 @@ func (c *testContext) channelRoutable(dest route.Vertex,
 	return err == nil
 }
 
+// disconnectPeer disconnects the given client from a peer, taking any channels
+// between them offline. lnd normally refuses to disconnect from a peer with an
+// active channel, but the itest lnd is a non-integration build where unsafe
+// disconnect is always permitted. The raw lnrpc client is used because the
+// high-level lndclient interface exposes no Disconnect, and the admin macaroon
+// is attached at call time since the shared connection carries none.
+func (c *testContext) disconnectPeer(client *lndclient.GrpcLndServices,
+	peer route.Vertex) {
+
+	c.t.Helper()
+
+	ctx, err := client.WithMacaroonAuthForService(
+		context.Background(), lndclient.AdminServiceMac,
+	)
+	require.NoError(c.t, err, "could not attach macaroon")
+
+	raw := lnrpc.NewLightningClient(client.ClientConn)
+	_, err = raw.DisconnectPeer(ctx, &lnrpc.DisconnectPeerRequest{
+		PubKey: peer.String(),
+	})
+	require.NoError(c.t, err, "could not disconnect peer")
+}
+
+// channelEventCounts returns how many online and offline events faraday has
+// recorded for the given channel up to the present.
+func (c *testContext) channelEventCounts(chanPoint string) (online,
+	offline int) {
+
+	c.t.Helper()
+
+	endTime := time.Now().Add(time.Second).Unix()
+	events, err := c.faradayClient.GetChannelEvents(
+		context.Background(), &frdrpc.ChannelEventsRequest{
+			ChanPoint: chanPoint,
+			EndTime:   endTime,
+		},
+	)
+	require.NoError(c.t, err, "could not get channel events")
+
+	for _, event := range events.Events {
+		switch event.EventType {
+		case frdrpc.ChannelEventType_CHAN_EVENT_ONLINE:
+			online++
+
+		case frdrpc.ChannelEventType_CHAN_EVENT_OFFLINE:
+			offline++
+		}
+	}
+
+	return online, offline
+}
+
 // findChannel finds a channel in a set of open channels, returning nil if it
 // is not found.
 // nolint:interfacer