diff --git a/README.md b/README.md index f87a827..5f32386 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ per-peer basis. Known and trusted peers for example can be assigned a higher maximum, while a new channel from a previously unseen node may be limited to only a few pending htlcs. -Why are limits needed? +## Why are limits needed? In today's Lighting Network payments are routed via a series of hops. Each of those hops will incur a cost for forwarding that payment. While the htlc of an @@ -31,6 +31,33 @@ This is where `circuitbreaker` comes in. It puts up a defense around that valuable channel liquidity and helps to keep the locked coins at work to maximize routing revenue. +## Hold fees + +An alternative to lowering limits is to charge peers for the actual costs that +they generate in both the success and failure cases. For more information about +this idea, see thread [Hold fees: 402 Payment Required for Lightning +itself](https://lists.linuxfoundation.org/pipermail/lightning-dev/2020-October/002826.html) +on the `lightning-dev` mailing list. + +Circuit Breaker does not support 'breaking the circuits' when hold fees aren't +paid, but this is a potential direction for the future. It would roughly entail +requiring peers to deposit money for hold fees and blocking forwards once the +peer's balance is zero. + +What is currently implemented is only the reporting of these (virtual) hold +fees. A fee schedule can be defined (see configuration below) and the hold fee +that _could have been charged_ is logged for every forward. Additionally a +periodic report is printed that contains the hold fees charged to peers during +the reporting period. Peers that did not offer any htlcs in that period will be +omitted. + +```log +2020-10-17T20:45:15.708+0200 INFO Forwarding htlc {"channel": 39778131669745664, "htlc": 52, "peer_alias": "tester", "peer": "03afe7da13950201562df3fdd6c8b209aab248daee82d773b9dadebba3eeecbb4c", "pending_htlcs": 1, "max_pending_htlcs": 5} +2020-10-17T20:45:15.852+0200 INFO Resolving htlc {"channel": 39778131669745664, "htlc": 52, "peer_alias": "tester", "peer": "03afe7da13950201562df3fdd6c8b209aab248daee82d773b9dadebba3eeecbb4c", "pending_htlcs": 0, "hold_time": "143.396033ms", "hold_fee_msat": 4} +2020-10-17T20:45:20.000+0200 INFO Hold fees report {"next_report_time": "2020-10-17T20:45:25.000+0200"} +2020-10-17T20:45:20.000+0200 INFO Report {"peer_alias": "tester", "peer": "03afe7da13950201562df3fdd6c8b209aab248daee82d773b9dadebba3eeecbb4c", "total_fees_msat": 74, "interval_fees_msat": 4} +``` + ## How to use ### Requirements @@ -40,20 +67,45 @@ maximize routing revenue. ### Configuration `circuitbreaker` by default reads its configuration from `~/.circuitbreaker/circuitbreaker.yaml`. -Below is an example of a configuration that limits the number of pending htlcs -to five by default. For two peers, the limit is lowered to two. A last peer is -allowed to have up to a hundred htlcs in-flight. -``` +This is an example configuration: + +```yaml +# Default max pendings htlcs. Limit the number of pending htlcs to five by +# default. maxPendingHtlcs: 5 +# Define exception groups. For two peers, the limit is lowered to two. A last +# peer is allowed to have up to a hundred htlcs in-flight. groups: - maxPendingHtlcs: 2 peers: - 033220600ae3949f40739955948ca43fc60174c9c51fb51e6debfc27091e58cebe - - 021561e3cf45345052912c88b0df7deb7c2ec4a1cf08333edb1ed8dbb3fd203d77 - - maxPendingHtlcs: 100 + - 033220600ae3949f40739955948ca43fc60174c9c51fb51e6debfc27091e58ceba + - maxPendingHtlcs: 3 peers: - - 02674dabd68df75f78b6b6dc35dd49dd70db5293ca7a68f9cafa76adafabd5dc7c + - 033220600ae3949f40739955948ca43fc60174c9c51fb51e6debfc27091e58cebf + +# Define hold fee so that a fully locked 1 BTC channel would yield ~8.5% +# annually. +# +# The hold fee is not actually paid by channel peers and currently only exists +# to raise awareness of the costs of long-holding htlcs. +holdFee: + # Set the base hold fee to 500 sat per hour to compensate for the usage of an + # htlc slot. If an imaginary channel of 1 BTC would have all of its 483 slots + # occupied for a full year, the total hold fee would be 24 * 365 * 483 = + # 4231080 sats. This translates to a yearly return on the staked bitcoin of + # ~4.2%. + baseSatPerHr: 1 + + # Set the hold fee rate to 5 parts per million. If an imaginary channel of 1 + # BTC would have all of its funds time-locked for a full year, the total hold + # fee would be 24 * 365 * 100000000 * 5 / 1000000 = 4380000. This translates + # to a yearly return on the staked bitcoin of ~4.4%. + ratePpmPerHr: 5 + + # Report (virtually) collected hold fees once per hour. + reportingInterval: 1h ``` ### Run diff --git a/circuitbreaker-example.yaml b/circuitbreaker-example.yaml new file mode 100644 index 0000000..cb174fa --- /dev/null +++ b/circuitbreaker-example.yaml @@ -0,0 +1,36 @@ +# Default max pendings htlcs. Limit the number of pending htlcs to five by +# default. +maxPendingHtlcs: 5 + +# Define exception groups. For two peers, the limit is lowered to two. A last +# peer is allowed to have up to a hundred htlcs in-flight. +groups: + - maxPendingHtlcs: 2 + peers: + - 033220600ae3949f40739955948ca43fc60174c9c51fb51e6debfc27091e58cebe + - 033220600ae3949f40739955948ca43fc60174c9c51fb51e6debfc27091e58ceba + - maxPendingHtlcs: 3 + peers: + - 033220600ae3949f40739955948ca43fc60174c9c51fb51e6debfc27091e58cebf + +# Define hold fee so that a fully locked 1 BTC channel would yield ~8.5% +# annually. +# +# The hold fee is not actually paid by channel peers and currently only exists +# to raise awareness of the costs of long-holding htlcs. +holdFee: + # Set the base hold fee to 500 sat per hour to compensate for the usage of an + # htlc slot. If an imaginary channel of 1 BTC would have all of its 483 slots + # occupied for a full year, the total hold fee would be 24 * 365 * 483 = + # 4231080 sats. This translates to a yearly return on the staked bitcoin of + # ~4.2%. + baseSatPerHr: 1 + + # Set the hold fee rate to 5 parts per million. If an imaginary channel of 1 + # BTC would have all of its funds time-locked for a full year, the total hold + # fee would be 24 * 365 * 100000000 * 5 / 1000000 = 4380000. This translates + # to a yearly return on the staked bitcoin of ~4.4%. + ratePpmPerHr: 5 + + # Report (virtually) collected hold fees once per hour. + reportingInterval: 1h \ No newline at end of file diff --git a/config.go b/config.go index 0c0f476..c611795 100644 --- a/config.go +++ b/config.go @@ -4,6 +4,7 @@ import ( "fmt" "io/ioutil" "os" + "time" "github.com/lightningnetwork/lnd/routing/route" "go.uber.org/zap" @@ -18,11 +19,43 @@ type yamlGroup struct { type yamlConfig struct { MaxPendingHtlcs int `yaml:"maxPendingHtlcs"` Groups []yamlGroup `yaml:"groups"` + HoldFee holdFee `yaml:"holdFee"` +} + +type holdFee struct { + BaseSatPerHr int64 `yaml:"baseSatPerHr"` + RatePpmPerHr int `yaml:"ratePpmPerHr"` + ReportingInterval yamlTimeDur `yaml:"reportingInterval"` +} + +type yamlTimeDur time.Duration + +func (t *yamlTimeDur) UnmarshalYAML(unmarshal func(interface{}) error) error { + var tm string + if err := unmarshal(&tm); err != nil { + return err + } + + td, err := time.ParseDuration(tm) + if err != nil { + return fmt.Errorf("failed to parse '%s' to time.Duration: %v", tm, err) + } + + *t = yamlTimeDur(td) + return nil +} + +func (t *yamlTimeDur) Duration() time.Duration { + return time.Duration(*t) } type config struct { MaxPendingHtlcs int MaxPendingHtlcsPerPeer map[route.Vertex]int + + BaseSatPerHr int64 + RatePpmPerHr int + ReportingInterval time.Duration } var defaultConfig = config{ @@ -62,6 +95,9 @@ func (c *configLoader) load() (*config, error) { config := config{ MaxPendingHtlcs: yamlCfg.MaxPendingHtlcs, MaxPendingHtlcsPerPeer: make(map[route.Vertex]int), + BaseSatPerHr: yamlCfg.HoldFee.BaseSatPerHr, + RatePpmPerHr: yamlCfg.HoldFee.RatePpmPerHr, + ReportingInterval: time.Duration(yamlCfg.HoldFee.ReportingInterval), } for _, group := range yamlCfg.Groups { @@ -82,7 +118,8 @@ func (c *configLoader) load() (*config, error) { } } - log.Info("Read config file", zap.String("file", c.path)) + log.Infow("Read config file", + "file", c.path) return &config, nil } diff --git a/go.mod b/go.mod index 5c7c6bb..eeda654 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,7 @@ require ( github.com/btcsuite/btcutil v1.0.2 github.com/lightninglabs/protobuf-hex-display v1.3.3-0.20191212020323-b444784ce75d github.com/lightningnetwork/lnd v0.11.1-beta.rc3 + github.com/stretchr/testify v1.5.1 github.com/urfave/cli v1.22.4 go.uber.org/zap v1.14.1 google.golang.org/grpc v1.27.0 diff --git a/go.sum b/go.sum index 84308b8..d07e627 100644 --- a/go.sum +++ b/go.sum @@ -263,6 +263,7 @@ github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4k github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.2.0 h1:Hbg2NidpLE8veEBkEZTL3CvlkUIVzuU9jDplZO54c48= github.com/stretchr/objx v0.2.0/go.mod h1:qt09Ya8vawLte6SNmTgCsAVtYtaKzEcn8ATUoHMkEqE= github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= diff --git a/lndclient.go b/lndclient.go index 5686852..181af3e 100644 --- a/lndclient.go +++ b/lndclient.go @@ -2,11 +2,11 @@ package main import ( "context" + "errors" "fmt" "io/ioutil" "github.com/lightningnetwork/lnd/routing/route" - "go.uber.org/zap" "github.com/lightningnetwork/lnd/lnrpc/routerrpc" @@ -19,14 +19,14 @@ import ( "gopkg.in/macaroon.v2" ) -type lndclient struct { +type lndclientGrpc struct { conn *grpc.ClientConn main lnrpc.LightningClient router routerrpc.RouterClient } -func newLndClient(ctx *cli.Context) (*lndclient, error) { +func newLndClient(ctx *cli.Context) (*lndclientGrpc, error) { // First, we'll parse the args from the command. tlsCertPath, macPath, err := extractPathArgs(ctx) if err != nil { @@ -73,14 +73,14 @@ func newLndClient(ctx *cli.Context) (*lndclient, error) { "unable to connect to RPC server: %v", err) } - return &lndclient{ + return &lndclientGrpc{ conn: conn, main: lnrpc.NewLightningClient(conn), router: routerrpc.NewRouterClient(conn), }, nil } -func (l *lndclient) getIdentity() (route.Vertex, error) { +func (l *lndclientGrpc) getIdentity() (route.Vertex, error) { ctx, cancel := context.WithTimeout(ctxb, rpcTimeout) defer cancel() @@ -96,11 +96,12 @@ type channelEdge struct { node1Pub, node2Pub route.Vertex } -func (l *lndclient) getChanInfo(channel uint64) (*channelEdge, error) { +func (l *lndclientGrpc) getChanInfo(channel uint64) (*channelEdge, error) { ctx, cancel := context.WithTimeout(ctxb, rpcTimeout) defer cancel() - log.Debug("Retrieving channel info", zap.Uint64("channel", channel)) + log.Debugw("Retrieving channel info", + "channel", channel) info, err := l.main.GetChanInfo(ctx, &lnrpc.ChanInfoRequest{ ChanId: channel, @@ -125,6 +126,40 @@ func (l *lndclient) getChanInfo(channel uint64) (*channelEdge, error) { }, nil } -func (l *lndclient) close() { +func (l *lndclientGrpc) subscribeHtlcEvents(ctx context.Context, + in *routerrpc.SubscribeHtlcEventsRequest) ( + routerrpc.Router_SubscribeHtlcEventsClient, error) { + + return l.router.SubscribeHtlcEvents(ctx, in) +} + +func (l *lndclientGrpc) htlcInterceptor(ctx context.Context) ( + routerrpc.Router_HtlcInterceptorClient, error) { + + return l.router.HtlcInterceptor(ctx) +} + +func (l *lndclientGrpc) close() { l.conn.Close() } + +func (l *lndclientGrpc) getNodeAlias(key route.Vertex) (string, error) { + ctx, cancel := context.WithTimeout(ctxb, rpcTimeout) + defer cancel() + + log.Debugw("Retrieving node info", + "key", key) + + info, err := l.main.GetNodeInfo(ctx, &lnrpc.NodeInfoRequest{ + PubKey: key.String(), + }) + if err != nil { + return "", err + } + + if info.Node == nil { + return "", errors.New("node info not available") + } + + return info.Node.Alias, nil +} diff --git a/lndclient_mock.go b/lndclient_mock.go new file mode 100644 index 0000000..5eab4e9 --- /dev/null +++ b/lndclient_mock.go @@ -0,0 +1,85 @@ +package main + +import ( + "context" + + "github.com/lightningnetwork/lnd/lnrpc/routerrpc" + "github.com/lightningnetwork/lnd/routing/route" +) + +var mockIdentity = route.Vertex{1, 2, 3} + +type lndclientMock struct { + htlcEvents chan *routerrpc.HtlcEvent + htlcInterceptorRequests chan *routerrpc.ForwardHtlcInterceptRequest + htlcInterceptorResponses chan *routerrpc.ForwardHtlcInterceptResponse +} + +func newLndclientMock() *lndclientMock { + return &lndclientMock{ + htlcEvents: make(chan *routerrpc.HtlcEvent), + htlcInterceptorRequests: make(chan *routerrpc.ForwardHtlcInterceptRequest), + htlcInterceptorResponses: make(chan *routerrpc.ForwardHtlcInterceptResponse), + } +} + +func (l *lndclientMock) getIdentity() (route.Vertex, error) { + return mockIdentity, nil +} + +func (l *lndclientMock) getChanInfo(channel uint64) (*channelEdge, error) { + return &channelEdge{ + node1Pub: mockIdentity, + node2Pub: route.Vertex{byte(channel & 0xff)}, + }, nil +} + +func (l *lndclientMock) subscribeHtlcEvents(ctx context.Context, + in *routerrpc.SubscribeHtlcEventsRequest) ( + routerrpc.Router_SubscribeHtlcEventsClient, error) { + + return &htlcEventsMock{ + htlcEvents: l.htlcEvents, + }, nil +} + +func (l *lndclientMock) htlcInterceptor(ctx context.Context) ( + routerrpc.Router_HtlcInterceptorClient, error) { + + return &htlcInterceptorMock{ + htlcInterceptorRequests: l.htlcInterceptorRequests, + htlcInterceptorResponses: l.htlcInterceptorResponses, + }, nil +} + +func (l *lndclientMock) getNodeAlias(key route.Vertex) (string, error) { + return "alias-" + key.String()[:6], nil +} + +type htlcEventsMock struct { + routerrpc.Router_SubscribeHtlcEventsClient + + htlcEvents chan *routerrpc.HtlcEvent +} + +func (h *htlcEventsMock) Recv() (*routerrpc.HtlcEvent, error) { + event := <-h.htlcEvents + return event, nil +} + +type htlcInterceptorMock struct { + routerrpc.Router_HtlcInterceptorClient + + htlcInterceptorRequests chan *routerrpc.ForwardHtlcInterceptRequest + htlcInterceptorResponses chan *routerrpc.ForwardHtlcInterceptResponse +} + +func (h *htlcInterceptorMock) Send(resp *routerrpc.ForwardHtlcInterceptResponse) error { + h.htlcInterceptorResponses <- resp + return nil +} + +func (h *htlcInterceptorMock) Recv() (*routerrpc.ForwardHtlcInterceptRequest, error) { + event := <-h.htlcInterceptorRequests + return event, nil +} diff --git a/log.go b/log.go index fb7d21c..ec2d474 100644 --- a/log.go +++ b/log.go @@ -5,11 +5,12 @@ import ( "go.uber.org/zap/zapcore" ) -var log *zap.Logger +var log *zap.SugaredLogger func init() { config := zap.NewDevelopmentConfig() config.EncoderConfig.EncodeLevel = zapcore.CapitalColorLevelEncoder config.EncoderConfig.EncodeCaller = nil - log, _ = config.Build() + rawLog, _ := config.Build() + log = rawLog.Sugar() } diff --git a/main.go b/main.go index 37eba46..8d37c5f 100644 --- a/main.go +++ b/main.go @@ -1,6 +1,7 @@ package main import ( + "context" "fmt" "os" "os/user" @@ -10,7 +11,6 @@ import ( "github.com/btcsuite/btcutil" "github.com/lightningnetwork/lnd/build" "github.com/urfave/cli" - "go.uber.org/zap" "google.golang.org/grpc" ) @@ -136,11 +136,11 @@ func main() { defer client.close() p := newProcess() - return p.run(client, config) + return p.run(context.Background(), client, config) } if err := app.Run(os.Args); err != nil { - log.Error("Unexpected exit", zap.Error(err)) + log.Errorw("Unexpected exit", "err", err) } } diff --git a/process.go b/process.go index 40e88da..bf9940d 100644 --- a/process.go +++ b/process.go @@ -9,7 +9,6 @@ import ( "github.com/lightninglabs/protobuf-hex-display/proto" "github.com/lightningnetwork/lnd/lnrpc/routerrpc" "github.com/lightningnetwork/lnd/routing/route" - "go.uber.org/zap" ) const maxPending = 1 @@ -19,6 +18,21 @@ var ( ctxb = context.Background() ) +type lndclient interface { + getIdentity() (route.Vertex, error) + + getChanInfo(channel uint64) (*channelEdge, error) + + getNodeAlias(key route.Vertex) (string, error) + + subscribeHtlcEvents(ctx context.Context, + in *routerrpc.SubscribeHtlcEventsRequest) ( + routerrpc.Router_SubscribeHtlcEventsClient, error) + + htlcInterceptor(ctx context.Context) ( + routerrpc.Router_HtlcInterceptorClient, error) +} + type circuitKey struct { channel uint64 htlc uint64 @@ -26,17 +40,19 @@ type circuitKey struct { type interceptEvent struct { circuitKey - resume chan bool + valueMsat int64 + resume chan bool } type process struct { - client *lndclient + client lndclient interceptChan chan interceptEvent resolveChan chan circuitKey identity route.Vertex pubkeyMap map[uint64]route.Vertex + aliasMap map[route.Vertex]string } func newProcess() *process { @@ -44,11 +60,17 @@ func newProcess() *process { interceptChan: make(chan interceptEvent), resolveChan: make(chan circuitKey), pubkeyMap: make(map[uint64]route.Vertex), + aliasMap: make(map[route.Vertex]string), } } -func (p *process) run(client *lndclient, cfg *config) error { +func (p *process) run(ctx context.Context, client lndclient, cfg *config) error { log.Info("CircuitBreaker started") + log.Infow("Hold fee", + "base", cfg.BaseSatPerHr, + "rate", float64(cfg.RatePpmPerHr)/1e6, + "reporting_interval", cfg.ReportingInterval, + ) p.client = client @@ -58,17 +80,17 @@ func (p *process) run(client *lndclient, cfg *config) error { return err } - log.Info("Connected to lnd node", - zap.String("pubkey", p.identity.String())) + log.Infow("Connected to lnd node", + "pubkey", p.identity.String()) - stream, err := p.client.router.SubscribeHtlcEvents( + stream, err := p.client.subscribeHtlcEvents( ctxb, &routerrpc.SubscribeHtlcEventsRequest{}, ) if err != nil { return err } - interceptor, err := p.client.router.HtlcInterceptor(ctxb) + interceptor, err := p.client.htlcInterceptor(ctxb) if err != nil { return err } @@ -78,23 +100,46 @@ func (p *process) run(client *lndclient, cfg *config) error { go func() { err := p.processHtlcEvents(stream) if err != nil { - log.Error("htlc events error", zap.Error(err)) + log.Errorw("htlc events error", + "err", err) } }() go func() { err := p.processInterceptor(interceptor) if err != nil { - log.Error("interceptor error", zap.Error(err)) + log.Errorw("interceptor error", + "err", err) } }() - return p.eventLoop(cfg) + return p.eventLoop(ctx, cfg) } -func (p *process) eventLoop(cfg *config) error { - pendingHtlcs := make(map[route.Vertex]map[circuitKey]struct{}) +type holdInfo struct { + fwdTime time.Time + valueMsat int64 +} + +type peerInfo struct { + htlcs map[circuitKey]*holdInfo + + totalHoldFees int64 + intervalHoldFees int64 +} + +func (p *process) eventLoop(ctx context.Context, cfg *config) error { + pendingHtlcs := make(map[route.Vertex]*peerInfo) + + intervalNs := int64(cfg.ReportingInterval) + nextReport := time.Unix(0, (time.Now().UnixNano()/intervalNs+1)* + intervalNs) + + log.Infow("First hold fees report scheduled", "next_report_time", nextReport) + for { + timeToReport := nextReport.Sub(time.Now()) + select { case interceptEvent := <-p.interceptChan: peer, err := p.getPubKey(interceptEvent.channel) @@ -102,9 +147,13 @@ func (p *process) eventLoop(cfg *config) error { return err } + alias := p.getNodeAlias(peer) + pending, ok := pendingHtlcs[peer] if !ok { - pending = make(map[circuitKey]struct{}) + pending = &peerInfo{ + htlcs: make(map[circuitKey]*holdInfo), + } pendingHtlcs[peer] = pending } @@ -113,27 +162,32 @@ func (p *process) eventLoop(cfg *config) error { maxPending = cfg.MaxPendingHtlcs } - if len(pending) >= maxPending { - log.Info("Rejecting htlc", - zap.Uint64("channel", interceptEvent.channel), - zap.Uint64("htlc", interceptEvent.htlc), - zap.String("peer", peer.String()), - zap.Int("pending_htlcs", len(pending)), - zap.Int("max_pending_htlcs", maxPending), + if len(pending.htlcs) >= maxPending { + log.Infow("Rejecting htlc", + "channel", interceptEvent.channel, + "htlc", interceptEvent.htlc, + "peer_alias", alias, + "peer", peer.String(), + "pending_htlcs", len(pending.htlcs), + "max_pending_htlcs", maxPending, ) interceptEvent.resume <- false continue } - pending[interceptEvent.circuitKey] = struct{}{} + pending.htlcs[interceptEvent.circuitKey] = &holdInfo{ + fwdTime: time.Now(), + valueMsat: interceptEvent.valueMsat, + } - log.Info("Forwarding htlc", - zap.Uint64("channel", interceptEvent.channel), - zap.Uint64("htlc", interceptEvent.htlc), - zap.String("peer", peer.String()), - zap.Int("pending_htlcs", len(pending)), - zap.Int("max_pending_htlcs", maxPending), + log.Infow("Forwarding htlc", + "channel", interceptEvent.channel, + "htlc", interceptEvent.htlc, + "peer_alias", alias, + "peer", peer.String(), + "pending_htlcs", len(pending.htlcs), + "max_pending_htlcs", maxPending, ) interceptEvent.resume <- true @@ -149,17 +203,66 @@ func (p *process) eventLoop(cfg *config) error { continue } - if _, ok := pending[resolvedKey]; !ok { + info, ok := pending.htlcs[resolvedKey] + if !ok { continue } - delete(pending, resolvedKey) + delete(pending.htlcs, resolvedKey) - log.Info("Resolving htlc", - zap.Uint64("channel", resolvedKey.channel), - zap.Uint64("htlc", resolvedKey.htlc), - zap.String("peer", peer.String()), - zap.Int("pending_htlcs", len(pending))) + holdTime := time.Since(info.fwdTime) + + holdFeeMsat := int64((1000*float64(cfg.BaseSatPerHr) + + float64(info.valueMsat)*float64(cfg.RatePpmPerHr)/1e6) * + holdTime.Hours()) + + pending.totalHoldFees += holdFeeMsat + pending.intervalHoldFees += holdFeeMsat + + log.Infow("Resolving htlc", + "channel", resolvedKey.channel, + "htlc", resolvedKey.htlc, + "peer_alias", p.getNodeAlias(peer), + "peer", peer.String(), + "pending_htlcs", len(pending.htlcs), + "hold_time", holdTime, + "hold_fee_msat", holdFeeMsat) + + case <-time.After(timeToReport): + changedPeers := []route.Vertex{} + for key, info := range pendingHtlcs { + if info.intervalHoldFees > 0 { + changedPeers = append( + changedPeers, key, + ) + } + } + + nextReport = nextReport.Add(cfg.ReportingInterval) + + if len(changedPeers) == 0 { + log.Infow("No hold fees to report", + "next_report_time", nextReport) + } else { + log.Infow("Hold fees report", + "next_report_time", nextReport) + + for _, key := range changedPeers { + log.Infow("Report", + "peer_alias", p.getNodeAlias(key), + "peer", key, + "total_fees_msat", pendingHtlcs[key].totalHoldFees, + "interval_fees_msat", pendingHtlcs[key].intervalHoldFees, + ) + + pendingHtlcs[key].intervalHoldFees = 0 + } + } + + case <-ctx.Done(): + log.Info("Exit") + + return nil } } } @@ -205,7 +308,8 @@ func (p *process) processInterceptor(interceptor routerrpc.Router_HtlcIntercepto channel: event.IncomingCircuitKey.ChanId, htlc: event.IncomingCircuitKey.HtlcId, }, - resume: resumeChan, + valueMsat: int64(event.OutgoingAmountMsat), + resume: resumeChan, } resume, ok := <-resumeChan @@ -229,6 +333,25 @@ func (p *process) processInterceptor(interceptor routerrpc.Router_HtlcIntercepto } } +func (p *process) getNodeAlias(key route.Vertex) string { + alias, ok := p.aliasMap[key] + if ok { + return alias + } + + alias, err := p.client.getNodeAlias(key) + if err != nil { + log.Warnw("cannot get node alias", + "err", err) + + return "" + } + + p.aliasMap[key] = alias + + return alias +} + func (p *process) getPubKey(channel uint64) (route.Vertex, error) { pubkey, ok := p.pubkeyMap[channel] if ok { diff --git a/process_test.go b/process_test.go new file mode 100644 index 0000000..e55c6ab --- /dev/null +++ b/process_test.go @@ -0,0 +1,53 @@ +package main + +import ( + "context" + "testing" + "time" + + "github.com/lightningnetwork/lnd/lnrpc/routerrpc" + "github.com/stretchr/testify/require" +) + +func TestProcess(t *testing.T) { + p := newProcess() + + cfg := &config{ + MaxPendingHtlcs: 2, + BaseSatPerHr: 1, + RatePpmPerHr: 5, + ReportingInterval: time.Minute * 1, + } + + client := newLndclientMock() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + exit := make(chan error) + go func() { + exit <- p.run(ctx, client, cfg) + }() + + key := &routerrpc.CircuitKey{ + ChanId: 2, + HtlcId: 5, + } + client.htlcInterceptorRequests <- &routerrpc.ForwardHtlcInterceptRequest{ + IncomingCircuitKey: key, + } + + resp := <-client.htlcInterceptorResponses + require.Equal(t, routerrpc.ResolveHoldForwardAction_RESUME, resp.Action) + + client.htlcEvents <- &routerrpc.HtlcEvent{ + EventType: routerrpc.HtlcEvent_FORWARD, + IncomingChannelId: key.ChanId, + IncomingHtlcId: key.HtlcId, + Event: &routerrpc.HtlcEvent_SettleEvent{}, + } + + time.Sleep(time.Second) + + cancel() + require.NoError(t, <-exit) +}