From 76633edc06f29b5047dfa857fabbedd9654172a8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20A=2EP?= <53834183+Jossec101@users.noreply.github.com> Date: Wed, 13 May 2026 13:36:59 +0200 Subject: [PATCH] Add MonitorRebalancesJob and related functionality for payment reconciliation - Introduced MonitorRebalancesJob to reconcile local Rebalance rows against LND's TrackPaymentV2. - Added MONITOR_REBALANCES_CRON constant for scheduling. - Implemented TrackPaymentV2Async in LightningService for payment state lookup. - Updated RebalanceService to persist payment hash for reconciliation. - Enhanced Constants with new configuration options for rebalancing. stack-info: PR: https://github.com/Elenpay/NodeGuard/pull/512, branch: Jossec101/stack/18 --- src/Helpers/Constants.cs | 13 ++ src/Jobs/MonitorRebalancesJob.cs | 181 +++++++++++++++++++++++++ src/Program.cs | 14 ++ src/Services/LightningRouterService.cs | 12 ++ src/Services/LightningService.cs | 49 +++++++ src/Services/RebalanceService.cs | 7 +- 6 files changed, 275 insertions(+), 1 deletion(-) create mode 100644 src/Jobs/MonitorRebalancesJob.cs diff --git a/src/Helpers/Constants.cs b/src/Helpers/Constants.cs index 1d21cedc..5518146d 100644 --- a/src/Helpers/Constants.cs +++ b/src/Helpers/Constants.cs @@ -60,6 +60,7 @@ public class Constants // Crons & Jobs public static readonly string MONITOR_WITHDRAWALS_CRON = "10 0/5 * * * ?"; public static readonly string MONITOR_CHANNELS_CRON = "0 0 */1 * * ?"; + public static string MONITOR_REBALANCES_CRON = "30 0/5 * * * ?"; public static readonly string JOB_RETRY_INTERVAL_LIST_IN_MINUTES = "1,2,5,10,20"; /// /// The interval in minutes for the SweepAllNodesWalletsJob to run. @@ -192,6 +193,13 @@ public class Constants /// public static int REBALANCE_MAX_PROBE_ROUTES_PER_AMOUNT = 6; + /// + /// How far back the MonitorRebalancesJob sweeps for already-terminal-but-possibly-wrong + /// rebalances. Caps the window during which a Failed/Timeout/NoRoute row can still be + /// corrected by LND truth — old rows are left alone. + /// + public static int REBALANCE_RECONCILE_TERMINAL_WINDOW_HOURS = 24; + public const string IsFrozenTag = "frozen"; public const string IsManuallyFrozenTag = "manually_frozen"; @@ -297,6 +305,8 @@ static Constants() MONITOR_CHANNELS_CRON = Environment.GetEnvironmentVariable("MONITOR_CHANNELS_CRON") ?? MONITOR_CHANNELS_CRON; + MONITOR_REBALANCES_CRON = Environment.GetEnvironmentVariable("MONITOR_REBALANCES_CRON") ?? MONITOR_REBALANCES_CRON; + JOB_RETRY_INTERVAL_LIST_IN_MINUTES = Environment.GetEnvironmentVariable("JOB_RETRY_INTERVAL_LIST_IN_MINUTES") ?? JOB_RETRY_INTERVAL_LIST_IN_MINUTES; var sweepIntervalMinutes = Environment.GetEnvironmentVariable("SWEEP_ALL_NODES_WALLETS_INTERVAL_MINUTES"); @@ -415,6 +425,9 @@ static Constants() var rebMaxProbeRoutes = Environment.GetEnvironmentVariable("REBALANCE_MAX_PROBE_ROUTES_PER_AMOUNT"); if (rebMaxProbeRoutes != null) REBALANCE_MAX_PROBE_ROUTES_PER_AMOUNT = int.Parse(rebMaxProbeRoutes); + var rebReconcileWindow = Environment.GetEnvironmentVariable("REBALANCE_RECONCILE_TERMINAL_WINDOW_HOURS"); + if (rebReconcileWindow != null) REBALANCE_RECONCILE_TERMINAL_WINDOW_HOURS = int.Parse(rebReconcileWindow); + // DB Initialization ALICE_PUBKEY = Environment.GetEnvironmentVariable("ALICE_PUBKEY") ?? ALICE_PUBKEY; ALICE_HOST = Environment.GetEnvironmentVariable("ALICE_HOST") ?? ALICE_HOST; diff --git a/src/Jobs/MonitorRebalancesJob.cs b/src/Jobs/MonitorRebalancesJob.cs new file mode 100644 index 00000000..9b790af1 --- /dev/null +++ b/src/Jobs/MonitorRebalancesJob.cs @@ -0,0 +1,181 @@ +/* + * NodeGuard + * Copyright (C) 2023 Elenpay + * + * This program is free software: you can redistribute it and/or modify + * it under the terms of the GNU Affero General Public License as published by + * the Free Software Foundation, either version 3 of the License, or + * (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU Affero General Public License for more details. + * + * You should have received a copy of the GNU Affero General Public License + * along with this program. If not, see http://www.gnu.org/licenses/. + * + */ + +using Lnrpc; +using NodeGuard.Data.Models; +using NodeGuard.Data.Repositories.Interfaces; +using NodeGuard.Helpers; +using NodeGuard.Services; +using Quartz; + +namespace NodeGuard.Jobs; + +/// +/// Reconciles local Rebalance rows against LND's TrackPaymentV2 truth. Closes the gap where +/// a Rebalance is left InFlight after a crash or stuck, or wrongly marked Failed when the stream was +/// cancelled but LND actually settled the payment. +/// +[DisallowConcurrentExecution] +public class MonitorRebalancesJob : IJob +{ + private readonly ILogger _logger; + private readonly IRebalanceRepository _rebalanceRepository; + private readonly ILightningService _lightningService; + private readonly IAuditService _auditService; + + public MonitorRebalancesJob( + ILogger logger, + IRebalanceRepository rebalanceRepository, + ILightningService lightningService, + IAuditService auditService) + { + _logger = logger; + _rebalanceRepository = rebalanceRepository; + _lightningService = lightningService; + _auditService = auditService; + } + + public async Task Execute(IJobExecutionContext context) + { + _logger.LogInformation("Starting {JobName}...", nameof(MonitorRebalancesJob)); + try + { + var window = TimeSpan.FromHours(Constants.REBALANCE_RECONCILE_TERMINAL_WINDOW_HOURS); + var rebalances = await _rebalanceRepository.GetReconcilable(window); + + foreach (var rebalance in rebalances) + { + try + { + await ReconcileAsync(rebalance, context.CancellationToken); + } + catch (Exception ex) + { + _logger.LogError(ex, + "Unexpected error reconciling rebalance {RebalanceId}. Monitoring will continue for other rebalances", + rebalance.Id); + } + } + } + catch (Exception e) + { + _logger.LogError(e, "Error on {JobName}", nameof(MonitorRebalancesJob)); + throw new JobExecutionException(e, false); + } + + _logger.LogInformation("{JobName} ended", nameof(MonitorRebalancesJob)); + } + + private async Task ReconcileAsync(Rebalance rebalance, CancellationToken ct) + { + if (string.IsNullOrEmpty(rebalance.PaymentHashHex)) + return; + + if (rebalance.Node == null) + { + _logger.LogWarning("Rebalance {RebalanceId} has no node loaded; skipping reconciliation", rebalance.Id); + return; + } + + byte[] paymentHash; + try + { + paymentHash = Convert.FromHexString(rebalance.PaymentHashHex); + } + catch (FormatException) + { + _logger.LogWarning("Rebalance {RebalanceId} has malformed PaymentHashHex; skipping reconciliation", + rebalance.Id); + return; + } + + var payment = await _lightningService.TrackPaymentV2Async(rebalance.Node, paymentHash, ct); + + if (payment == null) + { + // LND returned NotFound (or the call errored). Only act on the NotFound signal + // when our row is still in a non-terminal state: in that case LND never saw the + // payment, so the optimistic InFlight/Probing/Pending is wrong. For terminal rows + // we leave them alone — a transient track error shouldn't reopen a Failed row. + if (IsNonTerminal(rebalance.Status)) + { + var oldStatus = rebalance.Status; + rebalance.Status = RebalanceStatus.Failed; + _rebalanceRepository.Update(rebalance); + await _auditService.LogSystemAsync( + AuditActionType.RebalanceCompleted, + AuditEventType.Failure, + AuditObjectType.Rebalance, + rebalance.Id.ToString(), + new + { + Reason = "MonitorReconciliation", + Detail = "LND has no record of this payment hash", + OldStatus = oldStatus.ToString(), + NewStatus = rebalance.Status.ToString(), + }); + } + + return; + } + + if (IsNonTerminalLndStatus(payment.Status)) + { + // LND still in flight — nothing to reconcile yet. + return; + } + + var previousStatus = rebalance.Status; + RebalanceService.ApplyTerminalPayment(rebalance, payment); + + if (rebalance.Status == previousStatus) + return; + + _rebalanceRepository.Update(rebalance); + + var eventType = rebalance.Status == RebalanceStatus.Succeeded + ? AuditEventType.Success + : AuditEventType.Failure; + + await _auditService.LogSystemAsync( + AuditActionType.RebalanceCompleted, + eventType, + AuditObjectType.Rebalance, + rebalance.Id.ToString(), + new + { + Reason = "MonitorReconciliation", + OldStatus = previousStatus.ToString(), + NewStatus = rebalance.Status.ToString(), + LndPaymentStatus = payment.Status.ToString(), + LndFailureReason = payment.FailureReason.ToString(), + rebalance.FeePaidSats, + rebalance.EffectivePpm, + }); + } + + private static bool IsNonTerminal(RebalanceStatus status) => + status is RebalanceStatus.Pending + or RebalanceStatus.Probing + or RebalanceStatus.InFlight; + + private static bool IsNonTerminalLndStatus(Payment.Types.PaymentStatus status) => + status is Payment.Types.PaymentStatus.Initiated + or Payment.Types.PaymentStatus.InFlight; +} diff --git a/src/Program.cs b/src/Program.cs index 75756f61..838c0984 100644 --- a/src/Program.cs +++ b/src/Program.cs @@ -328,6 +328,20 @@ public static async Task Main(string[] args) .StartNow().WithCronSchedule(Constants.MONITOR_CHANNELS_CRON); }); + // MonitorRebalancesJob + q.AddJob(opts => + { + opts.DisallowConcurrentExecution(); + opts.WithIdentity(nameof(MonitorRebalancesJob)); + }); + + q.AddTrigger(opts => + { + opts.ForJob(nameof(MonitorRebalancesJob)) + .WithIdentity($"{nameof(MonitorRebalancesJob)}Trigger") + .StartNow().WithCronSchedule(Constants.MONITOR_REBALANCES_CRON); + }); + // Monitor Swaps Job q.AddJob(opts => { diff --git a/src/Services/LightningRouterService.cs b/src/Services/LightningRouterService.cs index 67d89c58..5d5a9c8d 100644 --- a/src/Services/LightningRouterService.cs +++ b/src/Services/LightningRouterService.cs @@ -16,6 +16,7 @@ public interface ILightningRouterService public AsyncServerStreamingCall SubscribeHtlcEvents(Node node, Router.RouterClient? client = null); public AsyncServerStreamingCall SendPaymentV2(Node node, SendPaymentRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null); public Task SendToRouteV2Async(Node node, Routerrpc.SendToRouteRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null); + public AsyncServerStreamingCall TrackPaymentV2(Node node, TrackPaymentRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null); } public class LightningRouterService : ILightningRouterService @@ -119,5 +120,16 @@ public async Task SendToRouteV2Async(Node node, Routerrpc.SendToRou new Metadata { { "macaroon", node.ChannelAdminMacaroon } }, cancellationToken: cancellationToken); } + + public AsyncServerStreamingCall TrackPaymentV2(Node node, TrackPaymentRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null) + { + CustomArgumentNullException.ThrowIfNull(node.ChannelAdminMacaroon, nameof(node.ChannelAdminMacaroon), "LND Macaroon for {NodeName} is not well configured", node.Name); + + client ??= GetRouterClient(node.Endpoint); + + return client.TrackPaymentV2(request, + new Metadata { { "macaroon", node.ChannelAdminMacaroon } }, + cancellationToken: cancellationToken); + } } diff --git a/src/Services/LightningService.cs b/src/Services/LightningService.cs index 4828058b..0a4c2c12 100644 --- a/src/Services/LightningService.cs +++ b/src/Services/LightningService.cs @@ -168,6 +168,15 @@ Task ProbeRouteAsync(Node node, long amountSats, long feeLimitMsat, Task SendPaymentV2Async(Node node, string paymentRequest, long feeLimitMsat, ulong[]? outgoingChanIds, string? lastHopPubkeyHex, int timeoutSeconds, CancellationToken ct); + /// + /// Looks up the current state of a payment in LND by hash. Drains the + /// Router.TrackPaymentV2 stream until a terminal Payment update arrives. + /// Returns null when LND reports the payment as unknown (NotFound) or when + /// the call errors out — callers should treat null as "no LND truth available" + /// rather than synthesising a Failed status. Intended for reconciliation jobs. + /// + Task TrackPaymentV2Async(Node node, byte[] paymentHash, CancellationToken ct); + /// /// Returns the local-outbound fee rate (ppm) of *some* channel with the given peer. /// Used to estimate "the fee we'd charge to forward through this peer" for ppm-cap @@ -1723,6 +1732,46 @@ public async Task SendPaymentV2Async(Node node, string paymentRequest, } } + public async Task TrackPaymentV2Async(Node node, byte[] paymentHash, CancellationToken ct) + { + if (paymentHash == null || paymentHash.Length == 0) + throw new ArgumentException("Payment hash must not be empty", nameof(paymentHash)); + + var request = new TrackPaymentRequest + { + PaymentHash = ByteString.CopyFrom(paymentHash), + NoInflightUpdates = true, + }; + + try + { + using var stream = _lightningRouterService.TrackPaymentV2(node, request, ct); + Payment? terminal = null; + await foreach (var update in stream.ResponseStream.ReadAllAsync(ct)) + { + terminal = update; + if (update.Status != Payment.Types.PaymentStatus.InFlight + && update.Status != Payment.Types.PaymentStatus.Initiated) + { + break; + } + } + + return terminal; + } + catch (RpcException e) when (e.StatusCode == StatusCode.NotFound) + { + // LND has no record of this payment hash — caller decides whether that means + // "never reached LND" (mark Failed) or "needs another lookup later". + return null; + } + catch (Exception e) + { + _logger.LogWarning(e, "TrackPaymentV2 failed for node {NodeId}", node.Id); + return null; + } + } + public async Task GetLocalOutboundFeeRatePpmAsync(Node node, ulong chanId) { var edge = await _lightningClientService.GetChanInfo(node, chanId); diff --git a/src/Services/RebalanceService.cs b/src/Services/RebalanceService.cs index eace6f2e..943f0054 100644 --- a/src/Services/RebalanceService.cs +++ b/src/Services/RebalanceService.cs @@ -195,6 +195,11 @@ await _auditService.LogAsync(AuditActionType.RebalanceCompleted, AuditEventType. return rebalance; } + // Persist the payment hash before sending so MonitorRebalancesJob can resolve + // the true outcome against LND if this process dies mid-stream. + rebalance.PaymentHashHex = Convert.ToHexString(invoice.RHash.ToByteArray()).ToLowerInvariant(); + _rebalanceRepository.Update(rebalance); + var feeLimitMsat = ComputeFeeLimitMsat(rebalance.SatsAmount, rebalance.MaxFeePct); rebalance.Status = RebalanceStatus.Probing; @@ -320,7 +325,7 @@ private static long ComputeFeeLimitMsat(long satsAmount, double maxFeePct) => (long)Math.Round(satsAmount * (decimal)maxFeePct * 10m, MidpointRounding.AwayFromZero); - private static void ApplyTerminalPayment(Rebalance rebalance, Payment payment) + internal static void ApplyTerminalPayment(Rebalance rebalance, Payment payment) { switch (payment.Status) {