frdrpcserver: implement ForwardingAbility RPC handler

faraday.go

@@ -177,10 +177,14 @@ func (f *Faraday) Start() error {
 		return fmt.Errorf("error initializing faraday: %v", err)
 	}
 
+	fwdAnalyzer := chanevents.NewForwardingAnalyzer(
+		f.stores.ChanEventsStore, f.lnd.LndServices,
+	)
 	cfg := &frdrpcserver.Config{
-		Lnd:           f.lnd.LndServices,
-		ChanEvents:    f.stores.ChanEventsStore,
-		BitcoinClient: f.bitcoinClient,
+		Lnd:                f.lnd.LndServices,
+		ChanEvents:         f.stores.ChanEventsStore,
+		ForwardingAnalyzer: fwdAnalyzer,
+		BitcoinClient:      f.bitcoinClient,
 	}
 
 	// Create the RPC server.
@@ -400,10 +404,14 @@ func (f *Faraday) StartAsSubserver(lndGrpc *lndclient.GrpcLndServices,
 		return fmt.Errorf("error initializing faraday: %v", err)
 	}
 
+	fwdAnalyzer := chanevents.NewForwardingAnalyzer(
+		f.stores.ChanEventsStore, lndGrpc.LndServices,
+	)
 	cfg := &frdrpcserver.Config{
-		Lnd:           lndGrpc.LndServices,
-		ChanEvents:    f.stores.ChanEventsStore,
-		BitcoinClient: f.bitcoinClient,
+		Lnd:                lndGrpc.LndServices,
+		ChanEvents:         f.stores.ChanEventsStore,
+		ForwardingAnalyzer: fwdAnalyzer,
+		BitcoinClient:      f.bitcoinClient,
 	}
 
 	// Create the RPC server, but don't start it.

frdrpcserver/forwarding_ability.go

