diff --git a/src/Data/Repositories/ChannelFeeStateRepository.cs b/src/Data/Repositories/ChannelFeeStateRepository.cs index f20e4710..5aa54312 100644 --- a/src/Data/Repositories/ChannelFeeStateRepository.cs +++ b/src/Data/Repositories/ChannelFeeStateRepository.cs @@ -43,6 +43,22 @@ public ChannelFeeStateRepository(IDbContextFactory dbConte .FirstOrDefaultAsync(x => x.ChannelId == channelId); } + public async Task> GetByManagedNodePubKey(string managedNodePubKey) + { + await using var context = await _dbContextFactory.CreateDbContextAsync(); + + // ChannelFeeState carries no node pubkey; the owning node lives on ChannelRoutingState + // (1:1 with the same Channel), so filter through it. + var channelIds = context.ChannelRoutingStates + .Where(s => s.ManagedNodePubKey == managedNodePubKey) + .Select(s => s.ChannelId); + + return await context.ChannelFeeStates + .Include(x => x.Channel) + .Where(x => channelIds.Contains(x.ChannelId)) + .ToListAsync(); + } + public async Task UpsertByChannelId(ChannelFeeState state) { await using var context = await _dbContextFactory.CreateDbContextAsync(); diff --git a/src/Data/Repositories/Interfaces/IChannelFeeStateRepository.cs b/src/Data/Repositories/Interfaces/IChannelFeeStateRepository.cs index 6ceaaa57..6382a444 100644 --- a/src/Data/Repositories/Interfaces/IChannelFeeStateRepository.cs +++ b/src/Data/Repositories/Interfaces/IChannelFeeStateRepository.cs @@ -28,5 +28,12 @@ public interface IChannelFeeStateRepository { Task GetByChannelId(int channelId); + /// + /// All fee-state rows for channels owned by the given managed node (joined to + /// ChannelRoutingState, which holds the owning-node pubkey), with their Channel loaded. + /// Used by the fee engine to batch per-node state and to drive restore-on-disable. + /// + Task> GetByManagedNodePubKey(string managedNodePubKey); + Task UpsertByChannelId(ChannelFeeState state); } diff --git a/src/Data/Repositories/Interfaces/IRebalanceRepository.cs b/src/Data/Repositories/Interfaces/IRebalanceRepository.cs index f8cca49b..6110e34e 100644 --- a/src/Data/Repositories/Interfaces/IRebalanceRepository.cs +++ b/src/Data/Repositories/Interfaces/IRebalanceRepository.cs @@ -58,6 +58,14 @@ public interface IRebalanceRepository /// Task GetInFlightByNode(int nodeId); + /// + /// True when a non-terminal rebalance (Pending/InFlight) has this channel as its source. + /// The Phase 2 fee engine uses this to enforce the fee-vs-rebalance authority split: while + /// a rebalance is moving a channel's ratio, the fee engine must not react to that manufactured + /// signal. is the primary key. + /// + Task HasInFlightRebalanceBySourceChannel(int sourceChannelId); + /// /// Budget consumption for a node since (by CreationDatetime): /// non-terminal rows count MAX(FeePaidSats, ReservedFeeSats) so in-flight spend counts diff --git a/src/Data/Repositories/RebalanceRepository.cs b/src/Data/Repositories/RebalanceRepository.cs index 668ef226..bcca4a3a 100644 --- a/src/Data/Repositories/RebalanceRepository.cs +++ b/src/Data/Repositories/RebalanceRepository.cs @@ -148,6 +148,15 @@ public async Task GetInFlightByNode(int nodeId) && (r.Status == RebalanceStatus.Pending || r.Status == RebalanceStatus.InFlight)); } + public async Task HasInFlightRebalanceBySourceChannel(int sourceChannelId) + { + await using var context = await _dbContextFactory.CreateDbContextAsync(); + + return await context.Rebalances + .AnyAsync(r => r.SourceChannelId == sourceChannelId + && (r.Status == RebalanceStatus.Pending || r.Status == RebalanceStatus.InFlight)); + } + public async Task GetConsumedFeesSince(int nodeId, DateTimeOffset since) { await using var context = await _dbContextFactory.CreateDbContextAsync(); diff --git a/src/Jobs/ChannelFeeOptimizerJob.cs b/src/Jobs/ChannelFeeOptimizerJob.cs new file mode 100644 index 00000000..d476e6b1 --- /dev/null +++ b/src/Jobs/ChannelFeeOptimizerJob.cs @@ -0,0 +1,462 @@ +/* + * 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 Microsoft.EntityFrameworkCore; +using NodeGuard.Data; +using NodeGuard.Data.Models; +using NodeGuard.Data.Repositories.Interfaces; +using NodeGuard.Helpers; +using NodeGuard.Services; +using Quartz; +using Channel = NodeGuard.Data.Models.Channel; + +namespace NodeGuard.Jobs; + +/// +/// Phase 2 of the routing engine — the dynamic fee actuator. For every eligible, owned channel +/// on a node with dynamic fee management enabled it reads the Phase 1 signal +/// (), runs the pure control +/// law, and applies the resulting outbound/inbound ppm via LND — enforcing the fee-vs-rebalance +/// authority split, a per-channel throttle + circuit breaker, and a baseline snapshot for +/// restore-on-disable. Real LND writes happen only when neither the global nor the per-node +/// dry-run flag is set; everything is gated by the global ROUTING_ENGINE_ENABLED kill switch. +/// +[DisallowConcurrentExecution] +public class ChannelFeeOptimizerJob : IJob +{ + // Prioritization looks at recent organic (forwarding) revenue over this window. + private static readonly TimeSpan OrganicFeesWindow = TimeSpan.FromDays(7); + + private readonly ILogger _logger; + private readonly INodeRepository _nodeRepository; + private readonly IChannelRepository _channelRepository; + private readonly IChannelRoutingStateRepository _routingStateRepository; + private readonly IChannelFeeStateRepository _feeStateRepository; + private readonly IChannelFlowAnalyticsRepository _flowAnalyticsRepository; + private readonly IRebalanceRepository _rebalanceRepository; + private readonly IFeeOptimizerService _feeOptimizerService; + private readonly ILightningService _lightningService; + private readonly ILightningClientService _lightningClientService; + private readonly IDbContextFactory _dbContextFactory; + + public ChannelFeeOptimizerJob( + ILogger logger, + INodeRepository nodeRepository, + IChannelRepository channelRepository, + IChannelRoutingStateRepository routingStateRepository, + IChannelFeeStateRepository feeStateRepository, + IChannelFlowAnalyticsRepository flowAnalyticsRepository, + IRebalanceRepository rebalanceRepository, + IFeeOptimizerService feeOptimizerService, + ILightningService lightningService, + ILightningClientService lightningClientService, + IDbContextFactory dbContextFactory) + { + _logger = logger; + _nodeRepository = nodeRepository; + _channelRepository = channelRepository; + _routingStateRepository = routingStateRepository; + _feeStateRepository = feeStateRepository; + _flowAnalyticsRepository = flowAnalyticsRepository; + _rebalanceRepository = rebalanceRepository; + _feeOptimizerService = feeOptimizerService; + _lightningService = lightningService; + _lightningClientService = lightningClientService; + _dbContextFactory = dbContextFactory; + } + + public async Task Execute(IJobExecutionContext context) + { + // Global kill switch — checked before any work. + if (!Constants.ROUTING_ENGINE_ENABLED) + { + return; + } + + _logger.LogInformation("Starting {JobName}...", nameof(ChannelFeeOptimizerJob)); + + try + { + var tunables = FeeOptimizerTunables.FromConstants(); + var managedNodes = await _nodeRepository.GetAllManagedByNodeGuard(withDisabled: false); + + var openChannelsByChanId = new Dictionary(); + foreach (var channel in await _channelRepository.GetAll()) + { + if (channel.Status == Channel.ChannelStatus.Open && channel.ChanId != 0) + { + openChannelsByChanId[channel.ChanId] = channel; + } + } + + foreach (var managedNode in managedNodes) + { + try + { + if (managedNode.DynamicFeeManagementEnabled) + { + await OptimizeNode(managedNode, managedNodes, openChannelsByChanId, tunables); + } + else + { + // Node opted out (or just flipped off) — restore baselines / freeze, once. + await HandleDisabledNode(managedNode); + } + } + catch (Exception ex) + { + _logger.LogError(ex, "Error optimizing fees for node {NodeName} ({NodePubKey})", + managedNode.Name, managedNode.PubKey); + } + } + } + catch (Exception ex) + { + // Never surface an exception from Execute — the next cycle is our retry. + _logger.LogError(ex, "Error in {JobName}", nameof(ChannelFeeOptimizerJob)); + } + + _logger.LogInformation("{JobName} ended", nameof(ChannelFeeOptimizerJob)); + } + + private sealed class Candidate + { + public required Lnrpc.Channel LndChannel { get; init; } + public required Channel DbChannel { get; init; } + public required ChannelRoutingState RoutingState { get; init; } + public ChannelFeeState? FeeState { get; set; } + public long OrganicFeesMsat { get; set; } + public double Deviation { get; set; } + } + + private async Task OptimizeNode( + Node node, + IReadOnlyCollection managedNodes, + IReadOnlyDictionary openChannelsByChanId, + FeeOptimizerTunables tunables) + { + var listResp = await _lightningClientService.ListChannels(node); + if (listResp == null) + { + _logger.LogWarning("Skipping fee optimization for node {NodeName}: ListChannels unavailable", node.Name); + return; + } + + var routingStates = (await _routingStateRepository.GetByManagedNodePubKey(node.PubKey)) + .ToDictionary(s => s.ChannelId); + var feeStates = (await _feeStateRepository.GetByManagedNodePubKey(node.PubKey)) + .ToDictionary(s => s.ChannelId); + + var now = DateTimeOffset.UtcNow; + var organicSince = now - OrganicFeesWindow; + + // Eligibility filter (before prioritization). + var candidates = new List(); + foreach (var lndChannel in listResp.Channels) + { + if (!ChannelOwnershipHelper.IsOwnedByManagedNode(lndChannel, managedNodes)) continue; + if (!openChannelsByChanId.TryGetValue(lndChannel.ChanId, out var dbChannel)) continue; + if (!dbChannel.IsDynamicFeeEnabled) continue; + if (lndChannel.Capacity < Constants.ROUTING_ENGINE_FEE_MIN_CHANNEL_SIZE_SATS) continue; + if (!routingStates.TryGetValue(dbChannel.Id, out var routingState)) continue; // no signal yet + + feeStates.TryGetValue(dbChannel.Id, out var feeState); + if (feeState is { ConsecutiveFailures: > 0 } + && feeState.ConsecutiveFailures >= Constants.ROUTING_ENGINE_FEE_MAX_CONSECUTIVE_FAILURES) + { + _logger.LogDebug("Channel {ChanId} on {NodeName} is circuit-broken; toggle IsDynamicFeeEnabled to reset", + lndChannel.ChanId, node.Name); + continue; + } + + candidates.Add(new Candidate + { + LndChannel = lndChannel, + DbChannel = dbChannel, + RoutingState = routingState, + FeeState = feeState, + OrganicFeesMsat = await _flowAnalyticsRepository.GetOrganicFeesEarnedMsat(node.PubKey, lndChannel.ChanId, organicSince), + Deviation = routingState.EmaLocalRatio - routingState.TargetLocalRatio, + }); + } + + // Prioritize by recent organic revenue, then by how far off-target the channel is. + var prioritized = candidates + .OrderByDescending(c => c.OrganicFeesMsat) + .ThenByDescending(c => Math.Abs(c.Deviation)); + + var examined = 0; + foreach (var candidate in prioritized) + { + if (examined >= Constants.ROUTING_ENGINE_FEE_MAX_UPDATES_PER_RUN) break; + + // Authority split: never touch a channel the rebalancer is actively moving. + if (await _rebalanceRepository.HasInFlightRebalanceBySourceChannel(candidate.DbChannel.Id)) + { + _logger.LogDebug("Skipping channel {ChanId} on {NodeName}: in-flight rebalance owns it", + candidate.LndChannel.ChanId, node.Name); + continue; + } + + try + { + examined++; + await OptimizeChannel(node, candidate, tunables, now); + } + catch (Exception ex) + { + _logger.LogError(ex, "Error optimizing fees for channel {ChanId} on node {NodeName}", + candidate.LndChannel.ChanId, node.Name); + } + } + } + + private async Task OptimizeChannel(Node node, Candidate candidate, FeeOptimizerTunables tunables, DateTimeOffset now) + { + var (managedPolicy, _) = await _lightningService.GetChannelFeePolicy(candidate.LndChannel.ChanId, node); + if (managedPolicy == null) + { + _logger.LogWarning("Skipping channel {ChanId} on {NodeName}: current fee policy unavailable", + candidate.LndChannel.ChanId, node.Name); + return; + } + + var routingState = candidate.RoutingState; + var feeState = candidate.FeeState ?? new ChannelFeeState { ChannelId = candidate.DbChannel.Id }; + + // Snapshot the operator's pre-engine policy once, before any write, for restore-on-disable. + if (feeState.BaselineCapturedAt == null) + { + feeState.BaselineCapturedAt = now; + feeState.BaselineOutboundBaseFeeMsat = managedPolicy.FeeBaseMsat; + feeState.BaselineOutboundPpm = (uint)managedPolicy.FeeRateMilliMsat; + feeState.BaselineInboundBaseMsat = managedPolicy.InboundFeeBaseMsat; + feeState.BaselineInboundPpm = managedPolicy.InboundFeeRateMilliMsat; + } + + var decision = _feeOptimizerService.ComputeNextPolicy( + routingState.EmaLocalRatio, + routingState.TargetLocalRatio, + routingState.PeerFlowCategory, + feeState.LastAppliedOutboundPpm, + feeState.LastAppliedInboundPpm, + node.AllowPositiveInboundFees, + tunables); + + feeState.LastObservedRatio = routingState.EmaLocalRatio; + feeState.LastComputedTarget = routingState.TargetLocalRatio; + + if (decision.Action != FeeAction.Update) + { + _logger.LogDebug("Channel {ChanId} on {NodeName}: {Action} ({Reason})", + candidate.LndChannel.ChanId, node.Name, decision.Action, decision.Reason); + await _feeStateRepository.UpsertByChannelId(feeState); + return; + } + + // Throttle: at most one applied change per MIN_UPDATE_INTERVAL, surviving restarts. + if (feeState.LastFeeUpdateAt is { } last + && (now - last).TotalMinutes < Constants.ROUTING_ENGINE_FEE_MIN_UPDATE_INTERVAL_MINUTES) + { + _logger.LogDebug("Channel {ChanId} on {NodeName}: throttled (last update {Last})", + candidate.LndChannel.ChanId, node.Name, last); + await _feeStateRepository.UpsertByChannelId(feeState); + return; + } + + var baseFeeMsat = managedPolicy.FeeBaseMsat; + var timeLockDelta = managedPolicy.TimeLockDelta; + var inboundBaseMsat = managedPolicy.InboundFeeBaseMsat; // engine only modulates the ppm rates + var chanPoint = $"{candidate.DbChannel.FundingTx}:{candidate.DbChannel.FundingTxOutputIndex}"; + + // Effective dry-run = global master OR per-node — lets operators go live one node at a time. + var dryRun = Constants.ROUTING_ENGINE_DRY_RUN || node.RoutingEngineDryRun; + if (dryRun) + { + _logger.LogInformation( + "[DRY-RUN] {NodeName} chan {ChanId}: would set outbound {Outbound}ppm inbound {Inbound}ppm ({Reason})", + node.Name, candidate.LndChannel.ChanId, decision.OutboundPpm, decision.InboundPpm, decision.Reason); + await _feeStateRepository.UpsertByChannelId(feeState); + return; + } + + try + { + await _lightningService.SetChannelFeePolicy( + chanPoint, + node.PubKey, + baseFeeMsat, + decision.OutboundPpm, + timeLockDelta, + inboundBaseMsat, + decision.InboundPpm, + allowPositiveInboundFees: node.AllowPositiveInboundFees); + + feeState.LastAppliedOutboundBaseFeeMsat = baseFeeMsat; + feeState.LastAppliedOutboundPpm = decision.OutboundPpm; + feeState.LastAppliedInboundBaseMsat = inboundBaseMsat; + feeState.LastAppliedInboundPpm = decision.InboundPpm; + feeState.LastFeeUpdateAt = now; + feeState.ConsecutiveFailures = 0; + + _logger.LogInformation("{NodeName} chan {ChanId}: set outbound {Outbound}ppm inbound {Inbound}ppm ({Reason})", + node.Name, candidate.LndChannel.ChanId, decision.OutboundPpm, decision.InboundPpm, decision.Reason); + } + catch (Exception ex) + { + feeState.ConsecutiveFailures++; + _logger.LogError(ex, "Failed to set fee policy for channel {ChanId} on {NodeName} (failure {Count}/{Max})", + candidate.LndChannel.ChanId, node.Name, feeState.ConsecutiveFailures, + Constants.ROUTING_ENGINE_FEE_MAX_CONSECUTIVE_FAILURES); + + if (feeState.ConsecutiveFailures >= Constants.ROUTING_ENGINE_FEE_MAX_CONSECUTIVE_FAILURES) + { + await WriteAuditAsync(candidate.DbChannel, node, "circuit-broken", new + { + candidate.LndChannel.ChanId, + feeState.ConsecutiveFailures, + }); + _logger.LogWarning("Channel {ChanId} on {NodeName} circuit-broken after {Count} consecutive failures", + candidate.LndChannel.ChanId, node.Name, feeState.ConsecutiveFailures); + } + } + + await _feeStateRepository.UpsertByChannelId(feeState); + } + + /// + /// A managed node with dynamic fee management OFF. For each channel we previously touched + /// (baseline captured), either restore the operator's baseline policy once (default) or leave + /// the last-set fees frozen with an audit marker — then clear our snapshot so this is one-shot. + /// + private async Task HandleDisabledNode(Node node) + { + var toHandle = (await _feeStateRepository.GetByManagedNodePubKey(node.PubKey)) + .Where(fs => fs.BaselineCapturedAt != null) + .ToList(); + if (toHandle.Count == 0) return; + + var dryRun = Constants.ROUTING_ENGINE_DRY_RUN || node.RoutingEngineDryRun; + + foreach (var feeState in toHandle) + { + var dbChannel = feeState.Channel; + var chanPoint = $"{dbChannel.FundingTx}:{dbChannel.FundingTxOutputIndex}"; + + try + { + if (node.RestoreFeeBaselineOnDisable) + { + if (dryRun) + { + _logger.LogInformation( + "[DRY-RUN] {NodeName} chan {ChanId}: would restore baseline fees (engine disabled)", + node.Name, dbChannel.ChanId); + } + else + { + // Engine never changes the timelock, so restore fee rates against the live one. + var (managedPolicy, _) = await _lightningService.GetChannelFeePolicy(dbChannel.ChanId, node); + if (managedPolicy == null) + { + _logger.LogWarning( + "Deferring baseline restore for chan {ChanId} on {NodeName}: current policy unavailable", + dbChannel.ChanId, node.Name); + continue; // keep baseline, retry next cycle + } + + await _lightningService.SetChannelFeePolicy( + chanPoint, + node.PubKey, + feeState.BaselineOutboundBaseFeeMsat ?? managedPolicy.FeeBaseMsat, + feeState.BaselineOutboundPpm ?? (uint)managedPolicy.FeeRateMilliMsat, + managedPolicy.TimeLockDelta, + feeState.BaselineInboundBaseMsat ?? managedPolicy.InboundFeeBaseMsat, + feeState.BaselineInboundPpm ?? managedPolicy.InboundFeeRateMilliMsat, + allowPositiveInboundFees: true); + + await WriteAuditAsync(dbChannel, node, "baseline-restored", new + { + dbChannel.ChanId, + feeState.BaselineOutboundPpm, + feeState.BaselineInboundPpm, + }); + _logger.LogInformation("{NodeName} chan {ChanId}: restored baseline fees (engine disabled)", + node.Name, dbChannel.ChanId); + } + } + else + { + _logger.LogInformation("{NodeName} chan {ChanId}: engine disabled, fees frozen", node.Name, dbChannel.ChanId); + } + } + catch (Exception ex) + { + _logger.LogError(ex, "Error handling disabled-node fee state for chan {ChanId} on {NodeName}", + dbChannel.ChanId, node.Name); + continue; // keep baseline, retry next cycle + } + + // One-shot: clear our tracking. A later re-enable recaptures a fresh baseline. + feeState.BaselineCapturedAt = null; + feeState.BaselineOutboundBaseFeeMsat = null; + feeState.BaselineOutboundPpm = null; + feeState.BaselineInboundBaseMsat = null; + feeState.BaselineInboundPpm = null; + feeState.LastAppliedOutboundBaseFeeMsat = null; + feeState.LastAppliedOutboundPpm = null; + feeState.LastAppliedInboundBaseMsat = null; + feeState.LastAppliedInboundPpm = null; + feeState.LastFeeUpdateAt = null; + feeState.ConsecutiveFailures = 0; + await _feeStateRepository.UpsertByChannelId(feeState); + } + } + + private async Task WriteAuditAsync(Channel channel, Node node, string action, object extraDetails) + { + try + { + await using var context = await _dbContextFactory.CreateDbContextAsync(); + await context.AuditLogs.AddAsync(new AuditLog + { + ActionType = AuditActionType.Update, + EventType = AuditEventType.Success, + ObjectAffected = AuditObjectType.Channel, + ObjectId = channel.Id.ToString(), + Username = "SYSTEM", + Details = System.Text.Json.JsonSerializer.Serialize(new + { + EngineWrite = true, + RoutingEngineAction = action, + ChannelId = channel.Id, + NodeId = node.Id, + NodePubKey = node.PubKey, + Extra = extraDetails, + }), + }); + await context.SaveChangesAsync(); + } + catch (Exception ex) + { + _logger.LogError(ex, "Failed to write routing-engine audit log ({Action}) for channel {ChannelId}", + action, channel.Id); + } + } +} diff --git a/src/Program.cs b/src/Program.cs index c416c1d4..fcf3a677 100644 --- a/src/Program.cs +++ b/src/Program.cs @@ -299,6 +299,30 @@ public static async Task Main(string[] args) }); }); + //Channel Fee Optimizer Job (Routing Engine — Phase 2, dynamic fee actuator) + q.AddJob(opts => + { + opts.DisallowConcurrentExecution(); + opts.WithIdentity(nameof(ChannelFeeOptimizerJob)); + }); + + q.AddTrigger(opts => + { + opts.ForJob(nameof(ChannelFeeOptimizerJob)) + .WithIdentity($"{nameof(ChannelFeeOptimizerJob)}Trigger") + .StartNow().WithSimpleSchedule(scheduleBuilder => + { + if (Constants.IS_DEV_ENVIRONMENT) + { + scheduleBuilder.WithIntervalInMinutes(5).RepeatForever(); + } + else + { + scheduleBuilder.WithIntervalInMinutes(Constants.ROUTING_ENGINE_JOB_INTERVAL_MINUTES).RepeatForever(); + } + }); + }); + //Monitor Withdrawals Job q.AddJob(opts => { diff --git a/test/NodeGuard.Tests/Data/Repositories/RebalanceRepositoryRoutingEngineTests.cs b/test/NodeGuard.Tests/Data/Repositories/RebalanceRepositoryRoutingEngineTests.cs index 2fe24478..c1b09d63 100644 --- a/test/NodeGuard.Tests/Data/Repositories/RebalanceRepositoryRoutingEngineTests.cs +++ b/test/NodeGuard.Tests/Data/Repositories/RebalanceRepositoryRoutingEngineTests.cs @@ -78,6 +78,28 @@ public async Task GetInFlightByNode_CountsOnlyPendingAndInFlightForThatNode() (await sut.GetInFlightByNode(NodeId)).Should().Be(2); } + [Fact] + public async Task HasInFlightRebalanceBySourceChannel_TrueOnlyForPendingOrInFlightSource() + { + var (sut, seed) = SetupDb(); + var now = DateTimeOffset.UtcNow; + const int chanPending = 10; + const int chanInFlight = 20; + const int chanSettled = 30; + + seed.Rebalances.AddRange( + new Rebalance { NodeId = NodeId, SourceChannelId = chanPending, Status = RebalanceStatus.Pending, CreationDatetime = now, UpdateDatetime = now }, + new Rebalance { NodeId = NodeId, SourceChannelId = chanInFlight, Status = RebalanceStatus.InFlight, CreationDatetime = now, UpdateDatetime = now }, + new Rebalance { NodeId = NodeId, SourceChannelId = chanSettled, Status = RebalanceStatus.Succeeded, CreationDatetime = now, UpdateDatetime = now } + ); + await seed.SaveChangesAsync(); + + (await sut.HasInFlightRebalanceBySourceChannel(chanPending)).Should().BeTrue(); + (await sut.HasInFlightRebalanceBySourceChannel(chanInFlight)).Should().BeTrue(); + (await sut.HasInFlightRebalanceBySourceChannel(chanSettled)).Should().BeFalse(); + (await sut.HasInFlightRebalanceBySourceChannel(999)).Should().BeFalse(); // unknown channel + } + [Fact] public async Task GetConsumedFeesSince_UsesReservedOrPaidForInFlightAndPaidForSucceeded() { diff --git a/test/NodeGuard.Tests/Jobs/ChannelFeeOptimizerJobTests.cs b/test/NodeGuard.Tests/Jobs/ChannelFeeOptimizerJobTests.cs new file mode 100644 index 00000000..31166021 --- /dev/null +++ b/test/NodeGuard.Tests/Jobs/ChannelFeeOptimizerJobTests.cs @@ -0,0 +1,226 @@ +/* + * 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 FluentAssertions; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; +using NodeGuard.Data; +using NodeGuard.Data.Models; +using NodeGuard.Data.Repositories.Interfaces; +using NodeGuard.Helpers; +using NodeGuard.Services; +using Quartz; +using Channel = NodeGuard.Data.Models.Channel; + +namespace NodeGuard.Jobs; + +public class ChannelFeeOptimizerJobTests +{ + private const string NodePubKey = "managedPubKey"; + private const ulong ChanId = 123; + private const int ChannelDbId = 10; + + private readonly Mock _nodeRepository = new(); + private readonly Mock _channelRepository = new(); + private readonly Mock _routingStateRepository = new(); + private readonly Mock _feeStateRepository = new(); + private readonly Mock _flowAnalyticsRepository = new(); + private readonly Mock _rebalanceRepository = new(); + private readonly Mock _lightningService = new(); + private readonly Mock _lightningClientService = new(); + + private Node BuildNode() => new() + { + Id = 20, + PubKey = NodePubKey, + Name = "alice", + DynamicFeeManagementEnabled = true, + AllowPositiveInboundFees = true, + RoutingEngineDryRun = false, + }; + + private void ArrangeSingleSinkChannel(Node node, bool inFlightRebalance) + { + _nodeRepository.Setup(x => x.GetAllManagedByNodeGuard(false)).ReturnsAsync(new List { node }); + + var dbChannel = new Channel + { + Id = ChannelDbId, + ChanId = ChanId, + Status = Channel.ChannelStatus.Open, + IsDynamicFeeEnabled = true, + FundingTx = "txid123", + FundingTxOutputIndex = 1, + }; + _channelRepository.Setup(x => x.GetAll()).ReturnsAsync(new List { dbChannel }); + + // Too remote (ema 0.40 < target 0.50) + Sink → raise outbound, negative inbound. + _routingStateRepository.Setup(x => x.GetByManagedNodePubKey(NodePubKey)).ReturnsAsync(new List + { + new() + { + ChannelId = ChannelDbId, + ManagedNodePubKey = NodePubKey, + ChanIdLnd = ChanId, + EmaLocalRatio = 0.40, + TargetLocalRatio = 0.50, + PeerFlowCategory = PeerFlowCategory.Sink, + }, + }); + _feeStateRepository.Setup(x => x.GetByManagedNodePubKey(NodePubKey)).ReturnsAsync(new List()); + _flowAnalyticsRepository.Setup(x => x.GetOrganicFeesEarnedMsat(NodePubKey, ChanId, It.IsAny())).ReturnsAsync(1_000_000); + _rebalanceRepository.Setup(x => x.HasInFlightRebalanceBySourceChannel(ChannelDbId)).ReturnsAsync(inFlightRebalance); + + var listResp = new Lnrpc.ListChannelsResponse + { + Channels = + { + new Lnrpc.Channel + { + ChanId = ChanId, + Capacity = 16_000_000, + LocalBalance = 4_000_000, + RemoteBalance = 12_000_000, + Active = true, + Initiator = true, + RemotePubkey = "peerPubKey", + }, + }, + }; + _lightningClientService + .Setup(x => x.ListChannels(It.IsAny(), It.IsAny())) + .ReturnsAsync(listResp); + + _lightningService + .Setup(x => x.GetChannelFeePolicy(ChanId, It.IsAny())) + .ReturnsAsync((new Lnrpc.RoutingPolicy + { + FeeBaseMsat = 1000, + FeeRateMilliMsat = 500, + TimeLockDelta = 40, + InboundFeeBaseMsat = 0, + InboundFeeRateMilliMsat = 0, + }, (Lnrpc.RoutingPolicy?)null)); + } + + private ChannelFeeOptimizerJob BuildJob() + { + var options = new DbContextOptionsBuilder() + .UseInMemoryDatabase($"FeeJob_{Guid.NewGuid()}") + .Options; + var dbFactory = new Mock>(); + dbFactory.Setup(x => x.CreateDbContextAsync(default)).ReturnsAsync(() => new ApplicationDbContext(options)); + + return new ChannelFeeOptimizerJob( + new Mock>().Object, + _nodeRepository.Object, + _channelRepository.Object, + _routingStateRepository.Object, + _feeStateRepository.Object, + _flowAnalyticsRepository.Object, + _rebalanceRepository.Object, + new FeeOptimizerService(), // real pure decision logic + _lightningService.Object, + _lightningClientService.Object, + dbFactory.Object); + } + + private static async Task WithEngine(bool enabled, bool globalDryRun, Func body) + { + var prevEnabled = Constants.ROUTING_ENGINE_ENABLED; + var prevDryRun = Constants.ROUTING_ENGINE_DRY_RUN; + Constants.ROUTING_ENGINE_ENABLED = enabled; + Constants.ROUTING_ENGINE_DRY_RUN = globalDryRun; + try + { + await body(); + } + finally + { + Constants.ROUTING_ENGINE_ENABLED = prevEnabled; + Constants.ROUTING_ENGINE_DRY_RUN = prevDryRun; + } + } + + [Fact] + public async Task Execute_LiveNode_AppliesComputedPolicy() + { + var node = BuildNode(); + ArrangeSingleSinkChannel(node, inFlightRebalance: false); + + await WithEngine(enabled: true, globalDryRun: false, async () => + { + await BuildJob().Execute(Mock.Of()); + }); + + // Sink, d = -0.10, first eval seeds p_last=2500 → outbound 2550, inbound -25. + _lightningService.Verify(x => x.SetChannelFeePolicy( + "txid123:1", NodePubKey, 1000, 2550u, 40u, 0, -25, true), Times.Once); + } + + [Fact] + public async Task Execute_GlobalDryRun_DoesNotWriteToLnd() + { + var node = BuildNode(); + ArrangeSingleSinkChannel(node, inFlightRebalance: false); + + await WithEngine(enabled: true, globalDryRun: true, async () => + { + await BuildJob().Execute(Mock.Of()); + }); + + _lightningService.Verify(x => x.SetChannelFeePolicy( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + // Observability state is still persisted. + _feeStateRepository.Verify(x => x.UpsertByChannelId(It.IsAny()), Times.Once); + } + + [Fact] + public async Task Execute_InFlightRebalance_SkipsChannelEntirely() + { + var node = BuildNode(); + ArrangeSingleSinkChannel(node, inFlightRebalance: true); + + await WithEngine(enabled: true, globalDryRun: false, async () => + { + await BuildJob().Execute(Mock.Of()); + }); + + // Authority split: skipped before reading policy or writing. + _lightningService.Verify(x => x.GetChannelFeePolicy(It.IsAny(), It.IsAny()), Times.Never); + _lightningService.Verify(x => x.SetChannelFeePolicy( + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny(), + It.IsAny(), It.IsAny(), It.IsAny(), It.IsAny()), Times.Never); + } + + [Fact] + public async Task Execute_KillSwitchOff_DoesNothing() + { + var node = BuildNode(); + ArrangeSingleSinkChannel(node, inFlightRebalance: false); + + await WithEngine(enabled: false, globalDryRun: false, async () => + { + await BuildJob().Execute(Mock.Of()); + }); + + _nodeRepository.Verify(x => x.GetAllManagedByNodeGuard(It.IsAny()), Times.Never); + } +}