Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 19 additions & 14 deletions docs/spark-swaps.md

Large diffs are not rendered by default.

45 changes: 38 additions & 7 deletions src/Data/Models/Node.cs
Original file line number Diff line number Diff line change
Expand Up @@ -80,19 +80,50 @@ public class Node : Entity
public bool AutoLiquidityManagementEnabled { get; set; }

/// <summary>
/// Minimum swap amount in satoshis
/// Minimum Loop swap amount in satoshis
/// </summary>
public long SwapMinAmountSats { get; set; }
public long LoopSwapMinAmountSats { get; set; }

/// <summary>
/// Maximum swap amount in satoshis
/// Maximum Loop swap amount in satoshis
/// </summary>
public long SwapMaxAmountSats { get; set; }
public long LoopSwapMaxAmountSats { get; set; }

/// <summary>
/// Maximum number of concurrent swaps to limit HTLC locking
/// Minimum 40swap swap amount in satoshis
/// </summary>
public int MaxSwapsInFlight { get; set; }
public long FortySwapMinAmountSats { get; set; }

/// <summary>
/// Maximum 40swap swap amount in satoshis
/// </summary>
public long FortySwapMaxAmountSats { get; set; }

/// <summary>
/// Minimum Spark swap amount in satoshis; null means SPARK_SWAP_MIN_SATS
/// </summary>
public long? SparkSwapMinAmountSats { get; set; }

/// <summary>
/// Maximum Spark swap amount in satoshis; null means SPARK_SWAP_MAX_SATS
/// </summary>
public long? SparkSwapMaxAmountSats { get; set; }

/// <summary>
/// Maximum number of concurrent Loop swaps, to limit HTLC locking
/// </summary>
public int LoopMaxSwapsInFlight { get; set; }

/// <summary>
/// Maximum number of concurrent 40swap swaps, to limit HTLC locking
/// </summary>
public int FortySwapMaxSwapsInFlight { get; set; }

/// <summary>
/// Maximum number of Spark swaps being paid at once, to limit HTLC locking. Once paid, a Spark swap waits in
/// its Spark wallet, whose max balance limits it
/// </summary>
public int SparkMaxSwapsInFlight { get; set; }

/// <summary>
/// Maximum swap routing fee ratio acceptable for swaps as a decimal between 0 and 1
Expand Down Expand Up @@ -135,7 +166,7 @@ public class Node : Entity

/// <summary>
/// Weight for the Spark swap provider selection (default 0). Spark is skipped while the node's
/// Spark wallet is unavailable; MaxSwapsInFlight limits Spark swaps like any other
/// Spark wallet is unavailable; SparkMaxSwapsInFlight limits its swaps
/// </summary>
public int SparkSwapWeight { get; set; } = 0;

Expand Down
6 changes: 6 additions & 0 deletions src/Data/Models/Swaps.cs
Original file line number Diff line number Diff line change
Expand Up @@ -189,6 +189,12 @@ public class SwapOut : Entity
/// </summary>
public long? SparkReceivedSats { get; set; }

/// <summary>
/// Spark swaps only: when the swap's exit was first attempted, in a batch with the other paid swaps of its Spark
/// wallet once the wallet was full. Until then the swap waits in the wallet
/// </summary>
public DateTimeOffset? SparkExitDatetime { get; set; }

/// <summary>
/// What the swap paid on-chain to <see cref="DestinationAddress"/>, once confirmed: what it delivered.
/// Recorded for Spark swaps
Expand Down
6 changes: 5 additions & 1 deletion src/Data/Repositories/Interfaces/ISwapOutRepository.cs
Original file line number Diff line number Diff line change
Expand Up @@ -42,9 +42,13 @@ public interface ISwapOutRepository
Task<(bool, string?)> AddAsync(SwapOut swap);
Task<(bool, string?)> AddRangeAsync(List<SwapOut> swaps);
(bool, string?) Update(SwapOut swap);

/// <summary>Saves several swaps in one transaction: all of them, or none</summary>
(bool, string?) UpdateRange(IReadOnlyCollection<SwapOut> swaps);

/// <summary>
/// Get all swaps that are currently in flight (not completed, failed, or timed out) for a specific node, swaps older than 24 hours are ignored to avoid swaps deadlocks if many are stuck
/// Get all swaps that are currently in flight (not completed, failed, or timed out) for a specific node, swaps older than 24 hours are ignored to avoid swaps deadlocks if many are stuck.
/// A paid Spark swap is not in flight any more: it waits in its Spark wallet, whose max balance limits it
/// </summary>
Task<List<SwapOut>> GetInFlightSwapsByNode(int nodeId);