@@ -0,0 +1,169 @@
+package frdrpcserver
+
+import (
+	"context"
+	"log/slog"
+	"math"
+	"time"
+
+	"github.com/btcsuite/btcd/btcutil"
+	"github.com/lightninglabs/faraday/frdrpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+// defaultLiquidityFloorSat is the liquidity floor applied when the request
+// leaves liquidity_floor_sat unset. It approximates the smallest amount a
+// rebalancer would still move, below which a pair is not economically
+// forwardable.
+const defaultLiquidityFloorSat = 50_000
+
+// defaultUptimeThreshold is the uptime fraction applied when the request leaves
+// uptime_threshold unset. A pair that was up at least this fraction of the
+// window but did not forward is reported as a single bit rather than a full
+// entry. It is high so that only reliably available pairs are flagged, keeping
+// the response sparse and the node-down guard meaningful.
+const defaultUptimeThreshold = 0.9
+
+// ForwardingAbility returns the raw effective-uptime and forwarded-volume facts
+// for each peer pair over the requested window. An unset end_time defaults to
+// the current time and an unset liquidity_floor_sat to
+// defaultLiquidityFloorSat.
+func (s *RPCServer) ForwardingAbility(ctx context.Context,
+	req *frdrpc.ForwardingAbilityRequest) (
+	*frdrpc.ForwardingAbilityResponse, error) {
+
+	log.DebugS(
+		ctx, "Handling ForwardingAbility request",
+		slog.Uint64("start_time", req.StartTime),
+		slog.Uint64("end_time", req.EndTime),
+		slog.Uint64("liquidity_floor_sat", req.LiquidityFloorSat),
+	)
+
+	// time.Unix takes an int64, so reject any request value that would
+	// overflow when its uint64 seconds are narrowed below.
+	if req.StartTime > math.MaxInt64 {
+		return nil, status.Error(
+			codes.InvalidArgument,
+			"start_time exceeds maximum allowed value",
+		)
+	}
+	if req.EndTime > math.MaxInt64 {
+		return nil, status.Error(
+			codes.InvalidArgument,
+			"end_time exceeds maximum allowed value",
+		)
+	}
+
+	startTime := time.Unix(int64(req.StartTime), 0)
+	endTime := time.Now()
+	if req.EndTime != 0 {
+		endTime = time.Unix(int64(req.EndTime), 0)
+	}
+
+	if startTime.After(endTime) {
+		return nil, status.Error(
+			codes.InvalidArgument,
+			"start_time must be less than or equal to end_time",
+		)
+	}
+
+	if s.cfg.ForwardingAnalyzer == nil {
+		return nil, status.Error(
+			codes.Unavailable,
+			"forwarding analyzer is not configured",
+		)
+	}
+
+	liquidityFloor := req.LiquidityFloorSat
+	if liquidityFloor == 0 {
+		liquidityFloor = defaultLiquidityFloorSat
+	}
+
+	uptimeThreshold := req.UptimeThreshold
+	if uptimeThreshold == 0 {
+		uptimeThreshold = defaultUptimeThreshold
+	}
+
+	// Reject NaN explicitly: NaN comparisons are always false, so a bare
+	// range check would let it slip through.
+	if math.IsNaN(uptimeThreshold) || uptimeThreshold < 0 ||
+		uptimeThreshold > 1 {
+
+		return nil, status.Error(
+			codes.InvalidArgument,
+			"uptime_threshold must be in [0, 1]",
+		)
+	}
+
+	abilities, err := s.cfg.ForwardingAnalyzer.EffectiveUptime(
+		ctx, startTime, endTime, btcutil.Amount(liquidityFloor),
+	)
+	if err != nil {
+		log.ErrorS(
+			ctx, "EffectiveUptime failed", err,
+			slog.Time("start_time", startTime),
+			slog.Time("end_time", endTime),
+			slog.Uint64("liquidity_floor_sat", liquidityFloor),
+		)
+
+		return nil, status.Errorf(codes.Internal, "failed to "+
+			"calculate effective uptime: %v", err)
+	}
+
+	// Convert the flat map into the nested map the codec expects, carrying
+	// the raw facts through unchanged. EffectiveUptime is truncated to
+	// whole seconds to match the second-granularity wire field. A pair with
+	// only sub-second qualifying uptime therefore reports zero uptime while
+	// still carrying its forwarded volume.
+	nested := make(map[string]map[string]frdrpc.ForwardingAbility)
+	for pair, ability := range abilities {
+		if _, ok := nested[pair.PeerIn]; !ok {
+			nested[pair.PeerIn] =
+				make(map[string]frdrpc.ForwardingAbility)
+		}
+
+		nested[pair.PeerIn][pair.PeerOut] = frdrpc.ForwardingAbility{
+			EffectiveUptimeS: int64(
+				ability.EffectiveUptime.Seconds(),
+			),
+			ForwardedSat: int64(ability.ForwardedAmount),
+		}
+	}
+
+	// Guard against returning data when the node itself was down for the
+	// window. If no pair held at least the threshold fraction of uptime,
+	// the response carries no signal and lowering the threshold to surface
+	// something would only inflate it, so fail loudly instead.
+	minUptimeS := frdrpc.MinQualifyingUptime(
+		uptimeThreshold, endTime.Unix()-startTime.Unix(),
+	)
+	var qualifying int
+	for _, outMap := range nested {
+		for _, ability := range outMap {
+			if ability.EffectiveUptimeS >= minUptimeS {
+				qualifying++
+			}
+		}
+	}
+	if qualifying == 0 {
+		return nil, status.Error(codes.FailedPrecondition, "no peer "+
+			"pair met the uptime threshold over the window; the "+
+			"node may have been offline")
+	}
+
+	resp, err := frdrpc.EncodeForwardingAbility(
+		nested, startTime.Unix(), endTime.Unix(), uptimeThreshold,
+	)
+	if err != nil {
+		log.ErrorS(
+			ctx, "EncodeForwardingAbility failed", err,
+			slog.Int("pairs", len(abilities)),
+		)
+
+		return nil, status.Errorf(codes.Internal, "failed to encode "+
+			"forwarding ability: %v", err)
+	}
+
+	return resp, nil
+}

frdrpcserver/forwarding_ability_test.go

@@ -0,0 +1,299 @@
+package frdrpcserver
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/btcsuite/btcd/btcutil"
+	"github.com/lightninglabs/faraday/chanevents"
+	"github.com/lightninglabs/faraday/frdrpc"
+	"github.com/stretchr/testify/require"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+)
+
+type mockForwardingAnalyzer struct {
+	effectiveUptimeFunc func(ctx context.Context, startTime, endTime time.Time,
+		liquidityFloor btcutil.Amount) (
+		map[chanevents.PeerPair]chanevents.ForwardingAbility, error)
+}
+
+func (m *mockForwardingAnalyzer) EffectiveUptime(ctx context.Context, startTime,
+	endTime time.Time, liquidityFloor btcutil.Amount) (
+	map[chanevents.PeerPair]chanevents.ForwardingAbility, error) {
+
+	return m.effectiveUptimeFunc(ctx, startTime, endTime, liquidityFloor)
+}
+
+// TestForwardingAbility tests the ForwardingAbility RPC method, covering both
+// successful and error cases.
+func TestForwardingAbility(t *testing.T) {
+	const (
+		peerIn  = "02aaaabbbbcccc0000000000000000000000000000000000000000000000000001"
+		peerOut = "02aaaabbbbcccc0000000000000000000000000000000000000000000000000002"
+	)
+
+	// analyzerResult is the canned analyzer return for a case. A nil
+	// analyzerResult means the case leaves ForwardingAnalyzer unconfigured.
+	type analyzerResult func() (
+		map[chanevents.PeerPair]chanevents.ForwardingAbility, error)
+
+	tests := []struct {
+		name     string
+		analyzer analyzerResult
+		req      *frdrpc.ForwardingAbilityRequest
+
+		// wantCode is the expected gRPC status; codes.OK denotes
+		// success.
+		wantCode codes.Code
+
+		// check runs on success with the response and the floor the
+		// handler resolved and passed to the analyzer.
+		check func(t *testing.T, resp *frdrpc.ForwardingAbilityResponse,
+			gotFloor btcutil.Amount)
+	}{
+		{
+			name: "encodes analyzer facts",
+			analyzer: func() (
+				map[chanevents.PeerPair]chanevents.ForwardingAbility,
+				error) {
+
+				return map[chanevents.PeerPair]chanevents.ForwardingAbility{
+					{
+						PeerIn:  peerIn,
+						PeerOut: peerOut,
+					}: {
+						EffectiveUptime: 90 * time.Second,
+						ForwardedAmount: 550,
+					},
+				}, nil
+			},
+			req: &frdrpc.ForwardingAbilityRequest{
+				StartTime:         100,
+				EndTime:           200,
+				LiquidityFloorSat: 1000,
+			},
+			wantCode: codes.OK,
+			check: func(t *testing.T,
+				resp *frdrpc.ForwardingAbilityResponse,
+				gotFloor btcutil.Amount) {
+
+				// The explicit floor passes straight through.
+				require.Equal(t, btcutil.Amount(1000), gotFloor)
+				require.Len(t, resp.Peers, 2)
+				require.Len(t, resp.Entries, 1)
+				require.EqualValues(t, 100, resp.StartTime)
+				require.EqualValues(t, 200, resp.EndTime)
+				require.EqualValues(
+					t, 90, resp.Entries[0].EffectiveUptimeS,
+				)
+				require.EqualValues(
+					t, 550, resp.Entries[0].ForwardedSat,
+				)
+
+				// An unset threshold echoes the server default,
+				// and a forwarded pair leaves the bitmask empty.
+				require.Equal(
+					t, defaultUptimeThreshold,
+					resp.UptimeThreshold,
+				)
+				require.Empty(t, resp.UpButIdleBitmask)
+			},
+		},
+		{
+			name: "unset floor uses server default",
+			analyzer: func() (
+				map[chanevents.PeerPair]chanevents.ForwardingAbility,
+				error) {
+
+				// Return a fully-up pair so the node-down guard
+				// passes and the default floor can be observed.
+				return map[chanevents.PeerPair]chanevents.ForwardingAbility{
+					{PeerIn: peerIn, PeerOut: peerOut}: {
+						EffectiveUptime: 100 * time.Second,
+					},
+				}, nil
+			},
+			req: &frdrpc.ForwardingAbilityRequest{
+				StartTime: 100,
+				EndTime:   200,
+			},
+			wantCode: codes.OK,
+			check: func(t *testing.T,
+				_ *frdrpc.ForwardingAbilityResponse,
+				gotFloor btcutil.Amount) {
+
+				require.Equal(
+					t, btcutil.Amount(
+						defaultLiquidityFloorSat,
+					), gotFloor,
+				)
+			},
+		},
+		{
+			name: "node down trips guard",
+			analyzer: func() (
+				map[chanevents.PeerPair]chanevents.ForwardingAbility,
+				error) {
+
+				// A single pair, up well below the default 0.9
+				// threshold over the 100s window and with no
+				// forwards, leaves nothing that clears it.
+				return map[chanevents.PeerPair]chanevents.ForwardingAbility{
+					{PeerIn: peerIn, PeerOut: peerOut}: {
+						EffectiveUptime: 10 * time.Second,
+					},
+				}, nil
+			},
+			req: &frdrpc.ForwardingAbilityRequest{
+				StartTime: 100,
+				EndTime:   200,
+			},
+			wantCode: codes.FailedPrecondition,
+		},
+		{
+			name: "low-uptime forward does not rescue guard",
+			analyzer: func() (
+				map[chanevents.PeerPair]chanevents.ForwardingAbility,
+				error) {
+
+				// Forwarded volume at sub-threshold uptime must
+				// not satisfy the guard.
+				return map[chanevents.PeerPair]chanevents.ForwardingAbility{
+					{PeerIn: peerIn, PeerOut: peerOut}: {
+						EffectiveUptime: 10 * time.Second,
+						ForwardedAmount: 999,
+					},
+				}, nil
+			},
+			req: &frdrpc.ForwardingAbilityRequest{
+				StartTime: 100,
+				EndTime:   200,
+			},
+			wantCode: codes.FailedPrecondition,
+		},
+		{
+			name: "explicit threshold flags up-but-idle pair",
+			analyzer: func() (
+				map[chanevents.PeerPair]chanevents.ForwardingAbility,
+				error) {
+
+				return map[chanevents.PeerPair]chanevents.ForwardingAbility{
+					{PeerIn: peerIn, PeerOut: peerOut}: {
+						EffectiveUptime: 60 * time.Second,
+					},
+				}, nil
+			},
+			req: &frdrpc.ForwardingAbilityRequest{
+				StartTime:       100,
+				EndTime:         200,
+				UptimeThreshold: 0.5,
+			},
+			wantCode: codes.OK,
+			check: func(t *testing.T,
+				resp *frdrpc.ForwardingAbilityResponse,
+				_ btcutil.Amount) {
+
+				// Up 60s of a 100s window at a 0.5 threshold:
+				// idle, so a bit and no entry.
+				require.Equal(t, 0.5, resp.UptimeThreshold)
+				require.Empty(t, resp.Entries)
+				require.NotEmpty(t, resp.UpButIdleBitmask)
+			},
+		},
+		{
+			name: "out of range threshold is rejected",
+			analyzer: func() (
+				map[chanevents.PeerPair]chanevents.ForwardingAbility,
+				error) {
+
+				return nil, nil
+			},
+			req: &frdrpc.ForwardingAbilityRequest{
+				StartTime:       100,
+				EndTime:         200,
+				UptimeThreshold: 1.5,
+			},
+			wantCode: codes.InvalidArgument,
+		},
+		{
+			name: "start after end is rejected",
+			req: &frdrpc.ForwardingAbilityRequest{
+				StartTime: 200,
+				EndTime:   100,
+			},
+			wantCode: codes.InvalidArgument,
+		},
+		{
+			name: "missing analyzer is unavailable",
+			req: &frdrpc.ForwardingAbilityRequest{
+				StartTime:         100,
+				EndTime:           200,
+				LiquidityFloorSat: 1000,
+			},
+			wantCode: codes.Unavailable,
+		},
+		{
+			name: "analyzer error is internal",
+			analyzer: func() (
+				map[chanevents.PeerPair]chanevents.ForwardingAbility,
+				error) {
+
+				return nil, errors.New("db lookup failed")
+			},
+			req: &frdrpc.ForwardingAbilityRequest{
+				StartTime:         100,
+				EndTime:           200,
+				LiquidityFloorSat: 1000,
+			},
+			wantCode: codes.Internal,
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(
+			tc.name,
+			func(t *testing.T) {
+				var gotFloor btcutil.Amount
+
+				cfg := &Config{}
+				if tc.analyzer != nil {
+					cfg.ForwardingAnalyzer = &mockForwardingAnalyzer{
+						effectiveUptimeFunc: func(
+							_ context.Context, _,
+							_ time.Time,
+							floor btcutil.Amount) (
+							map[chanevents.PeerPair]chanevents.ForwardingAbility,
+							error) {
+
+							gotFloor = floor
+
+							return tc.analyzer()
+						},
+					}
+				}
+
+				server := NewRPCServer(cfg)
+				resp, err := server.ForwardingAbility(
+					t.Context(), tc.req,
+				)
+
+				if tc.wantCode != codes.OK {
+					st, ok := status.FromError(err)
+					require.True(t, ok)
+					require.Equal(t, tc.wantCode, st.Code())
+
+					return
+				}
+
+				require.NoError(t, err)
+				require.NotNil(t, resp)
+				if tc.check != nil {
+					tc.check(t, resp, gotFloor)
+				}
+			},
+		)
+	}
+}

frdrpcserver/perms/perms.go

@@ -37,4 +37,8 @@ var RequiredPermissions = map[string][]bakery.Op{
 		Entity: "events",
 		Action: "read",
 	}},
