faraday: merge ForwardingAbility feature branch#244
Conversation
This bump adds functionality for being able to subscribe to channel update events.
mod: use chan events lnd and lndclient
Move the gRPC/REST server lifecycle, macaroon service management, and TLS setup from frdrpcserver.RPCServer to a new Faraday struct in the faraday package. This leaves RPCServer as a pure RPC request handler with only business logic methods. The Faraday struct embeds *frdrpcserver.RPCServer and takes ownership of all infrastructure concerns: gRPC/REST server creation, macaroon service setup, TLS configuration, and start/stop lifecycle. The frdrpcserver.Config is slimmed down to just Lnd and BitcoinClient. The macaroons.go file is also moved from frdrpcserver to the faraday package, as the macaroon constants are now used by the Faraday struct directly.
Refine the Faraday struct's lifecycle management: - Use atomic.Bool for cleaner start/stop guards with CompareAndSwap instead of the old atomic.AddInt32 pattern. - Extract initialize() to consolidate macaroon and bitcoin client setup shared between Start() and StartAsSubserver(). - Extract startRPCServer()/stopRPCServer() from the monolithic Start()/Stop() methods for better separation of concerns. - Use the lndOwned flag so Stop() only closes the lnd connection in standalone mode, leaving it open for the parent in subserver mode. - Move lnd connection and bitcoin client creation from Main() into the Faraday struct so callers only need New() and Start(). - Mark the struct as permanently stopped on Start failure to prevent retry with stale internal state. Guard startRPCServer against a nil macaroon service. - Reorder Stop() to wait for in-flight RPC goroutines before tearing down the macaroon service.
faraday+frdrpcserver: make faraday an active component
db: introduce sqldbv2 scaffolding
db: implement sql tables and store for channel events
This was forgotten previously.
Introduce requireEqualEvent to reduce boilerplate in store tests, comparing all user-set fields while ignoring the auto-assigned ID.
We want to know if an update came from an initial sync. This also helps us to identify data gaps and one can be sure it was not due to an actual event. Modify the migration as it's unreleased.
Also refactor the root logger.
chanevents+faraday: add chan events store and start
Logs every AddChannelEvent at trace level so operators can confirm which lnd events are being persisted.
Switches the GetChannelEvents query from a timestamp-ordered scan to an id-keyset cursor (WHERE id > $cursor ORDER BY id ASC LIMIT $n). The keyset cursor is stable under concurrent inserts and survives a future retention job that prunes the oldest rows: a positional OFFSET would silently skip events whenever rows below the cursor are deleted, while "id > $cursor" keeps advancing past whatever the caller has already seen. The id field is documented in the proto as a server-assigned monotonic identity, so callers persist last_id as their sync watermark. Adds a (channel_id, id) composite index to back the new query; the existing (channel_id, timestamp) index does not cover it and would force a per-channel filter after a global id scan.
Adds the GetChannelEvents RPC to the proto and regenerates the gRPC, gateway, swagger, and JSON stubs. The request carries a chan_point, inclusive start_time and exclusive end_time bounds, a max_events cap, and a last_id keyset cursor; the response echoes last_id and a has_more flag so callers can drive pagination without server-side state.
Threads the chanevents.Store the daemon already constructs into the frdrpcserver.Config so the upcoming GetChannelEvents handler has a read path. Both standalone and subserver startup wire the same store instance.
Implements the handler against the chanevents store. A zero end_time defaults to the server's current wall clock so callers can omit it for "up to now" queries. max_events is clamped to a 10000-row hard cap that also serves as the implicit default when the caller leaves the field at zero. An unknown chan_point maps to NotFound; negative time bounds and start_time after end_time map to InvalidArgument. The response sets has_more whenever the page filled to the requested limit so the client knows to keep paginating. Also registers the new endpoint in the macaroon permissions table under the channels:read entitlement.
Drives a regtest channel through open, payment, and force-close and asserts that GetChannelEvents surfaces the expected mix of online, offline, and balance-update events. The same window is then walked with a small page size to verify last_id round-trip, has_more termination, and MaxEvents clamping in one pass.
Adds a `chanevents` subcommand that wraps GetChannelEvents and exposes chan_point, start/end time, max_events, and last_id flags. The help text documents the manual pagination contract: keep re-running with --last_id set to the previous response's last_id until has_more is false, while leaving --start_time and --end_time fixed across calls.
Add GetChannelByShortChanID, the inverse of AddChannel. The forwarding-ability analyzer receives scids from lnd's forwarding history and must map them back to the chanevents store's internal channel id to query events.
Add Quantile, a generic linear-interpolation q-quantile over a slice of sortable numeric values. The forwarding-ability analyzer needs to characterise the distribution of historical forwarded amounts and uses a configurable percentile as the headline statistic; lifting the computation into its own helper keeps the analyzer focused on forwarding logic and gives the quantile contract its own table-driven test that covers the interpolation rule and the empty/out-of-bounds error paths.
forwarding ability: introduce db helpers
The per-pair uptime walk derives forwarding ability for both (A→B) and (B→A) directions in a single chronological pass over the merged event stream. Two independent state copies, one rooted at each peer, accumulate each direction's uptime against its own balance threshold. A liquidity check that would otherwise be O(channels-per-peer) per tick becomes O(1) via four running balance sums that shadow each side's online inbound and outbound capacity. The merge loop adjusts these sums as events mutate channel state, and the per-tick threshold check reads the mins inline.
Establish the boundary between the chanevents store and the upcoming forwarding analyzer plus the seed walk every pair calculation depends on. EventsSource is the read surface the analyzer consumes, and ForwardingAnalyzer carries it alongside an lnd handle so the upcoming driver can fold lnd's open and closed channel sets into the considered population. getInitialChannelState reconstructs a channel's state at startTime by seeding from the latest pre-window update and replaying any residual same-timestamp siblings the SQL keyset may have surfaced. The residual range is bounded by definition (events between two adjacent timestamps within a channel), so streaming would buy nothing over materialising the slice in one call. A guard error flags later-timestamp updates in the residual walk as schema drift.
Wire EffectiveUptime as the analyzer's public entry. The pipeline resolves channels to peers from the store, folds lnd's forwarding history into per-pair success amounts, augments the considered set with lnd's open and closed channels to hedge survivorship bias, seeds each channel's state at startTime, and dispatches every peer pair to the bidirectional walk. calculateAllPairsUptime walks the unordered cross-product (i, j>=i) with a lazy per-peer event cache. Each peer's events are fetched once and replayed across every pair that consumes them. The cached events live as a slice so the cross-pair merge can use a two-pointer walk instead of an iter-based merge that would need goroutine plus channel synchronisation per event.
Promote the existing test-store and test-DB constructors from *testing.T to testing.TB so the upcoming EffectiveUptime benchmark can share the same fixture path as the existing tests. testing.TB is the shared interface of *testing.T and *testing.B, so every current caller keeps type-checking unchanged.
forwarding ability: add main algo
loadPeerEvents fetched an entire channel's events in a single math.MaxInt32-limited query. Page through the store in id-ascending batches so no single query is unbounded, and document the paging protocol on the EventsSource contract.
Add the ForwardingAbility RPC that reports, for every (peerIn, peerOut) pair, the raw forwarding facts over a window: effective uptime in seconds and forwarded volume in satoshis. The response is a sparse, packed encoding (deduplicated peer keys plus packed pair indices) carrying only pairs with a non-zero signal; an absent pair means zero over the window. A single liquidity_floor_sat request parameter sets the directional liquidity a pair must hold to count as economically forwardable.
Fetch lnd's forwarding history in paginated batches rather than a single call, so the analysis is not silently truncated at lnd's default page size. The stub forwarding client returns an empty page for non-zero offsets so the pagination loop terminates in tests.
Extend the forwarding analyzer to produce a ForwardingAbility per peer pair: the effective uptime (time the pair held at least the liquidity floor of directional forwardable liquidity) and the total forwarded amount, both as raw facts with no derived rates or categories. A single uniform floor is applied to every pair so effective uptime is comparable across pairs, and forwarded volume is reported even when uptime is zero so consumers keep the demand signal. The percentile threshold model and its quantile helper are removed in favour of the single floor.
Encode the per-pair forwarding abilities into the packed, deduplicated wire form and decode them back. Only pairs with non-zero effective uptime or forwarded volume are emitted; the window the metrics cover is carried on the response so consumers can derive uptime fraction and velocity themselves.
Add the frcli forwardingability command to query the RPC over a time range and optional liquidity floor. It decodes the sparse response and prints each pair with its raw effective uptime and forwarded volume, plus uptime fraction and velocity derived from the reported window.
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.
rpc: add sparse `ForwardingAbility` endpoint
Summary of ChangesHello, I'm Gemini Code Assist1! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed! This pull request merges the Highlights
New Features🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console. Ignored Files
Using Gemini Code AssistThe full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips. Invoking Gemini You can request assistance from Gemini at any point by creating a comment using either
Customization To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new chanevents package to monitor, store, and analyze LND channel events (online, offline, and balance updates) using SQLite or Postgres. It adds a ForwardingAnalyzer to compute forwarding velocity and effective uptime, a background Monitor to track live events and perform initial syncs, and exposes this functionality via new gRPC/REST endpoints (GetChannelEvents and ForwardingAbility) and CLI commands. The review feedback is highly constructive, suggesting memory optimizations in forwarding history pagination, avoiding redundant event processing for self-pairs, improving error handling for zero-duration query windows, making the sqlc generation script more robust, and preventing redundant database sync events on daemon restarts.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| 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(events), | ||
| ), | ||
| ) | ||
|
|
||
| channelPeersConsidered := make(map[uint64]string) | ||
| successfulForwards := make(map[PeerPair][]btcutil.Amount) | ||
| for _, fwd := range events { | ||
| inPeer, ok := scidToPeer[fwd.ChannelIn] | ||
| if !ok { | ||
| log.WarnS( | ||
| ctx, "Could not find peer for incoming channel", | ||
| nil, slog.Uint64("channelIn", fwd.ChannelIn), | ||
| ) | ||
| continue | ||
| } | ||
|
|
||
| outPeer, ok := scidToPeer[fwd.ChannelOut] | ||
| if !ok { | ||
| log.WarnS( | ||
| ctx, "Could not find peer for outgoing channel", | ||
| nil, slog.Uint64("channelOut", fwd.ChannelOut), | ||
| ) | ||
| continue | ||
| } | ||
|
|
||
| channelPeersConsidered[fwd.ChannelIn] = inPeer | ||
| channelPeersConsidered[fwd.ChannelOut] = outPeer | ||
|
|
||
| pair := PeerPair{ | ||
| PeerIn: inPeer, | ||
| PeerOut: outPeer, | ||
| } | ||
|
|
||
| amt := fwd.AmountMsatOut.ToSatoshis() | ||
| successfulForwards[pair] = append(successfulForwards[pair], amt) | ||
| } | ||
|
|
||
| return successfulForwards, channelPeersConsidered, nil |
There was a problem hiding this comment.
In getForwardingData, the events slice accumulates all forwarding events in memory before processing them. For routing nodes with millions of forwarding events, this can lead to high memory consumption or Out-Of-Memory (OOM) crashes. Since the events are only used to populate channelPeersConsidered and successfulForwards, you can process the events page-by-page inside the pagination loop and avoid allocating the events slice entirely.
var offset uint32
const forwardingPageSize = 1000
channelPeersConsidered := make(map[uint64]string)
successfulForwards := make(map[PeerPair][]btcutil.Amount)
var totalEvents int
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
}
totalEvents += len(fwds.Events)
for _, fwd := range fwds.Events {
inPeer, ok := scidToPeer[fwd.ChannelIn]
if !ok {
log.WarnS(
ctx, "Could not find peer for incoming channel",
nil, slog.Uint64("channelIn", fwd.ChannelIn),
)
continue
}
outPeer, ok := scidToPeer[fwd.ChannelOut]
if !ok {
log.WarnS(
ctx, "Could not find peer for outgoing channel",
nil, slog.Uint64("channelOut", fwd.ChannelOut),
)
continue
}
channelPeersConsidered[fwd.ChannelIn] = inPeer
channelPeersConsidered[fwd.ChannelOut] = outPeer
pair := PeerPair{
PeerIn: inPeer,
PeerOut: outPeer,
}
amt := fwd.AmountMsatOut.ToSatoshis()
successfulForwards[pair] = append(successfulForwards[pair], amt)
}
if len(fwds.Events) < forwardingPageSize {
break
}
if fwds.LastIndexOffset <= offset {
break
}
offset = fwds.LastIndexOffset
}
log.DebugS(
ctx, "Found forwarding events",
slog.Int("count", totalEvents),
)
return successfulForwards, channelPeersConsidered, nilThere was a problem hiding this comment.
Good point, but this is an optimization, which can be fixed later. I want to minimize changes here.
| sliceB := sliceA | ||
| if i != j { | ||
| sliceB, err = loadPeer(peerB) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| } |
There was a problem hiding this comment.
When i == j (self-pair), passing sliceA as both arguments to mergeEventSlices causes every event to be yielded and processed twice. Since statesA and statesB are copies of the same map, a single pass of the events is sufficient to update both states. Passing nil for sliceB when i == j avoids this redundant processing.
| sliceB := sliceA | |
| if i != j { | |
| sliceB, err = loadPeer(peerB) | |
| if err != nil { | |
| return nil, err | |
| } | |
| } | |
| var sliceB []*ChannelEvent | |
| if i != j { | |
| sliceB, err = loadPeer(peerB) | |
| if err != nil { | |
| return nil, err | |
| } | |
| } |
There was a problem hiding this comment.
Optimization, can be addressed later (minimizing diffs here).
| if startTime.After(endTime) { | ||
| return nil, status.Error( | ||
| codes.InvalidArgument, | ||
| "start_time must be less than or equal to end_time", | ||
| ) | ||
| } |
There was a problem hiding this comment.
If startTime == endTime, the zero-duration window is technically allowed by the check startTime.After(endTime), but it will always fail the node-down guard later with codes.FailedPrecondition ("no peer pair met the uptime threshold..."). This is misleading because the failure is due to an invalid/empty query window, not because the node was offline. It is better to reject a zero-duration window with codes.InvalidArgument up front.
| if startTime.After(endTime) { | |
| return nil, status.Error( | |
| codes.InvalidArgument, | |
| "start_time must be less than or equal to end_time", | |
| ) | |
| } | |
| if !startTime.Before(endTime) { | |
| return nil, status.Error( | |
| codes.InvalidArgument, | |
| "start_time must be strictly less than end_time", | |
| ) | |
| } |
There was a problem hiding this comment.
Good point, will fix that in a later PR.
| for file in db/sqlc/migrations/*.up.sql.bak; do | ||
| mv "$file" "${file%.bak}" | ||
| done |
There was a problem hiding this comment.
If sed fails or no .up.sql.bak files are created, the glob db/sqlc/migrations/*.up.sql.bak will not expand, and the mv command will fail with an error, causing the script to exit prematurely (due to set -e). It is safer to check if the file exists before attempting to move it.
| for file in db/sqlc/migrations/*.up.sql.bak; do | |
| mv "$file" "${file%.bak}" | |
| done | |
| if [ -f "$file" ]; then | |
| mv "$file" "${file%.bak}" | |
| fi |
There was a problem hiding this comment.
Low prio, fixing later.
| // latest state is captured as a baseline. Events recorded during initial sync | ||
| // are marked with IsSync=true to distinguish them from real-time events | ||
| // received via the subscription. | ||
| func (m *Monitor) initialSync(ctx context.Context) error { |
There was a problem hiding this comment.
To prevent the channel_events table from being flooded with redundant IsSync events on every daemon restart, consider checking the latest recorded event for each channel before inserting a new sync event. If the channel's active status and balances haven't changed since the last recorded event, the insertion can be safely skipped.
There was a problem hiding this comment.
Tracked in #238. I'm linking to the review findings here as well in that issue.
|
I'm updating to current lndclient and lnd and will add a new commit for that. |
Update the Go toolchain version across CI, the Dockerfiles and the go.mod go directives from 1.25.5 to 1.25.10.
Bump the compile-time lnd dependency to v0.21.0-beta and lndclient to v0.21.0-1 to not rely on commit hash versions.
Bump the integration test lnd binary to v0.21.0-beta. Two test fixes are required for the new version: - nodereport: the anchor commitment close fee changed by 10 sat, update the hardcoded CHANNEL_CLOSE_FEE expectation from 4535 to 4525 sat. - test_context: closeChannel mined a block while still waiting for the pending close update, which confirmed the force close tx out of the mempool before its fee could be read, causing a 'Transaction not in mempool' failure. Only start mining once the close fee has been captured from the mempool.
ViktorT-11
left a comment
ViktorT-11
left a comment
There was a problem hiding this comment.
ACK on the merge! Is the plan to get a new faraday release out before the next official litd release?
I think we'd want to at least add some background cleaner to prevent the database from filling indefinitely. Only then I'd be fine with releasing it (and to include it in litd). Not a lot of work for that though. |
3 participants
Merges the
ForwardingAbilityfeature branch. This contains the pure reviewed work from the feature branch without a master rebase.