Expand Down
24 changes: 23 additions & 1 deletion src/Data/Repositories/SwapOutRepository.cs
Original file line number Diff line number Diff line change
Expand Up @@ -172,6 +172,27 @@ public async Task<List<SwapOut>> GetAllPending()
return _repository.Update(swap, context);
}

public (bool, string?) UpdateRange(IReadOnlyCollection<SwapOut> swaps)
{
using var context = _dbContextFactory.CreateDbContext();

foreach (var swap in swaps)
{
swap.SetUpdateDatetime();
}

try
{
// One SaveChanges: one transaction
context.UpdateRange(swaps);
return (context.SaveChanges() > 0, null);
}
catch (Exception e)
{
return (false, e.Message);
}
}

public async Task<List<SwapOut>> GetInFlightSwapsByNode(int nodeId)
{
await using var context = await _dbContextFactory.CreateDbContextAsync();
Expand All @@ -182,7 +203,8 @@ public async Task<List<SwapOut>> GetInFlightSwapsByNode(int nodeId)
var swaps = await context.SwapOuts
.Where(s => s.NodeId == nodeId &&
s.Status == SwapOutStatus.Pending &&
s.CreationDatetime >= twentyFourHoursAgo)
s.CreationDatetime >= twentyFourHoursAgo &&
!(s.Provider == SwapProvider.Spark && s.LightningFeeSats != null))
.ToListAsync();

return swaps;
Expand Down
150 changes: 83 additions & 67 deletions src/Jobs/AutoLiquidityManagementJob.cs
Original file line number Diff line number Diff line change
Expand Up @@ -120,16 +120,9 @@ public async Task Execute(IJobExecutionContext context)
_logger.LogInformation("{JobName} ended", nameof(AutoLiquidityManagementJob));
}

/// <summary>
/// Weighted random choice among the providers with a weight that can take a swap now. Spark is left
/// out while the node has no Spark wallet, or its wallet is unavailable or has less room under its max
/// balance than the node's minimum swap (the node's Max swaps in flight already limits its swaps); null when
/// no weighted provider is left. All weights at 0 means Loop, as before. Also returns the Spark wallet's room,
/// to clamp a Spark swap.
/// </summary>
private async Task<(SwapProvider? Provider, long SparkRoomSats)> SelectSwapProviderAsync(Node node)
/// <summary>The providers with a weight; all weights at 0 means Loop, as before.</summary>
private List<(SwapProvider Provider, int Weight)> WeightedProviders(Node node)
{
var sparkRoom = 0L;
var weights = new List<(SwapProvider Provider, int Weight)>
{
(SwapProvider.Loop, node.LoopSwapWeight),
Expand All @@ -140,53 +133,90 @@ public async Task Execute(IJobExecutionContext context)
if (weights.All(w => w.Weight <= 0))
{
_logger.LogWarning("Node {NodeName} has total weight of 0, defaulting to Loop", node.Name);
return (SwapProvider.Loop, sparkRoom);
return [(SwapProvider.Loop, 1)];
}

if (node.SparkSwapWeight > 0)
return weights.Where(w => w.Weight > 0).ToList();
}

/// <summary>
/// Weighted random choice among <paramref name="weights"/> (the weighted providers below their max swaps in
/// flight) that can take a swap now, and the swap it takes: the most it may, within its max swap, the excess
/// balance and, for Spark, the room left in the node's Spark wallet under its max balance. A provider is left out
/// when that is below its min swap, and Spark also while the node has no Spark wallet or its wallet is
/// unavailable. Null when no provider is left, with whether the excess balance was below every one's min swap.
/// </summary>
private async Task<(SwapProvider? Provider, long AmountSats, bool BelowMinimum)> SelectSwapProviderAsync(Node node,
IReadOnlyList<(SwapProvider Provider, int Weight)> weights, long excessBalance)
{
var candidates = new List<(SwapProvider Provider, int Weight, long AmountSats)>();
var belowMinimum = true;
foreach (var (provider, weight) in weights)
{
var (reason, room) = await SparkAvailabilityAsync(node);
sparkRoom = room;
if (reason is not null)
var limits = SwapProviderLimits.For(node, provider, _sparkSettings);
var amount = Math.Min(excessBalance, limits.MaxSats);
if (amount < limits.MinSats)
{
_logger.LogInformation("Node {NodeName}: Spark cannot take a swap now ({Reason}), choosing among the other providers",
node.Name, reason);
weights.RemoveAll(w => w.Provider == SwapProvider.Spark);
_logger.LogDebug("Node {NodeName}: {Provider} left out, the excess balance ({Excess} sats) is below its min swap ({Min} sats)",
node.Name, provider, excessBalance, limits.MinSats);
continue;
}

belowMinimum = false;
if (provider == SwapProvider.Spark)
{
var (reason, room) = await SparkAvailabilityAsync(node, limits.MinSats);
if (reason is not null)
{
_logger.LogInformation("Node {NodeName}: Spark cannot take a swap now ({Reason}), choosing among the other providers",
node.Name, reason);
continue;
}

// A Spark swap only takes what the node's Spark wallet can still hold under its max balance
if (amount > room)
{
_logger.LogInformation("Node {NodeName}: Spark swap clamped from {Amount} to {Room} sats by its Spark wallet's max balance",
node.Name, amount, room);
amount = room;
}
}

candidates.Add((provider, weight, amount));
}

var candidates = weights.Where(w => w.Weight > 0).ToList();
var totalWeight = candidates.Sum(w => w.Weight);
if (totalWeight == 0)
if (candidates.Count == 0)
{
return (null, sparkRoom);
return (null, 0, belowMinimum);
}

// Weighted random selection
var totalWeight = candidates.Sum(c => c.Weight);
var randomValue = Random.Shared.Next(0, totalWeight);
var selectedProvider = candidates[^1].Provider;
var selected = candidates[^1];
var cumulative = 0;
foreach (var (provider, weight) in candidates)
foreach (var candidate in candidates)
{
cumulative += weight;
cumulative += candidate.Weight;
if (randomValue < cumulative)
{
selectedProvider = provider;
selected = candidate;
break;
}
}

_logger.LogDebug("Selected {Provider} for node {NodeName} (Loop weight: {LoopWeight}, 40swap weight: {FortySwapWeight}, Spark weight: {SparkWeight}, random: {Random})",
selectedProvider, node.Name, node.LoopSwapWeight, node.FortySwapWeight, node.SparkSwapWeight, randomValue);
selected.Provider, node.Name, node.LoopSwapWeight, node.FortySwapWeight, node.SparkSwapWeight, randomValue);

return (selectedProvider, sparkRoom);
return (selected.Provider, selected.AmountSats, false);
}

