diff --git a/faraday.go b/faraday.go index 314c0cf..23b5255 100644 --- a/faraday.go +++ b/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. diff --git a/frdrpcserver/forwarding_ability.go b/frdrpcserver/forwarding_ability.go new file mode 100644 index 0000000..2d4088b --- /dev/null +++ b/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 +} diff --git a/frdrpcserver/forwarding_ability_test.go b/frdrpcserver/forwarding_ability_test.go new file mode 100644 index 0000000..f842dc6 --- /dev/null +++ b/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) + } + }, + ) + } +} diff --git a/frdrpcserver/perms/perms.go b/frdrpcserver/perms/perms.go index c17c541..6e20801 100644 --- a/frdrpcserver/perms/perms.go +++ b/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", + }}, } diff --git a/frdrpcserver/rpcserver.go b/frdrpcserver/rpcserver.go index f248490..c0e7b12 100644 --- a/frdrpcserver/rpcserver.go +++ b/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.