+	"/frdrpc.FaradayServer/ForwardingAbility": {{
+		Entity: "insights",
+		Action: "read",
+	}},
 }

frdrpcserver/rpcserver.go

@@ -12,7 +12,9 @@ package frdrpcserver
 import (
 	"context"
 	"errors"
+	"time"
 
+	"github.com/btcsuite/btcd/btcutil"
 	"github.com/lightninglabs/faraday/accounting"
 	"github.com/lightninglabs/faraday/chain"
 	"github.com/lightninglabs/faraday/chanevents"
@@ -53,6 +55,15 @@ type RPCServer struct {
 	cfg *Config
 }
 
+// ForwardingAnalyzer is the seam the RPC server uses to compute per-peer-pair
+// forwarding facts. It is an interface so the handler can be exercised against
+// a mock analyzer in tests.
+type ForwardingAnalyzer interface {
+	EffectiveUptime(ctx context.Context, startTime, endTime time.Time,
+		liquidityFloor btcutil.Amount) (
+		map[chanevents.PeerPair]chanevents.ForwardingAbility, error)
+}
+
 // Config provides closures and settings required to run the rpc server.
 type Config struct {
 	// Lnd is a client which can be used to query lnd.
@@ -61,6 +72,11 @@ type Config struct {
 	// ChanEvents is a database of channel events.
 	ChanEvents *chanevents.Store
 
+	// ForwardingAnalyzer computes forwarding-ability facts for the
+	// ForwardingAbility RPC. When nil, that endpoint returns
+	// codes.Unavailable.
+	ForwardingAnalyzer ForwardingAnalyzer
+
 	// BitcoinClient is an optional client which can be used to query
 	// on-chain data from a connected bitcoin node. If nil, faraday will
 	// not be able to serve endpoints which require on-chain data.