/// <summary>
/// Why Spark can't take this node's swap now (null when it can), and how many sats the node's Spark wallet
/// can still take under its max balance, counting what its swaps not paid yet will still bring in
/// can still take under its max balance, counting what its swaps not paid yet will still bring in. Room below
/// <paramref name="minSwapSats"/> (the node's Spark min swap) can't take a swap either.
/// </summary>
private async Task<(string? Reason, long RoomSats)> SparkAvailabilityAsync(Node node)
private async Task<(string? Reason, long RoomSats)> SparkAvailabilityAsync(Node node, long minSwapSats)
{
if (_sparkSettings.WalletFor(node) is not { } sparkWallet) return ("the node has no Spark wallet", 0);

Expand All @@ -197,11 +227,10 @@ public async Task Execute(IJobExecutionContext context)
if (!status.IsReady) return (status.Reason ?? status.State.ToString(), 0);

var balance = (await _sparkWallet.GetBalanceAsync(sparkWallet)).SatsBalance;
var unpaid = SparkGuardrails.UnpaidSats((await _swapOutRepository.GetAllPending())
var room = SparkGuardrails.RoomSats(wallet.MaxBalanceSats, balance, (await _swapOutRepository.GetAllPending())
.Where(s => string.Equals(s.SparkIdentity, status.IdentityPublicKey, StringComparison.OrdinalIgnoreCase)));
var room = wallet.MaxBalanceSats - balance.Owned - balance.Incoming - unpaid;
return room < node.SwapMinAmountSats
? ($"{wallet.Name} has {Math.Max(0, room)} sats of room under its max balance, below the node's minimum swap", room)
return room < minSwapSats
? ($"{wallet.Name} has {Math.Max(0, room)} sats of room under its max balance, below the node's Spark min swap", room)
: (null, room);
}

Expand Down Expand Up @@ -241,17 +270,21 @@ public async Task<ManageNodeLiquidityResult> ManageNodeLiquidity(Node node, Canc
return ManageNodeLiquidityResult.BalanceBelowThreshold;
}

// Get in-flight swaps for this node (optimistic locking)
// Swaps in flight count per provider (optimistic locking): one may run serialized while another runs in
// parallel. A provider at its max is left out; with every weighted one at its max, nothing is swapped
var inFlightSwaps = await _swapOutRepository.GetInFlightSwapsByNode(node.Id);
var inFlightCount = inFlightSwaps.Count;
var weighted = WeightedProviders(node);
var available = weighted
.Where(w => inFlightSwaps.Count(s => s.Provider == w.Provider) < SwapProviderLimits.For(node, w.Provider, _sparkSettings).MaxInFlight)
.ToList();

_logger.LogDebug("Node {NodeName} has {InFlightCount} swaps in flight, max: {MaxSwapsInFlight}",
node.Name, inFlightCount, node.MaxSwapsInFlight);
_logger.LogDebug("Node {NodeName} has {InFlightCount} swaps in flight; providers below their max: {Providers}",
node.Name, inFlightSwaps.Count, string.Join(", ", available.Select(w => w.Provider)));

if (inFlightCount >= node.MaxSwapsInFlight)
if (available.Count == 0)
{
_logger.LogInformation("Node {NodeName} has reached max swaps in flight ({MaxSwapsInFlight})",
node.Name, node.MaxSwapsInFlight);
_logger.LogInformation("Node {NodeName} has reached max swaps in flight of every weighted provider ({Providers})",
node.Name, string.Join(", ", weighted.Select(w => w.Provider)));
return ManageNodeLiquidityResult.MaxSwapsInFlightReached;
}

Expand All @@ -276,39 +309,22 @@ public async Task<ManageNodeLiquidityResult> ManageNodeLiquidity(Node node, Canc
return ManageNodeLiquidityResult.BudgetExhausted;
}

// Determine swap amount (bounded by min, max, and excess balance)
// We can only swap what's above the threshold (excess balance)
var maxPossibleSwap = Math.Min(excessBalance, node.SwapMaxAmountSats);

// If we don't have enough to meet the minimum swap size, skip
if (maxPossibleSwap < node.SwapMinAmountSats)
{
var maxPossibleSwapBtc = new Money(maxPossibleSwap, MoneyUnit.Satoshi).ToDecimal(MoneyUnit.BTC);
var minSwapBtc = new Money(node.SwapMinAmountSats, MoneyUnit.Satoshi).ToDecimal(MoneyUnit.BTC);
_logger.LogDebug("Node {NodeName} - max possible swap {MaxPossible} BTC is below minimum {Min} BTC. Excess: {Excess} BTC",
node.Name, maxPossibleSwapBtc, minSwapBtc, excessBalanceBtc);
return ManageNodeLiquidityResult.ExcessBalanceBelowMinimum;
}

// Swap the maximum we can, ensuring it meets the minimum
var swapAmount = Math.Max(node.SwapMinAmountSats, maxPossibleSwap);

// Select swap provider based on weights, before reserving an address for a swap that may not happen
var (provider, sparkRoom) = await SelectSwapProviderAsync(node);
// Pick the provider among those whose own min and max swap fit what's above the threshold (the excess
// balance), before reserving an address for a swap that may not happen: the swap is the most it takes
var (provider, swapAmount, belowMinimum) = await SelectSwapProviderAsync(node, available, excessBalance);
if (provider is not { } selectedProvider)
{
if (belowMinimum)
{
_logger.LogDebug("Node {NodeName} - excess balance {Excess} BTC is below the min swap of every weighted provider",
node.Name, excessBalanceBtc);
return ManageNodeLiquidityResult.ExcessBalanceBelowMinimum;
}

_logger.LogInformation("Node {NodeName}: no weighted swap provider can take a swap now, skipping", node.Name);
return ManageNodeLiquidityResult.NoProviderAvailable;
}

// A Spark swap only takes what the node's Spark wallet can still hold under its max balance
if (selectedProvider == SwapProvider.Spark && swapAmount > sparkRoom)
{
_logger.LogInformation("Node {NodeName}: Spark swap clamped from {Amount} to {Room} sats by its Spark wallet's max balance",
node.Name, swapAmount, sparkRoom);
swapAmount = sparkRoom;
}

var swapAmountBtc = new Money(swapAmount, MoneyUnit.Satoshi).ToDecimal(MoneyUnit.BTC);
_logger.LogDebug("Node {NodeName} - Initiating swap for {Amount} BTC", node.Name, swapAmountBtc);

Expand Down
Loading
Loading