Add MonitorRebalancesJob and related functionality for payment reconciliation

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";
     /// <summary>
     /// The interval in minutes for the SweepAllNodesWalletsJob to run. 
@@ -192,6 +193,13 @@ public class Constants
     /// </summary>
     public static int REBALANCE_MAX_PROBE_ROUTES_PER_AMOUNT = 6;
 
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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;

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;
+
+/// <summary>
+/// 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.
+/// </summary>
+[DisallowConcurrentExecution]
+public class MonitorRebalancesJob : IJob
+{
+    private readonly ILogger<MonitorRebalancesJob> _logger;
+    private readonly IRebalanceRepository _rebalanceRepository;
+    private readonly ILightningService _lightningService;
+    private readonly IAuditService _auditService;
+
+    public MonitorRebalancesJob(
+        ILogger<MonitorRebalancesJob> 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;
+}

src/Program.cs

@@ -328,6 +328,20 @@ public static async Task Main(string[] args)
                         .StartNow().WithCronSchedule(Constants.MONITOR_CHANNELS_CRON);
                 });
 
+                // MonitorRebalancesJob
+                q.AddJob<MonitorRebalancesJob>(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<MonitorSwapsJob>(opts =>
                 {

src/Services/LightningRouterService.cs

@@ -16,6 +16,7 @@ public interface ILightningRouterService
     public AsyncServerStreamingCall<HtlcEvent> SubscribeHtlcEvents(Node node, Router.RouterClient? client = null);
     public AsyncServerStreamingCall<Payment> SendPaymentV2(Node node, SendPaymentRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null);
     public Task<HTLCAttempt> SendToRouteV2Async(Node node, Routerrpc.SendToRouteRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null);
+    public AsyncServerStreamingCall<Payment> TrackPaymentV2(Node node, TrackPaymentRequest request, CancellationToken cancellationToken, Router.RouterClient? client = null);
 }
 
 public class LightningRouterService : ILightningRouterService
@@ -119,5 +120,16 @@ public async Task<HTLCAttempt> SendToRouteV2Async(Node node, Routerrpc.SendToRou
             new Metadata { { "macaroon", node.ChannelAdminMacaroon } },
             cancellationToken: cancellationToken);
     }
+
+    public AsyncServerStreamingCall<Payment> 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);
+    }
 }
 

src/Services/LightningService.cs

@@ -168,6 +168,15 @@ Task<ProbeResult> ProbeRouteAsync(Node node, long amountSats, long feeLimitMsat,
         Task<Payment> SendPaymentV2Async(Node node, string paymentRequest, long feeLimitMsat,
             ulong[]? outgoingChanIds, string? lastHopPubkeyHex, int timeoutSeconds, CancellationToken ct);
 
+        /// <summary>
+        /// Looks up the current state of a payment in LND by hash. Drains the
+        /// <c>Router.TrackPaymentV2</c> stream until a terminal Payment update arrives.
+        /// Returns <c>null</c> when LND reports the payment as unknown (NotFound) or when
+        /// the call errors out — callers should treat <c>null</c> as "no LND truth available"
+        /// rather than synthesising a Failed status. Intended for reconciliation jobs.
+        /// </summary>
+        Task<Payment?> TrackPaymentV2Async(Node node, byte[] paymentHash, CancellationToken ct);
+
         /// <summary>
         /// 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<Payment> SendPaymentV2Async(Node node, string paymentRequest,
             }
         }
 
+        public async Task<Payment?> 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<long?> GetLocalOutboundFeeRatePpmAsync(Node node, ulong chanId)
         {
             var edge = await _lightningClientService.GetChanInfo(node, chanId);

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)
         {